1. Flume 进阶

2. Flume 事务

05. Flume 进阶 - 图1

3. Agent 内部原理

05. Flume 进阶 - 图2

  1. ChannelSelector
    ChannelSelector的作用就是选出Event将要被发往哪个Channel。其共有两种类型，分别是Replicating（复制）和Multiplexing（多路复用）。 默认为Replicating（复制）
    ReplicatingSelector会将同一个Event发往所有的Channel，Multiplexing会根据相应的原则，将不同的Event发往不同的Channel。

  2. SinkProcessor
    SinkProcessor共有三种类型，分别是DefaultSinkProcessor、LoadBalancingSinkProcessor和FailoverSinkProcessor

    1. DefaultSinkProcessor对应的是单个的Sink
    2. LoadBalancingSinkProcessor和FailoverSinkProcessor对应的是Sink Group，
    3. LoadBalancingSinkProcessor可以实现负载均衡的功能
    4. FailoverSinkProcessor可以错误恢复的功能。

4. Flume 拓扑结构

4.1. 简单串联

05. Flume 进阶 - 图3

这种模式是将多个flume顺序连接起来了，从最初的source开始到最终sink传送的目的存储系统。此模式不建议桥接过多的flume数量， flume数量过多不仅会影响传输速率，而且一旦传输过程中某个节点flume宕机，会影响整个传输系统。

4.1.1. 实现串联 输出到arvo

flume1 配置文件

  1. #agent1   netcatsource --> memorychannel --> arvosink
  2. a1.sources = r1  
  3. a1.sinks = k1
  4. a1.channels = c1 
  6. a1.sources.r1.type = netcat 
  7. a1.sources.r1.bind = hadoop102
  8. a1.sources.r1.port = 22222 
  10. #设置为arvosink 向指定地址:端口输出数据
  11. a1.sinks.k1.type = arvo 
  12. #输出数据的地址 
  13. a1.sinks.k1.hostname = hadoop103
  14. #输出数据的地址
  15. a1.sinks.k1.port = 33333
  17. a1.channels.c1.type = memory 
  18. a1.channels.c1.capacity = 1000 
  19. a1.channels.c1.transactionCapacity = 100 
  22. a1.sources.r1.channels = c1 
  23. a1.sinks.k1.channel = c1

flume2 配置文件

  1. #agent2   netcatsource --> memorychannel --> loggersink
  2. a1.sources = r1
  3. a1.sinks = k1
  4. a1.channels = c1
  6. #输入数据类型改为 arvo
  7. a1.sources.r1.type = arvo 
  8. #输入地址
  9. a1.sources.r1.bind = hadoop102
  10. #输入端口
  11. a1.sources.r1.port = 33333
  13. #设置为logger 写入到log文件 持久化
  14. a1.sinks.k1.type = logger
  16. a1.channels.c1.type = memory 
  17. a1.channels.c1.capacity = 1000 
  18. a1.channels.c1.transactionCapacity = 100 
  21. a1.sources.r1.channels = c1 
  22. a1.sinks.k1.channel = c1

测试

  1. #先启动103 否则102发送数据无人接收  在hadoop103操作 
  2. flume-ng agent --conf conf/ --name a1 --conf-file datas/avrosource_loggersink.conf -Dflume.root.logger=INFO,console
  1. #在hadoop102操作
  2. flume-ng agent --conf conf/ --name a1 --conf-file datas/netcatsource_avrosink.conf -Dflume.root.logger=INFO,console
  3. #在另外个ssh窗口中操作
  4. nc hadoop102 22222

4.2. 复制和多路复用

05. Flume 进阶 - 图4

Flume支持将事件流向一个或者多个目的地。这种模式可以将相同数据复制到多个channel中，或者将不同数据分发到不同的channel中，sink可以选择传送到不同的目的地。

4.2.1. 实现复制 selector=replicating

05. Flume 进阶 - 图5

从指定文件中读取日志 复制转发到个channel中 channel再转发给指定的sink方

  1. #agent 1  hadoop102
  2. a1.sources = r1
  4. a1.sinks = k1 k2
  5. a1.channels = c1 c2
  7. a1.sources.r1.type = exec 
  8. #读取hive日志文件
  9. a1.sources.r1.command = tail -F /opt/module/hive/logs/hive.log
  10. #selector频道选择器 默认为replicating 为复制 不配置type也是这个方案
  11. a1.sources.r1.selector.type = replicating 
  12. #可选的channel
  13. #a1.sources.r1.selector.optional = c3
  15. #设置为arvosink 向指定地址:端口输出数据
  16. a1.sinks.k1.type = arvo 
  17. #输出数据的地址 
  18. a1.sinks.k1.hostname = hadoop103
  19. #输出数据的地址
  20. a1.sinks.k1.port = 33333
  22. #第二个sinks
  23. a1.sinks.k2.type = arvo 
  24. a1.sinks.k2.hostname = hadoop104
  25. a1.sinks.k2.port = 44444
  27. a1.channels.c1.type = memory 
  28. a1.channels.c1.capacity = 1000 
  29. a1.channels.c1.transactionCapacity = 100 
  30. #第二个channel
  31. a1.channels.c2.type = memory 
  32. a1.channels.c2.capacity = 1000 
  33. a1.channels.c2.transactionCapacity = 100 
  35. #一个sources 对接两个channels
  36. a1.sources.r1.channels = c1 c2
  37. #每个sinks对应一个channel
  38. a1.sinks.k1.channel = c1
  39. a1.sinks.k2.channel = c2

