三十八、几个Flume的使用案例

骑猪看日落 2021-09-07 05:59 231阅读 0赞

上一篇文章咱们介绍了Flume的安装以及它的简单使用，本章咱们基于Flume来介绍几个具体的案例，具体包括：实时抽取本地文件或目录到HDFS、Flume的扇入扇出等。关注专栏[《破茧成蝶——大数据篇》][Link 1]，查看更多相关的内容~

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

**目录**

一、实时抽取本地文件到HDFS

二、实时抽取本地目录到HDFS

三、Flume的扇入

四、Flume的扇出

4.1 选择器

4.2 sink组

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

# 一、实时抽取本地文件到HDFS #

**1、需求说明**

监控Hive的日志，并实时的上传到HDFS上。

**2、具体实现**

（1）要想使用Flume监控文件上传到HDFS，首先需要拷贝几个jar包到flume的lib目录下，如下所示：

cp /opt/modules/hadoop-2.7.2/share/hadoop/hdfs/hadoop-hdfs-2.7.2.jar /opt/modules/flume/lib/
    cp /opt/modules/hadoop-2.7.2/share/hadoop/common/hadoop-common-2.7.2.jar /opt/modules/flume/lib/
    cp /opt/modules/hadoop-2.7.2/share/hadoop/common/lib/hadoop-auth-2.7.2.jar /opt/modules/flume/lib/
    cp /opt/modules/hadoop-2.7.2/share/hadoop/common/lib/commons-configuration-1.6.jar /opt/modules/flume/lib/
    cp /opt/modules/hadoop-2.7.2/share/hadoop/common/lib/commons-io-2.4.jar /opt/modules/flume/lib/
    cp /opt/modules/hadoop-2.7.2/share/hadoop/common/lib/htrace-core-3.1.0-incubating.jar /opt/modules/flume/lib/

（2）创建配置文件

vim flume-filetoHDFS.conf

# 声明source、channel、sink
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # 配置source
    # 要想读取Linux系统中的文件，就得按照Linux命令的规则执行命令。由于Hive日志在Linux系统中，所以读取文件的类型选择：exec，即execute执行的意思。
    a1.sources.r1.type = exec
    # 日志文件的位置
    a1.sources.r1.command = tail -F /tmp/root/hive.log
    a1.sources.r1.shell = /bin/bash -c
    
    # 配置sink
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://master:9000/flume/%Y%m%d/%H
    # 设置上传文件的前缀
    a1.sinks.k1.hdfs.filePrefix = logs-
    # 设置是否按照时间滚动文件夹
    a1.sinks.k1.hdfs.round = true
    # 设置多少时间单位创建一个新的文件夹
    a1.sinks.k1.hdfs.roundValue = 1
    # 重新定义时间单位
    a1.sinks.k1.hdfs.roundUnit = hour
    # 是否使用本地时间戳
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    # 积攒多少个Event才flush到HDFS一次
    a1.sinks.k1.hdfs.batchSize = 1000
    # 设置文件类型，可支持压缩
    a1.sinks.k1.hdfs.fileType = DataStream
    # 多久生成一个新的文件
    a1.sinks.k1.hdfs.rollInterval = 60
    # 设置每个文件的滚动大小
    a1.sinks.k1.hdfs.rollSize = 134217700
    # 文件的滚动与Event数量无关
    a1.sinks.k1.hdfs.rollCount = 0
    
    # 设置channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # 连接source、channel、sink
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

**3、执行flume**

bin/flume-ng agent --conf conf/ --name a1 --conf-file conf/flume-filetoHDFS.conf

**4、启动hive查看HDFS上是否有数据**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70][]

# 二、实时抽取本地目录到HDFS #

**1、需求说明**

监控本地某目录，并实时的上传到HDFS上。

**2、具体实现**

