flume拓扑结构之故障转移和负载均衡

关于故障转移：Sink groups允许组织多个sink到一个实体上。 Sink processors能够提供在组内所有Sink之间实现负载均衡的能力，而且在失败的情况下能够进行故障转移从一个Sink到另一个Sink。

我们去官网看一下Failover Sink Processor（故障转移）,FailoverSink Processor会通过配置维护了一个优先级列表。保证每一个有效的事件都会被处理。在这配置中，要设置sinkgroups processor为failover，需要为所有的sink分配优先级，所有的优先级数字必须是唯一的，这个得格外注意。此外，failover time的上限可以通过maxpenalty 属性来进行设置。

故障转移的工作原理是将连续失败的sink分配到一个失败的消息队列中这个sink不会做任何事。假如在一个Agent中，一个Channen对应三个Sink,Sink1和Sink2,Sink3,优先级顺序为Sink1>Sink2>Sink3，所以Channel过来的数据总是先给Sink3，如果Sink3挂掉以后的话在的最大回退周期(也即是Failover Sink Processor中的processor.maxpenalty属性值)内自动起来的话Channel传来的数据是不会再考虑Sink1的，他会传给下一个优先级比较高的Sink2，如果Sink1恢复的时间大于processor.maxpenalty这个值的话那么在下一次传输数据给Sink的时候还是会优先的考虑给Sink1.

故障转移的案例需求： 使用Flume1监控一个端口，其sink组中的sink分别对接Flume2和Flume3，采用FailoverSinkProcessor，实现故障转移的功能。

在主节点的hadoop102上的在/opt/module/flume/job目录下创建group2目录

1.在group2目录中创建flume-netcat-flume.conf文件 配置1个netcat source和1个channel、1个sink group（2个sink），分别输送给flume-flume-console1和flume-flume-console2

在新创建的flume-netcat-flume.conf文件中添加如下内容

# Name the components on this agent a1.sources = r1 a1.channels = c1 a1.sinkgroups = g1 a1.sinks = k1 k2 # Describe/configure the source a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 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 # Describe the sink a1.sinks.k1.type = avro a1.sinks.k1.hostname = hadoop102 a1.sinks.k1.port = 4141 a1.sinks.k2.type = avro a1.sinks.k2.hostname = hadoop102 a1.sinks.k2.port = 4142 # Describe the channel a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinkgroups.g1.sinks = k1 k2 a1.sinks.k1.channel = c1 a1.sinks.k2.channel = c1

2.在group2目录中创建flume-flume-console1.conf在文件中填入如下内容 目的是配置上级Flume输出的Source，输出是到本地控制台

# Name the components on this agent a2.sources = r1 a2.sinks = k1 a2.channels = c1 # Describe/configure the source a2.sources.r1.type = avro a2.sources.r1.bind = hadoop102 a2.sources.r1.port = 4141 # Describe the sink a2.sinks.k1.type = logger # Describe the channel a2.channels.c1.type = memory a2.channels.c1.capacity = 1000 a2.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a2.sources.r1.channels = c1 a2.sinks.k1.channel = c1

3.创建flume-flume-console2.conf添加如下内容作用是配置上级Flume输出的Source，输出是到本地控制台，只不过他的优先级高于上面一个

# Name the components on this agent a3.sources = r1 a3.sinks = k1 a3.channels = c2 # Describe/configure the source a3.sources.r1.type = avro a3.sources.r1.bind = hadoop102 a3.sources.r1.port = 4142 # Describe the sink a3.sinks.k1.type = logger # Describe the channel a3.channels.c2.type = memory a3.channels.c2.capacity = 1000 a3.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a3.sources.r1.channels = c2 a3.sinks.k1.channel = c2

我们可以复制一下hadoop102，因为每次启动一个flume的话都会处于阻塞状态，然后分别开启对应配置文件：flume-flume-console2，flume-flume-console1，flume-netcat-flume

分别在4hadoop102,3hadoop102,2hadoop102的/opt/module/flume目录下执行

bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group2/flume-flume-console2.conf -Dflume.root.logger=INFO,console bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group2/flume-flume-console1.conf -Dflume.root.logger=INFO,console bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group2/flume-netcat-flume.conf

在2hadoop102上面执行完bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group2/flume-netcat-flume.conf后发现3hadoop102和4hadoop102的机器上会在之前的启动上的日志后面显示绑定和连接上了Avro Sink

这时候我们再去最初的那个1hadoop102机器上输入nc localhost 44444 回车以后随便输入内容

由于我们当时规定的是k2的优先级高，k2对应的是flume-flume-console2.conf这个配置文件，而这个配置文件我们是在4hadoop102开着的,所以我们去4hadoop102上面看一下也确实收到了消息。

因为k1的优先级低，所以3hadoop102上面的flume-flume-console1.conf 自然就收不到消息，这时我们去kill掉4hadoop102上面的flume进程发现只有3hadoop102的机器能接收到消息，这也验证了我们的故障转移的配置是可行的。

负载均衡的案例： 讲完了故障转移负载均衡进很简单了在原来的基础上稍微修改一下就可以了

我们在hadoop102机器上的/opt/module/flume/job目录下新建目录group3 然后cp -r group2/ group3

然后我们只需要在group3修改一下flume-netcat-flume.conf就可以了

我们先去官网找一下关于Load balancing Sink Processor的内容

将红线框的4行替换为如下内容然后保存退出。

a1.sinkgroups.g1.processor.type = load_balance a1.sinkgroups.g1.processor.backoff = true a1.sinkgroups.g1.processor.selector = random

然后我们在将上面的步骤启动一遍flume(注意文件的所在路径，之前是在group2里面，现在是在group3里面

这时我们再去最初的hadoop102里面输入nc localhost 44444回车后输入任意内容。

这时我们发现接收的消息是随机的，也即是实现了我们的负载平衡。