sink方1 从hadoop102接收数据 再存储到hdfs中

  1. #agent2 hadoop103
  2. a1.sources = r1  
  3. a1.sinks = k1
  4. a1.channels = c1 
  6. a1.sources.r1.type = avro
  7. a1.sources.r1.bind = hadoop103
  8. a1.sources.r1.port = 33333
  10. # Describe the sink
  11. a1.sinks.k1.type = hdfs
  12. a1.sinks.k1.hdfs.path = hdfs://hadoop102:8020/flume2/%Y%m%d/%H
  13. #上传文件的前缀
  14. a1.sinks.k1.hdfs.filePrefix = flume2-
  15. #是否按照时间滚动文件夹
  16. a1.sinks.k1.hdfs.round = true
  17. #多少时间单位创建一个新的文件夹
  18. a1.sinks.k1.hdfs.roundValue = 1
  19. #重新定义时间单位
  20. a1.sinks.k1.hdfs.roundUnit = hour
  21. #是否使用本地时间戳
  22. a1.sinks.k1.hdfs.useLocalTimeStamp = true
  23. #积攒多少个Event才flush到HDFS一次
  24. a1.sinks.k1.hdfs.batchSize = 100
  25. #设置文件类型，可支持压缩
  26. a1.sinks.k1.hdfs.fileType = DataStream
  27. #多久生成一个新的文件
  28. a1.sinks.k1.hdfs.rollInterval = 600
  29. #设置每个文件的滚动大小大概是128M
  30. a1.sinks.k1.hdfs.rollSize = 134217700
  31. #文件的滚动与Event数量无关
  32. a1.sinks.k1.hdfs.rollCount = 0
  34. a1.channels.c1.type = memory 
  35. a1.channels.c1.capacity = 1000 
  36. a1.channels.c1.transactionCapacity = 100 
  39. a1.sources.r1.channels = c1 
  40. a1.sinks.k1.channel = c1

sink2 从hadoop102接收数据 再通过File_roll sink存储到本地目录中

  1. #agent3 hadoop104
  2. a1.sources = r1  
  3. a1.sinks = k1
  4. a1.channels = c1 
  6. a1.sources.r1.type = avro
  7. a1.sources.r1.bind = hadoop104
  8. a1.sources.r1.port = 44444
  10. #将event数据存储在指定sink为file_roll 本地存储模式
  11. a1.sinks.k1.type = file_roll
  12. #存放目录 输出的本地目录必须是已经存在的目录
  13. a1.sinks.k1.sink.directory = /opt/module/flume/demo
  14. #默认为30s 滚动文件 设置为0将不再滚动
  15. a1.sinks.k1.sink.rollInterval = 30
  17. a1.channels.c1.type = memory 
  18. a1.channels.c1.capacity = 1000 
  19. a1.channels.c1.transactionCapacity = 100 
  22. a1.sources.r1.channels = c1 
  23. a1.sinks.k1.channel = c1

启动 先启动103和104的监听 再启动102的监听

4.2.2. 实现多路复用 selector=multiplexing

agent1

  1. #agent 1  hadoop102
  2. a1.sources = r1
  4. a1.sinks = k1 k2
  5. a1.channels = c1 c2
  7. a1.sources.r1.type = exec 
  8. #读取hive日志文件
  9. a1.sources.r1.command = tail -F /opt/module/hive/logs/hive.log
  12. #复用配置
  13. #selector频道选择器 默认为replicating  multiplexing为复用  需要配合拦截器使用
  14. a1.sources.r1.selector.type = multiplexing 
  15. # header的key 根据event的header里面指定key 判断值 分发给哪个channel 
  16. a1.sources.r1.selector.header = state
  17. #CZ为自定义value值 为上面指定key中对应值 如key中值为CZ 则分发给 c1 channel
  18. a1.sources.r1.selector.mapping.CZ = c1
  19. #值为US 则分发到 c2 channel
  20. a1.sources.r1.selector.mapping.US = c2
  23. # 设置拦截器 (用于向headers中添加指定键值对)
  24. #拦截器名称
  25. a1.sources.r1.interceptors = i1
  26. #拦截器类型 static 向header添加 自定义键值对
  27. a1.sources.r1.interceptors.i1.type = static
  28. a1.sources.r1.interceptors.i1.key = state
  29. #多个值只能通过自定义拦截器定义 此处是写死为CZ
  30. a1.sources.r1.interceptors.i1.value = CZ
  33. #设置为arvosink 向指定地址:端口输出数据
  34. a1.sinks.k1.type = arvo 
  35. #输出数据的地址 
  36. a1.sinks.k1.hostname = hadoop103
  37. #输出数据的地址
  38. a1.sinks.k1.port = 33333
  40. #第二个sinks
  41. a1.sinks.k2.type = arvo 
  42. a1.sinks.k2.hostname = hadoop104
  43. a1.sinks.k2.port = 44444
  45. a1.channels.c1.type = memory 
  46. a1.channels.c1.capacity = 1000 
  47. a1.channels.c1.transactionCapacity = 100 
  48. #第二个channel
  49. a1.channels.c2.type = memory 
  50. a1.channels.c2.capacity = 1000 
  51. a1.channels.c2.transactionCapacity = 100 
  53. #一个sources 对接两个channels
  54. a1.sources.r1.channels = c1 c2
  55. #每个sinks对应一个channel
  56. a1.sinks.k1.channel = c1
  57. a1.sinks.k2.channel = c2

