Flume之使用Failover Sink Processor实现sink故障转移

本文主要是介绍Flume之使用Failover Sink Processor实现sink故障转移，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

前言

Failover Sink Processor 维护着Sink组中Sinks的优先级表，根据优先级尝试将Event传输给不同的Sink直到Event成功发送。当优先级高的Sink不可用时，会将Event传输给下一优先级Sink，以此来确保每个Event都能被投递。当Sink不可用时，Failover Sink Processor和Load balancing Sink Processor一样，也会进行指数回退backoff，并可以设置最大回退时间（即在黑名单中的保存时间），在倒计时结束后会再次尝试访问之前挂掉的Sink

使用示例

1）flume1.properties

# flume1:此配置用于监控某个窗口将其追加内容输出到flume2和flume3中
# 并将两个Sink组成一个sink group,并将Sink Processor设置成Failover类型
# a1:Netcat Source->Memory Channel->Avro Sink# Agent
a1.sources = r1
a1.channels = c1
a1.sinks = k1 k2# Sink groups
a1.sinkgroups = g1
# 设置sink group中的sinks
a1.sinkgroups.g1.sinks = k1 k2
# 设置Failover sink processor(只有sink group才可以使用sink processor)
a1.sinkgroups.g1.processor.type = failover
# 设置Failover sink processor优先级表
a1.sinkgroups.g1.processor.priority.k1 = 5
a1.sinkgroups.g1.processor.priority.k2 = 10
# 设置最大避让时间(ms)
a1.sinkgroups.g1.processor.maxpenalty = 10000# Sources
# 配置a1.sources.r1的各项属性参数,类型/绑定主机ip/端口号
a1.sources.r1.type = netcat
a1.sources.r1.bind = hadoop101
a1.sources.r1.port = 44444# Channels
# 配置a1.channerls.c1的各项属性参数,缓存方式/最多缓存的Event个数/单次传输的Event个数
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100# Sinks
# sinks.k1
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141
# sinks.k2
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop103
a1.sinks.k2.port = 4141# Bind
# 注意:source可以绑定多个channel,但是sink/sink group只能绑定单个channel
# r1->c1->g1
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1

2）flume2.properties

# flume2:此配置用于将来自指定Avro端口的数据输出到控制台
# a2:Avro Source->Memory Channel->Logger Sink# Agent
a2.sources = r1
a2.channels = c1
a2.sinks = k1# Sources
# a2.sources.r1
a2.sources.r1.type = avro
# 设置监听本地IP
a2.sources.r1.bind = 0.0.0.0
# 设置监听端口号
a2.sources.r1.port = 4141# Channels
# a2.channels.c1
# 使用内存作为缓存/最多缓存的Event个数/单次传输的Event个数
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100# Sinks
# 运行时设置参数 -Dflume.root.logger=INFO,console 即输出到控制台实时显示
a2.sinks.k1.type = logger
# 设置Event的Body中写入log的最大字节数(默认值为16)
a2.sinks.k1.maxBytesToLog = 256# Bind
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

3）flume3.properties

# flume3:此配置用于将来自指定Avro端口的数据输出到控制台
# a3:Avro Source->Memory Channel->Logger Sink# Agent
a3.sources = r1
a3.channels = c1
a3.sinks = k1# Sources
# a3.sources.r1
a3.sources.r1.type = avro
# 设置监听本地IP
a3.sources.r1.bind = 0.0.0.0
# 设置监听端口号
a3.sources.r1.port = 4141# Channels
# a3.channels.c1
# 使用内存作为缓存/最多缓存的Event个数/单次传输的Event个数
a3.channels.c1.type = memory
a3.channels.c1.capacity = 1000
a3.channels.c1.transactionCapacity = 100# Sinks
# 运行时设置参数 -Dflume.root.logger=INFO,console 即输出到控制台实时显示
a3.sinks.k1.type = logger
# 设置Event的Body中写入log的最大字节数(默认值为16)
a3.sinks.k1.maxBytesToLog = 256# Bind
a3.sources.r1.channels = c1
a3.sinks.k1.channel = c1

4）启动命令

Flume Agent a1至a3分别运行在主机hadoop101、hadoop102、hadoop103上

./bin/flume-ng agent -n a1 -c conf -f flume1.properties
./bin/flume-ng agent -n a2 -c conf -f flume2.properties -Dflume.root.logger=INFO,console
./bin/flume-ng agent -n a3 -c conf -f flume3.properties -Dflume.root.logger=INFO,console

5）实现功能

Aent a1将指定端口的监听数据输出到a2或者a3的控制台
当Event从Channel中传输给Sink Group之前，首先会根据配置Failover sink processor优先级表尝试将此Event发送给优先级最高的可用Sink，如果成功则继续处理下一个Event。如果在发送过程中，当前Sink宕机，则将其加入黑名单，一定时间内不再尝试将Event发往此Sink，并且退避时间呈指数增长，直到最大退避时间maxpenalty，以此来实现Sink的故障转移