Flume案例九：多数据源聚合

不念不忘少年蓝@ 2022-10-10 12:39 26阅读 0赞

> 本文接上篇博客：[Flume介绍、安装、使用案例、自定义Source/Sink、监控][Flume_Source_Sink]  
> Flume 版本：1.9.0  
> 本文hdfs sink，需 Hadoop 支持，Hadoop相关内容，请参考：[Hadoop专栏][Hadoop]

## 1.多数据源聚合（多数据源聚合：2进1出） ##

> 案例：  
>   Flume-1 监听 hive.log 日志文件；  
>   Flume-2 监听 test.log 日志文件；  
>   Flume-3 聚合汇总 Flume-1 和 Flume-2 的日志，并将日志 sink 到 hdfs   
>    
> 选型：  
>  Flume-1：`taildir source` \+ `memory channel` \+ `avro sink`  
>  Flume-2：`taildir source` \+ `memory channel` \+ `avro sink`  
>  Flume-3：`avro source` \+ `memory channel` \+ `hdfs 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]  
>  **hdfs sink**：[http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html\#hdfs-sink][http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html_hdfs-sink]

## 2.需求分析 ##

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

## 3.flume配置 ##

### Ⅰ.Flume-1 ###

flume-taildir-avro.conf

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = TAILDIR
    a1.sources.r1.positionFile = /opt/module/flume/position/taildir_aggregate_position.json
    a1.sources.r1.filegroups = f1
    a1.sources.r1.filegroups.f1 = /opt/module/testdir/hive.log
    
    # Describe the sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = 192.168.204.203
    a1.sinks.k1.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.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

### Ⅱ.Flume-2 ###

flume-taildir-avro.conf

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = TAILDIR
    a1.sources.r1.positionFile = /opt/module/flume/position/taildir_aggregate_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.203
    a1.sinks.k1.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.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

### Ⅲ.Flume-3 ###

flume-avro-hdfs.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 = hdfs
    a1.sinks.k1.hdfs.path = /flume/aggregate/%Y-%m-%d/%H
    a1.sinks.k1.hdfs.filePrefix = aggregate
    # 是否使用本地时间戳
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    # 是否按照时间滚动文件夹
    a1.sinks.k1.hdfs.round = true
    # 多少时间单位创建一个新的文件夹
    a1.sinks.k1.hdfs.roundValue = 1
    # 重新定义时间单位
    a1.sinks.k1.hdfs.roundUnit = hour
    # 积攒多少个 Event 才 flush 到 HDFS 一次
    a2.sinks.k2.hdfs.batchSize = 1000
    # 多久生成一个新的文件(seconds)
    a1.sinks.k1.hdfs.rollInterval = 300
    # 设置每个文件的滚动大小
    a1.sinks.k1.hdfs.rollSize = 134217700
    # 文件的滚动与 Event 数量无关
    a1.sinks.k1.hdfs.rollCount = 0
    # 设置文件类型，可支持压缩(不加该配置的话，Flume写入HDFS的文件会出现SEQ !org.apache.hadoop.io.LongWritable"org.apache.hadoop.io.BytesWritable)
    a1.sinks.k1.hdfs.fileType = DataStream
    
    # 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-3，再启动 Flume-1 和 Flume-2。

# Flume-3 启动命令
    bin/flume-ng agent -c conf -n a1 -f job/aggregate/flume-avro-hdfs.conf
    # Flume-1 启动命令
    bin/flume-ng agent -c conf -n a1 -f job/aggregate/flume-taildir-avro.conf
    # Flume-2 启动命令
    bin/flume-ng agent -c conf -n a1 -f job/aggregate/flume-taildir-avro.conf

## 5.异常处理 ##

写入HDFS，报如下错误：`java.lang.NoSuchMethodError:com.google.common.base.Preconditions.checkArgument(ZLjava/lang/String;Ljava/lang/Object;)V`

这是因为hadoop目录下得guava版本和flume下的guava版本的问题。进入 flume/lib 目录下，将 guava-11.0.2.jar 包移除即可

## 6.测试图示 ##

> Taildir Source 实时监听 `testdir/test.log` 和 `testdir/hive.log`
> 
> 1.  echo 方式追加数据至 `hive.log` 文件，模拟实时日志；会将数据写入hdfs 的 `/flume/aggregate` 目录下文件中
> 2.  echo 方式追加数据至 `test.log` 文件，模拟实时日志；也会将数据写入hdfs 的 `/flume/aggregate` 目录下文件中

**测试结果，如图所示：**  
![在这里插入图片描述][20210628152625586.gif]

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

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

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

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

[Flume_Source_Sink]: https://blog.csdn.net/lzb348110175/article/details/118154430
[Hadoop]: https://blog.csdn.net/lzb348110175/category_7818038.html
[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_hdfs-sink]: http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#hdfs-sink
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6YjM0ODExMDE3NQ_size_16_color_FFFFFF_t_70]: /images/20221005/4bccd74e76b34f6bb41fd264611f94b3.png
[20210628152625586.gif]: /images/20221005/da2abc3e654f441a99d54338b0309e67.png