agent2和agent3 与上面复制一样 或 自定义

4.3. 负载均衡和故障转移

05. Flume 进阶 - 图6

Flume支持使用将多个sink逻辑上分到一个sink组，sink组配合不同的SinkProcessor可以实现负载均衡和错误恢复的功能。

4.3.1. 实现故障转移 processor=failover

05. Flume 进阶 - 图7

  1. #agent1 hadoop102
  2. a1.sources = r1  
  3. a1.sinks = k1 k2
  4. a1.channels = c1 
  6. a1.sources.r1.type = netcat 
  7. a1.sources.r1.bind = hadoop102
  8. a1.sources.r1.port = 22222
  10. #sinks1
  11. a1.sinks.k1.type = avro
  12. a1.sinks.k1.hostname = hadoop103
  13. a1.sinks.k1.port = 33333
  14. #sinks2
  15. a1.sinks.k2.type = avro
  16. a1.sinks.k2.hostname = hadoop104
  17. a1.sinks.k2.port = 44444
  19. #定义sinkgroups
  20. a1.sinkgroups = g1
  21. #该组下面有哪些sink实例
  22. a1.sinkgroups.g1.sinks = k1 k2
  23. #failover为故障转移  默认为一对一
  24. a1.sinkgroups.g1.processor.type = failover
  25. #优先级 值越大优先级越大
  26. a1.sinkgroups.g1.processor.priority.k1 = 5
  27. a1.sinkgroups.g1.processor.priority.k2 = 10
  28. #sink连接超时时间 默认为30000毫秒
  29. a1.sinkgroups.g1.processor.maxpenalty = 10000
  32. a1.channels.c1.type = memory 
  33. a1.channels.c1.capacity = 1000 
  34. a1.channels.c1.transactionCapacity = 100 
  37. a1.sources.r1.channels = c1 
  38. #sinks绑定的channel 应为一个
  39. a1.sinks.k1.channel = c1
  40. a1.sinks.k2.channel = c1

4.3.2. 实现负载均衡 processor=load_balance

#agent1 hadoop102
a1.sources = r1  
a1.sinks = k1 k2
a1.channels = c1 

a1.sources.r1.type = netcat 
a1.sources.r1.bind = hadoop102
a1.sources.r1.port = 22222

#sinks1
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop103
a1.sinks.k1.port = 33333
#sinks2
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop104
a1.sinks.k2.port = 44444

#定义sinkgroups
a1.sinkgroups = g1
#该组下面有哪些sink实例
a1.sinkgroups.g1.sinks = k1 k2
#load_balance 为负载均衡
a1.sinkgroups.g1.processor.type = load_balance
# 默认为round_robin轮询sink    random为随机发给某个sink
a1.sinkgroups.g1.processor.selector = random
#连接超时时间 30000毫秒
a1.sinkgroups.g1.processor.maxpenalty = 10000


a1.channels.c1.type = memory 
a1.channels.c1.capacity = 1000 
a1.channels.c1.transactionCapacity = 100 


a1.sources.r1.channels = c1 
#sinks绑定的channel 应为一个
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1

4.4. 聚合

05. Flume 进阶 - 图8

这种模式是我们最常见的，也非常实用，日常web应用通常分布在上百个服务器，大者甚至上千个、上万个服务器。产生的日志，处理起来也非常麻烦。用flume的这种组合方式能很好的解决这一问题，每台服务器部署一个flume采集日志，传送到一个集中收集日志的flume，再由此flume上传到hdfs、hive、hbase等，进行日志分析。

4.4.1. 实现聚合

05. Flume 进阶 - 图9

将分开的agent的sink全部汇总到一个agent上 再进行持久化

5. 拦截器 Interceptor

更多类型拦截器查看官方文档

通过配置文件 配置拦截器 agent名称.sources.r1.interceptors

a1.sources = r1
a1.sinks = k1
a1.channels = c1 

a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# 设置拦截器 (用于向headers中添加时间戳)
#拦截器名称
a1.sources.r1.interceptors = i1
#拦截器类型 timestamp 向header添加时间戳
a1.sources.r1.interceptors.i1.type = timestamp

a1.sinks.k1.type = logger

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100


a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

05. Flume 进阶 - 图10