创建配置文件并添加如下内容：

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    a1.sources.r1.type = spooldir
    a1.sources.r1.spoolDir = /root/files
    a1.sources.r1.fileSuffix = .COMPLETED
    a1.sources.r1.fileHeader = true
    # 忽略所有以.jar结尾的文件，不上传
    a1.sources.r1.ignorePattern = ([^ ]*\.jar)
    
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://master:9000/flume/upload/%Y%m%d/%H
    # 上传文件的前缀
    a1.sinks.k1.hdfs.filePrefix = dir-
    # 是否按照时间滚动文件夹
    a1.sinks.k1.hdfs.round = true
    # 多少时间单位创建一个新的文件夹
    a1.sinks.k1.hdfs.roundValue = 1
    # 重新定义时间单位
    a1.sinks.k1.hdfs.roundUnit = hour
    # 是否使用本地时间戳
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    # 积攒多少个Event才flush到HDFS一次
    a1.sinks.k1.hdfs.batchSize = 100
    # 设置文件类型，可支持压缩
    a1.sinks.k1.hdfs.fileType = DataStream
    # 多久生成一个新的文件
    a1.sinks.k1.hdfs.rollInterval = 60
    # 设置每个文件的滚动大小大概是128M
    a1.sinks.k1.hdfs.rollSize = 134217700
    # 文件的滚动与Event数量无关
    a1.sinks.k1.hdfs.rollCount = 0
    
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

**3、执行flume**

bin/flume-ng agent --conf conf/ --name a1 --conf-file conf/flume-dirtoHDFS.conf

**4、查看HDFS上是否有同步**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1][]

# 三、Flume的扇入 #

顾名思义，扇入的意思就是说Flume有多个数据源汇入到一个source，如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2][]

**1、需求说明**

slave01上的Flume监控一个日志文件，slave02上的Flume监控某个端口的数据流，最后两台slave节点的Flume将数据发送给master节点的Flume，master节点接收到数据之后打印到控制台。

**2、具体实现**

（1）分发Flume到slave节点

xsync /opt/modules/flume

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 3][]

（2）编写slave01节点的配置文件

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /root/files/xzw.log
    a1.sources.r1.shell = /bin/bash -c
    
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = master
    a1.sinks.k1.port = 6868
    
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

（3）编写slave02节点的配置文件

a2.sources = r1
    a2.sinks = k1
    a2.channels = c1
    
    a2.sources.r1.type = netcat
    a2.sources.r1.bind = slave02
    a2.sources.r1.port = 44444
    
    a2.sinks.k1.type = avro
    a2.sinks.k1.hostname = master
    a2.sinks.k1.port = 6868
    
    a2.channels.c1.type = memory
    a2.channels.c1.capacity = 1000
    a2.channels.c1.transactionCapacity = 100
    
    a2.sources.r1.channels = c1
    a2.sinks.k1.channel = c1

（4）编写master节点的配置文件

a3.sources = r1
    a3.sinks = k1
    a3.channels = c1
    
    a3.sources.r1.type = avro
    a3.sources.r1.bind = master
    a3.sources.r1.port = 6868
    
    a3.sinks.k1.type = logger
    
    a3.channels.c1.type = memory
    a3.channels.c1.capacity = 1000
    a3.channels.c1.transactionCapacity = 100
    
    a3.sources.r1.channels = c1
    a3.sinks.k1.channel = c1

（5）启动flume

bin/flume-ng agent --conf conf/ --name a3 --conf-file conf/flume-mtocon.conf -Dflume.root.logger=INFO,console
    bin/flume-ng agent --conf conf/ --name a2 --conf-file conf/flume-s2tom.conf -Dflume.root.logger=INFO,console
    bin/flume-ng agent --conf conf/ --name a1 --conf-file conf/flume-s1tom.conf -Dflume.root.logger=INFO,console

（6）测试

向log中追加数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 4][]

向端口发送数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 5][]

在控制台可以发现已经打印出了数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 6][]

# 四、Flume的扇出 #

Flume的扇出是指Flume在接收到数据后sink到不同的地点，我们分为两部分来讲，分别是用选择器实现和sink组实现。

## 4.1 选择器 ##

使用选择器实现扇出功能，是用的单source、多channel和sink的方法，如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 7][]

