首页 > 其他分享> > Flume的容错性测试和双层Flume架构

Flume的容错性测试和双层Flume架构

2021-03-11 21:03:24 作者：互联网

转：

Flume的容错性测试和双层Flume架构

一.Failover Sink Processor测试

官网解释Failover Sink Processor：
Failover Sink Processor维护一个按优先级排列的sink列表，确保只要有一个sink可用，事件就会被处理(交付)。
Failover机制的工作原理是将失败的接收转移到池中，在池中为它们分配一个冷却期，在重新尝试它们之前，随着顺序故障的增加而增加。一旦接收器成功地发送了一个事件，它就会被恢复到活动池。sink有一个与它们相关联的优先级，数量越大，优先级越高。如果一个接收器在发送事件时失败，下一个具有最高优先级的接收器将被尝试下一步发送事件。例如，优先级为100的接收器在优先级为80的接收器之前被激活。如果没有指定优先级，则thr优先级根据配置中指定的sink的顺序确定。
要进行配置，将sink组处理器设置为Failover，并为所有单个的sink设置优先级。所有指定的优先级必须是唯一的。此外，可以使用maxpenalty属性设置Failover时间的上限(以毫秒为单位)。

下图中44446的优先级更高：
Flume的容错性测试和双层Flume架构
左边agent的配置failover.conf：

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

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

a1.channels.c1.type = memory

a1.sinkgroups = g1
a1.sinkgroups.g1.sinks = k1 k2
a1.sinkgroups.g1.processor.type = failover
a1.sinkgroups.g1.processor.priority.k1 = 5
a1.sinkgroups.g1.processor.priority.k2 = 10
a1.sinkgroups.g1.processor.maxpenalty = 10000

a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop000
a1.sinks.k1.port = 44445

a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop000
a1.sinks.k2.port = 44446

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

k2即agent1的44446端口的优先级高（数字越大优先级越高）。

发送数据：

[hadoop@hadoop000 apache-flume-1.6.0-cdh5.15.1-bin]$ telnet localhost 44444
Trying 192.168.198.128...
Connected to localhost.
Escape character is '^]'.
aaa
OK
bbb
OK
ccc
OK
ddd
OK
eee
OK
fff
OK

44446接收到信息：

21/01/25 18:14:47 INFO ipc.NettyServer: [id: 0x0ce2a19e, /192.168.198.128:45240 => /192.168.198.128:44446] OPEN
21/01/25 18:14:47 INFO ipc.NettyServer: [id: 0x0ce2a19e, /192.168.198.128:45240 => /192.168.198.128:44446] BOUND: /192.168.198.128:44446
21/01/25 18:14:47 INFO ipc.NettyServer: [id: 0x0ce2a19e, /192.168.198.128:45240 => /192.168.198.128:44446] CONNECTED: /192.168.198.128:45240
21/01/25 18:15:40 INFO sink.LoggerSink: Event: { headers:{} body: 61 61 61 0D                                     aaa. }
21/01/25 18:16:11 INFO sink.LoggerSink: Event: { headers:{} body: 62 62 62 0D                                     bbb. }

将agent3 kill掉，44445端口被激活：

21/01/25 18:14:46 INFO ipc.NettyServer: [id: 0x946f8c34, /192.168.198.128:55142 => /192.168.198.128:44445] OPEN
21/01/25 18:14:46 INFO ipc.NettyServer: [id: 0x946f8c34, /192.168.198.128:55142 => /192.168.198.128:44445] BOUND: /192.168.198.128:44445
21/01/25 18:14:46 INFO ipc.NettyServer: [id: 0x946f8c34, /192.168.198.128:55142 => /192.168.198.128:44445] CONNECTED: /192.168.198.128:55142
21/01/25 18:16:42 INFO sink.LoggerSink: Event: { headers:{} body: 63 63 63 0D                                     ccc. }
21/01/25 18:16:48 INFO sink.LoggerSink: Event: { headers:{} body: 64 64 64 0D                                     ddd. }
21/01/25 18:47:19 INFO sink.LoggerSink: Event: { headers:{} body: 65 65 65 0D                                     eee. }

重启agent3，44446端口再次被激活：

21/01/25 18:50:10 INFO ipc.NettyServer: [id: 0x58750737, /192.168.198.128:45596 => /192.168.198.128:44446] OPEN
21/01/25 18:50:10 INFO ipc.NettyServer: [id: 0x58750737, /192.168.198.128:45596 => /192.168.198.128:44446] BOUND: /192.168.198.128:44446
21/01/25 18:50:10 INFO ipc.NettyServer: [id: 0x58750737, /192.168.198.128:45596 => /192.168.198.128:44446] CONNECTED: /192.168.198.128:45596
21/01/25 18:50:13 INFO sink.LoggerSink: Event: { headers:{} body: 66 66 66 0D                                     fff. }

二.双层的Flume架构

这篇博客写的特别详细：Flume日志收集分层架构应用实践.

双层Flume的好处：

解耦，hdfs或者kafka需要升级时，第二层flume可以进行缓冲，不会影响第一层。
安全，hdfs或者kafka直接暴露给第一层不安全（第一层很多flume来自其他部门，第二层在本地）。
利于业务的分组管理，将第一组的繁杂业务在第二层可以进行分组。
小文件的数量会大大减少。
外部某个类型的业务日志数据节点需要扩容，直接在L1层将数据流指向数据平台内部与之相对应的L2层Flume Agent节点组即可。

三.单source多channel多sink

Flume的容错性测试和双层Flume架构第一层source发送一个消息，channel1和channel2都会传输，agent2和agent3都会收到相同的数据，所以这种架构可以将同一份数据，即可以导入hdfs进行离线计算，也可同时导入实时框架进行实时计算，实现多用途。

转：