Flume案例七：故障转移(Failover Sink Processor)

落日映苍穹つ 2022-10-10 01:35 182阅读 0赞

> 本文接上篇博客：[Flume介绍、安装、使用案例、自定义Source/Sink、监控][Flume_Source_Sink]  
> Flume 版本：1.9.0

## 1.Flume故障转移 ##

> 需求：  
>   使用 Flume-1 监控文件变动，Flume-1 将变动内容传递给 Flume-2，Flume-2 负责打印到控制台。如果 Flume-2 挂掉，Flume-1 将变动内容传递给 Flume-3，Flume-3 负责打印到控制台。  
>    
> 选型：  
>  Flume-1：`taildir source` \+ `memory channel` \+ `avro sink` \+ `Failover Sink Processor(故障转移)`  
>  Flume-2：`avro source` \+ `memory channel` \+ `logger sink`  
>  Flume-3：`avro source` \+ `memory channel` \+ `logger sink`  
>    
> 文档参考：  
>  **taildir source**：[http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html\#taildir-source][http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_taildir-source]  
>  **memory channel**：[http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html\#memory-channel][http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_memory-channel]  
>  **avro sink**：[http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html\#avro-sink][http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_avro-sink]  
>  **logger sink**：[http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html\#logger-sink][http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_logger-sink]  
>  **Failover Sink Processor**：[http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html\#failover-sink-processor][http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_failover-sink-processor]

## 2.需求分析 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6YjM0ODExMDE3NQ_size_16_color_FFFFFF_t_70]

## 3.flume配置 ##

### Ⅰ.Flume-1 ###

flume-netcat-avro-failover.conf

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1 k2
    a1.channels = c1
    a1.sinkgroups = g1
    
    # Failover configure
    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/configure the source
    a1.sources.r1.type = TAILDIR
    a1.sources.r1.positionFile = /opt/module/flume/position/taildir_failover_position.json
    a1.sources.r1.filegroups = f1
    a1.sources.r1.filegroups.f1 = /opt/module/testdir/test.log
    
    # Describe the sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = 192.168.204.202
    a1.sinks.k1.port = 41414
    
    a1.sinks.k2.type = avro
    a1.sinks.k2.hostname = 192.168.204.203
    a1.sinks.k2.port = 41414
    
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # Bind the source and sink to the channel
    a1.sinkgroups.g1.sinks = k1 k2
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c1

### Ⅱ.Flume-2 ###

flume-avro-logger.conf

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = avro
    a1.sources.r1.bind = 192.168.204.202
    a1.sources.r1.port = 41414
    
    # Describe the sink
    a1.sinks.k1.type = logger
    
    # Use a channel which buffers events in memory
    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.sinks.k1.channel = c1

### Ⅲ.Flume-3 ###

flume-avro-logger.conf

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = avro
    a1.sources.r1.bind = 192.168.204.203
    a1.sources.r1.port = 41414
    
    # Describe the sink
    a1.sinks.k1.type = logger
    
    # Use a channel which buffers events in memory
    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.sinks.k1.channel = c1

## 4.启动命令 ##

> 注意：  
>   必须先启动 Flume-2 和 Flume-3，再启动 Flume-1。如果先启动 Flume-1，会报错误:Connection refused: /192.168.204.202:41414 和 Connection refused: /192.168.204.203:41414  
>    
> 提示：  
>   Flume-2 sink优先级为5，Flume-3 sink优先级为10，所以 Flume-3 优先接收 sink 数据。此时手动将 Flume-3 关停，Flume-1 后台日志会报`Caused by: java.io.IOException: Error connecting to /192.168.204.203:41414`。然后 Flume-2 开始 sink数据。这就是故障转移

# Flume-2 启动命令
    bin/flume-ng agent -c conf -n a1 -f job/failover/flume-avro-logger.conf -Dflume.root.logger=INFO,console
    # Flume-3 启动命令
    bin/flume-ng agent -c conf -n a1 -f job/failover/flume-avro-logger.conf -Dflume.root.logger=INFO,console
    # Flume-1 启动命令
    bin/flume-ng agent -c conf -n a1 -f job/failover/flume-taildir-avro-failover.conf

## 5.测试图示 ##

> Taildir Source 实时监听 `testdir/test.log`
> 
> 1.  echo 方式追加数据至 `test.log` 文件，模拟实时日志；
> 2.  优先级问题，发现 Flume-3 会接收到数据，logger sink 输出至控制台
> 3.  人工手动关停 Flume-3 服务，此时 Flume-1 会报错连接不上 Flume-3，转而通过 Flume-2 开始sink

**测试结果，如图所示：**

> 提示：下图由于过大，CSDN最大支持5M，所以对图片有做处理进行取舍，有点快。原图在这里：[https://pan.baidu.com/s/1UcS7cGrQtlIRleOtRMnO2Q][https_pan.baidu.com_s_1UcS7cGrQtlIRleOtRMnO2Q] 提取码：cmwk

![在这里插入图片描述][20210625161454983.gif]

--------------------

**博主写作不易，加个关注呗**

**求关注、求点赞，加个关注不迷路 ヾ(◍°∇°◍)ﾉﾞ**

**我不能保证所写的内容都正确，但是可以保证不复制、不粘贴。保证每一句话、每一行代码都是亲手敲过的，错误也请指出，望轻喷 Thanks♪(･ω･)ﾉ**

[Flume_Source_Sink]: https://blog.csdn.net/lzb348110175/article/details/118154430
[http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_taildir-source]: http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#taildir-source
[http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_memory-channel]: http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#memory-channel
[http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_avro-sink]: http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#avro-sink
[http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_logger-sink]: http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#logger-sink
[http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_failover-sink-processor]: http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#failover-sink-processor
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6YjM0ODExMDE3NQ_size_16_color_FFFFFF_t_70]: /images/20221005/ccd3f861f06540f28967650a6d4fff2e.png
[https_pan.baidu.com_s_1UcS7cGrQtlIRleOtRMnO2Q]: https://pan.baidu.com/s/1UcS7cGrQtlIRleOtRMnO2Q
[20210625161454983.gif]: /images/20221005/fe9a2292d1534bffb2b9986ccf33f833.png