**1、需求说明**

使用Flume A监控某个文件的变化，Flume A将变动的文件内容实时的传送给Flume B，Flume B将内容sink到HDFS上。同时，Flume A也将变动内容传送给Flume C，Flume C将内容存储到本地。

**2、具体实现**

（1）编写Flume A的配置文件

a1.sources = r1
    a1.sinks = k1 k2
    a1.channels = c1 c2
    # 将数据流复制给所有channel
    a1.sources.r1.selector.type = replicating
    
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /tmp/root/hive.log
    a1.sources.r1.shell = /bin/bash -c
    
    # sink端的avro是一个数据发送者，Avro是由Hadoop创始人Doug Cutting创建的一种语言无关的数据序列化和RPC框架。
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = master 
    a1.sinks.k1.port = 8888
    
    a1.sinks.k2.type = avro
    a1.sinks.k2.hostname = master
    a1.sinks.k2.port = 6868
    
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    a1.channels.c2.type = memory
    a1.channels.c2.capacity = 1000
    a1.channels.c2.transactionCapacity = 100
    
    a1.sources.r1.channels = c1 c2
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c2

（2）编写Flume B的配置文件

a2.sources = r1
    a2.sinks = k1
    a2.channels = c1
    
    # source端的avro是一个数据接收服务
    a2.sources.r1.type = avro
    a2.sources.r1.bind = master
    a2.sources.r1.port = 8888
    
    a2.sinks.k1.type = hdfs
    a2.sinks.k1.hdfs.path = hdfs://master:9000/flume-b/%Y%m%d/%H
    #上传文件的前缀
    a2.sinks.k1.hdfs.filePrefix = flume-b-
    #是否按照时间滚动文件夹
    a2.sinks.k1.hdfs.round = true
    #多少时间单位创建一个新的文件夹
    a2.sinks.k1.hdfs.roundValue = 1
    #重新定义时间单位
    a2.sinks.k1.hdfs.roundUnit = hour
    #是否使用本地时间戳
    a2.sinks.k1.hdfs.useLocalTimeStamp = true
    #积攒多少个Event才flush到HDFS一次
    a2.sinks.k1.hdfs.batchSize = 100
    #设置文件类型，可支持压缩
    a2.sinks.k1.hdfs.fileType = DataStream
    #多久生成一个新的文件
    a2.sinks.k1.hdfs.rollInterval = 600
    #设置每个文件的滚动大小大概是128M
    a2.sinks.k1.hdfs.rollSize = 134217700
    #文件的滚动与Event数量无关
    a2.sinks.k1.hdfs.rollCount = 0
    
    a2.channels.c1.type = memory
    a2.channels.c1.capacity = 1000
    a2.channels.c1.transactionCapacity = 100
    
    a2.sources.r1.channels = c1
    a2.sinks.k1.channel = c1

（3）编写Flume C的配置文件

a3.sources = r1
    a3.sinks = k1
    a3.channels = c2
    
    a3.sources.r1.type = avro
    a3.sources.r1.bind = master
    a3.sources.r1.port = 6868
    
    a3.sinks.k1.type = file_roll
    a3.sinks.k1.sink.directory = /root/files/flume-c
    
    a3.channels.c2.type = memory
    a3.channels.c2.capacity = 1000
    a3.channels.c2.transactionCapacity = 100
    
    a3.sources.r1.channels = c2
    a3.sinks.k1.channel = c2

（4）分别启动三个Flume

bin/flume-ng agent --conf conf/ --name a3 --conf-file conf/flume-c.conf
    bin/flume-ng agent --conf conf/ --name a2 --conf-file conf/flume-b.conf
    bin/flume-ng agent --conf conf/ --name a1 --conf-file conf/flume-a.conf

（5）测试发现，在本地和HDFS上均发现了上传的日志文件

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 8][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 9][]

## 4.2 sink组 ##

选择使用sink组实现扇出的功能，是用的单source、channel，多sink的方法，这种方法实际上是实现了Flume的负载均衡。如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 10][]

**1、需求说明**

使用Flume D监控端口的变化，Flume D将监控到的内容实时的传送给Flume E，Flume E将内容打印到控制台。同时，Flume D也将变动内容传送给Flume F，Flume F将内容也打印到控制台。

**2、具体实现**

（1）编写Flume D的配置文件

a1.sources = r1
    a1.channels = c1
    a1.sinkgroups = g1
    a1.sinks = k1 k2
    
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    
    a1.sinkgroups.g1.processor.type = load_balance
    a1.sinkgroups.g1.processor.backoff = true
    a1.sinkgroups.g1.processor.selector = round_robin
    a1.sinkgroups.g1.processor.selector.maxTimeOut=10000
    
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = master
    a1.sinks.k1.port = 8888
    
    a1.sinks.k2.type = avro
    a1.sinks.k2.hostname = master
    a1.sinks.k2.port = 6868
    
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    a1.sources.r1.channels = c1
    a1.sinkgroups.g1.sinks = k1 k2
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c1

（2）编写Flume E的配置文件

a2.sources = r1
    a2.sinks = k1
    a2.channels = c1
    
    a2.sources.r1.type = avro
    a2.sources.r1.bind = master
    a2.sources.r1.port = 8888
    
    a2.sinks.k1.type = logger
    
    a2.channels.c1.type = memory
    a2.channels.c1.capacity = 1000
    a2.channels.c1.transactionCapacity = 100
    
    a2.sources.r1.channels = c1
    a2.sinks.k1.channel = c1

（3）编写Flume F的配置文件

a3.sources = r1
    a3.sinks = k1
    a3.channels = c2
    
    a3.sources.r1.type = avro
    a3.sources.r1.bind = master
    a3.sources.r1.port = 6868
    
    a3.sinks.k1.type = logger
    
    a3.channels.c2.type = memory
    a3.channels.c2.capacity = 1000
    a3.channels.c2.transactionCapacity = 100
    
    a3.sources.r1.channels = c2
    a3.sinks.k1.channel = c2

（4）分别启动三个Flume

bin/flume-ng agent --conf conf/ --name a3 --conf-file conf/flume-f.conf -Dflume.root.logger=INFO,console
    bin/flume-ng agent --conf conf/ --name a2 --conf-file conf/flume-e.conf -Dflume.root.logger=INFO,console
    bin/flume-ng agent --conf conf/ --name a1 --conf-file conf/flume-d.conf -Dflume.root.logger=INFO,console

（5）开启端口，生产数据

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 11][]

可以看到控制台打印出了传输的内容：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 12][]

![20210107085159538.png][]

好了，以上就是Flume的几个使用案例，比较简单，你们在这个过程中遇到了什么问题，欢迎留言，让我看看你们遇到了什么问题~

[Link 1]: https://blog.csdn.net/gdkyxy2013/category_10006538.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70]: /images/20210726/30d29a7f2f844b20bd9b42e4aa7fc58a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1]: /images/20210726/d558b04ac31f40d5bbfac983a3512003.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2]: /images/20210726/d29226ab5ba3464399c00b979aab0377.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 3]: /images/20210726/19e6da45f9d147f48c53263110953600.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 4]: /images/20210726/43623d10fcaa471d9896e52f08976d12.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 5]: /images/20210726/88918b9ae5c94a7da4299d3b446f4785.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 6]: /images/20210726/fd9ba32120aa40a4b3c5b1ec3fe1220f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 7]: /images/20210726/c44813dd8eec43bda9f5c61079f3c4d0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 8]: /images/20210726/35651ef6ee014ad8b128e7a9054980eb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 9]: /images/20210726/208fa12dc9be4d8fb47b982874c1d680.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 10]: /images/20210726/0c03242e6b69404682b0d81c44f513a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 11]: /images/20210726/2bd957e5c0e04190a80cd7a01ae7a74a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 12]: /images/20210726/ea7da3114b1d4576ac06e9edeb0394ba.png
[20210107085159538.png]: /images/20210726/3ddf6d6dddf54d1daebf2d3157c0caa9.png