Flume学习

一、Flume概述

Flume最主要的作用就是，实时读取服务器本地磁盘的数据，将数据写入到HDFS。

二、Flume基础架构

三、Flume安装部署

配置Flume的前提是要配置好JDK和Hadoop

1.解压

[root@lxm148 soft]# tar -zxvf ./apache-flume-1.9.0-bin.tar.gz -C /opt/soft/

2.修改文件名

[root@lxm148 soft]# mv ./apache-flume-1.9.0-bin/ ./flume190

3.将 lib 文件夹下的 guava-11.0.2.jar 删除以兼容 Hadoop313

[root@lxm148 flume190]# rm /opt/soft/flume190/lib/guava-11.0.2.jar 
rm: remove regular file ‘/opt/soft/flume190/lib/guava-11.0.2.jar’? y

4.配置环境变量

# FLUME_HOME
export FLUME_HOME=/opt/soft/flume190
export PATH=$PATH:$FLUME_HOME/bin# 刷新环境变量
source /etc/profile

5.修改配置文件

[root@lxm148 flume190]# cd ./conf/[root@lxm148 conf]# lsflume-conf.properties.template
flume-env.ps1.template
flume-env.sh.template
log4j.properties

6.修改flume-env.sh文件

[root@lxm148 conf]# mv ./flume-env.sh.template ./flume-env.sh[root@lxm148 conf]# vim ./flume-env.sh 
# 第22行加载JAVA路径
# Enviroment variables can be set here.
export JAVA_HOME=/opt/soft/jdk180

7.检验是否安装成功

[root@lxm148 conf]# flume-ng versionFlume 1.9.0
Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
Revision: d4fcab4f501d41597bc616921329a4339f73585e
Compiled by fszabo on Mon Dec 17 20:45:25 CET 2018
From source with checksum 35db629a3bda49d23e9b3690c80737f9

四、Flume入门案例

(一)监控端口数据官方案例

1.案例需求：

使用 Flume 监听一个端口，收集该端口数据，并打印到控制台。

2.安装步骤：

通过netcat工具向本机的44444端口发送数据。
Flume监控本机的44444端口，通过Flume的source端读取数据。
Flume将获取的数据通过Sink端写出到控制台。

(1)安装 netcat 工具

[root@lxm148 flume190]#  yum install -y nc

(2)查看一个没有用过的端口，查看它是否被占用，例如9999(可忽略)

[root@lxm148 flume190]#  netstat -nlp | grep 9999

(3)如果没有被占用，就启动该端口，相当于开启服务端(可忽略)

[root@lxm148 flume190]# nc -lk 9999

(4)新开一个窗口，监听该端口，相当于开启客户端(可忽略)

[root@lxm148 flume190]# nc localhost 9999

(5)测试服务端与客户端能否连接(可忽略)

-- 服务器输入hello
[root@lxm148 flume190]# nc -lk 9999
hello-- 客户端接收并输出hello
[root@lxm148 flume190]# nc localhost 9999
hello

(6)新建目录和文件，并输入文件内容

[root@lxm148 flume190]# mkdir job
[root@lxm148 flume190]# cd job/
[root@lxm148 job]# ll
total 0
[root@lxm148 job]# touch net-flume-logger.conf
[root@lxm148 job]# ll
total 0
-rw-r--r--. 1 root root 0 Feb 28 23:12 net-flume-logger.conf
[root@lxm148 job]# vim ./net-flume-logger.conf -----------------------------文件中粘贴下面的内容--------------------------------# Name the components on this agent # a1:表示agent的名称
#r1:表示a1的Source的名称
a1.sources = r1		
#k1:表示a1的Sink的名称
a1.sinks = k1		
#c1:表示a1的Channel的名称
a1.channels = c1	# Describe/configure the source	
#表示a1的输入源类型为netcat端口类型
a1.sources.r1.type = netcat		
#表示a1的监听的主机
a1.sources.r1.bind = localhost	
#表示a1的监听的端口号
a1.sources.r1.port = 44444		
# Describe the sink
#表示a1的输出目的地是控制台logger类型
a1.sinks.k1.type = logger		# Use a channel which buffers events in memory
#表示a1的channel类型是memory内存型
a1.channels.c1.type = memory	
#表示a1的channel总容量1000个event
a1.channels.c1.capacity = 1000	
#表示a1的channel传输时收集到了100条event以后再去提交事务
a1.channels.c1.transactionCapacity = 100	
# Bind the source and sink to the channel
#表示将r1和c1连接起来，一个sources可以绑定多个channel
a1.sources.r1.channels = c1		
#表示将k1和c1连接起来，一个sink只能绑定一个channel
a1.sinks.k1.channel = c1

上述代码说明：

代码来源于：https://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html

a1是agent的名字，同一台flume启用多个agent，agent的名字不能相同
source、sink、channel后面的s表示后面的可以配置多个
事务的容量(transactionCapacity)<总容量(capacity)
一个source可以绑定多个channel，一个sink必须绑定一个channel

(7)开启监听

# 第一种写法：
[root@lxm148 flume190]# bin/flume-ng agent -n a1 -c conf/ -f job/net-flume-logger.conf -Dflume.root.logger=INFO,console# 第二种写法：
[root@lxm148 flume190]# bin/flume-ng agent -c conf/ -n a1 -f job/flume-netcat-logger.conf -Dflume.root.logger=INFO,console

开启监听参数说明：

--conf/-c：表示配置文件存储在 conf/目录

--name/-n：表示给 agent 起名为 a1

--conf-file/-f：flume 本次启动读取的配置文件是在 job 文件夹下的 flume-telnet.conf文件。

-Dflume.root.logger=INFO,console ：-D 表示 flume 运行时动态修改 flume.root.logger

参数属性值，并将控制台日志打印级别设置为 INFO 级别。日志级别包括:log、info、warn、

error

因为文件中配置的端口是44444，所以这里监听的是44444端口

(8)开启客户端

[root@lxm148 ~]# nc localhost 44444

尝试在客户端输入，服务器会进行接收

(二)实时监控单个文件到HDFS

1.案例需求：

实时监控Hive日志，并上传到HDFS中，即实时读取本地文件到HDFS。

2.实现步骤：

创建符合条件的 flume 配置文件

执行配置文件，开启监控

开启 Hive，生成日志

查看HDFS上的数据

(1)创建 flume-file-hdfs.conf 文件

[root@lxm148 ~]# vim /opt/soft/flume190/job/flume-file-hdfs.conf# Name the components on this agent
a2.sources = r2
a2.sinks = k2
a2.channels = c2
# Describe/configure the source
# 定义scource类型为exec可执行命令的
a2.sources.r2.type = exec
a2.sources.r2.command = tail -F /opt/soft/hive312/logs/hive.log
# Describe the sink
a2.sinks.k2.type = hdfs
# 这里的端口号要与/opt/soft/hadoop313/etc/hadoop/core-site.xml中的hdfs的端口号一致，否则不会生成flume目录
a2.sinks.k2.hdfs.path = hdfs://lxm148:9000/flume/%Y%m%d/%H
#上传文件的前缀
a2.sinks.k2.hdfs.filePrefix = logs-
#是否按照时间滚动文件夹
a2.sinks.k2.hdfs.round = true
#多少时间单位创建一个新的文件夹
a2.sinks.k2.hdfs.roundValue = 1
#重新定义时间单位
a2.sinks.k2.hdfs.roundUnit = hour
#是否使用本地时间戳
a2.sinks.k2.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a2.sinks.k2.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a2.sinks.k2.hdfs.fileType = DataStream
#多久生成一个新的文件，单位是秒s
a2.sinks.k2.hdfs.rollInterval = 30
#设置每个文件的滚动大小
a2.sinks.k2.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a2.sinks.k2.hdfs.rollCount = 0
# Use a channel which buffers events in memory
a2.channels.c2.type = memory 
a2.channels.c2.capacity = 1000 
a2.channels.c2.transactionCapacity = 100
# Bind the source and sink to the channel
a2.sources.r2.channels = c2
a2.sinks.k2.channel = c2

(2)启动HDFS

start-dfs.sh
start-yarn.sh

(3)运行Flume

[root@lxm148 flume190]# flume-ng agent --conf conf/ --name a2 --conf-file job/flume-file-hdfs.conf# 或者
[root@lxm148 flume190]# flume-ng agent -n a2 -c conf/ -f job/flume-file-hdfs.conf

(4)启动Hive

[root@lxm148 hive312]# hive

(5)刷新HDFS，flume目录生成

hive每操作一次，就会生成一个.tmp日志文件，30秒后.tmp就会去掉

(三)实时监控目录下多个新文件到HDFS

1.案例需求：

使用 Flume 监听整个目录的文件，并上传至 HDFS。

2.实现步骤：

创建符合条件的flume配置文件
执行配置文件，开启监控
向upload目录中添加文件
查看HDFS上的数据
查看/opt/soft/flume190/upload目录中上传的文件是否已经标记为.COMPLETED结尾；.tmp后缀结尾文件没有上传。

(1)配置文件

[root@lxm148 flume190]# vim ./job/flume-dir-hdfs.confa3.sources = r3
a3.sinks = k3
a3.channels = c3
# Describe/configure the source
# 定义source类型为目录
a3.sources.r3.type = spooldir
a3.sources.r3.spoolDir = /opt/soft/flume190/upload
a3.sources.r3.fileSuffix = .COMPLETED
a3.sources.r3.fileHeader = true
# 忽略所有以.tmp结尾的文件，不上传
a3.sources.r3.ignorePattern = ([^ ]*\.tmp)# Describe the sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path = hdfs://lxm148:9000/flume/upload/%Y%m%d/%H
#上传文件的前缀
a3.sinks.k3.hdfs.filePrefix = upload-
#是否按照时间滚动文件夹
a3.sinks.k3.hdfs.round = true
#多少时间单位创建一个新的文件夹
a3.sinks.k3.hdfs.roundValue = 1
#重新定义时间单位
a3.sinks.k3.hdfs.roundUnit = hour
#是否使用本地时间戳
a3.sinks.k3.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a3.sinks.k3.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a3.sinks.k3.hdfs.fileType = DataStream
#多久生成一个新的文件
a3.sinks.k3.hdfs.rollInterval = 60
#设置每个文件的滚动大小大概是 128M
a3.sinks.k3.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a3.sinks.k3.hdfs.rollCount = 0
# Use a channel which buffers events in memory
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000
a3.channels.c3.transactionCapacity = 100
# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3

Taildir 说明：

        Taildir Source 维护了一个 json 格式的 position File，其会定期的往 position File中更新每个文件读取到的最新的位置，因此能够实现断点续传。

        Position File 的格式如下：

        {"inode":2496272,"pos":12,"file":"/opt/module/flume/files/file1.txt"}

        {"inode":2496275,"pos":12,"file":"/opt/module/flume/files/file2.txt"}

        注：Linux 中储存文件元数据的区域就叫做 inode ，每个 inode 都有一个号码，操作系统

用 inode 号码来识别不同的文件， Unix/Linux 系统内部不使用文件名，而使用 inode 号码来

识别文件。

(2)创建upload目录

[root@lxm148 flume190]# mkdir upload[root@lxm148 flume190]# hdfs dfs -mkdir /flume/upload

(3)执行文件

[root@lxm148 flume190]# bin/flume-ng agent -n a3 -c conf/ -f job/flume-dir-hdfs.conf

(4)新建测试文件，并放入/opt/soft/flume190/upload目录下

[root@lxm148 flume190]# vim ./test.txthello java
hello flume
hello world[root@lxm148 flume190]# mv ./test.txt ./upload/[root@lxm148 flume190]# ls ./upload/
test.txt.COMPLETED

(5)查看HDFS上的目录

(6)不会上传到HDFS上的情况

1)如果上传的文件名相同就会报错

[root@lxm148 flume190]# vim ./test.txt
[root@lxm148 flume190]# mv ./test.txt ./upload/

java.lang.IllegalStateException: File name has been re-used with different files. Spooling assumptions violated for /opt/soft/flume190/upload/test.txt.COMPLETEDat org.apache.flume.client.avro.ReliableSpoolingFileEventReader.rollCurrentFile(ReliableSpoolingFileEventReader.java:528)at org.apache.flume.client.avro.ReliableSpoolingFileEventReader.retireCurrentFile(ReliableSpoolingFileEventReader.java:475)at org.apache.flume.client.avro.ReliableSpoolingFileEventReader.readEvents(ReliableSpoolingFileEventReader.java:386)at org.apache.flume.source.SpoolDirectorySource$SpoolDirectoryRunnable.run(SpoolDirectorySource.java:263)at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)at java.lang.Thread.run(Thread.java:750)

每次都会扫描该文件进行上传

因此，需要将该文件删除

[root@lxm148 flume190]# rm -rf ./upload/test.txt

2)文件名不同，文件后缀名是.COMPLETED或.tmp也不会上传

3)=号后面不要换行，如下图，两行内容要放在一行

a3.sinks.k3.hdfs.path = 
hdfs://lxm148:9000/flume/upload/%Y%m%d/%H

修改后：

a3.sinks.k3.hdfs.path = hdfs://lxm148:9000/flume/upload/%Y%m%d/%H

(7)总结

上传的文件后缀名不能是忽略掉的后缀(.COMPLETED或.tmp)
上传的文件名不可以相同

(四)实时监控目录下的多个追加文件到HDFS

Exec source 适用于监控一个实时追加的文件，不能实现断点续传；Spooldir Source

适合用于同步新文件，但不适合对实时追加日志的文件进行监听并同步；而 Taildir Source

适合用于监听多个实时追加的文件，并且能够实现断点续传。

1.案例需求：

使用 Flume 监听整个目录的实时追加文件，并上传至 HDFS。

2.实现步骤：

创建符合条件的flume配置文件
执行配置文件，开启监控
向监控文件追加内容 echo hello >> files/file1.txt echo hello >> files/file2.txt
查看HDFS上的数据

(1)在/opt/soft/flume190/job目录下创建一个文件

[root@lxm148 flume190]# vim ./job/flume-taildir-hdfs.conf a3.sources = r3
a3.sinks = k3
a3.channels = c3
# Describe/configure the source
a3.sources.r3.type = TAILDIR
a3.sources.r3.positionFile = /opt/soft/flume190/tail_dir.json
a3.sources.r3.filegroups = f1 f2
a3.sources.r3.filegroups.f1 = /opt/soft/flume190/files/.*file.*
a3.sources.r3.filegroups.f2 = /opt/soft/flume190/files2/.*log.*
# Describe the sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path = hdfs://lxm148:9000/flume/upload2/%Y%m%d/%H
#上传文件的前缀
a3.sinks.k3.hdfs.filePrefix = upload-
#是否按照时间滚动文件夹
a3.sinks.k3.hdfs.round = true
#多少时间单位创建一个新的文件夹
a3.sinks.k3.hdfs.roundValue = 1
#重新定义时间单位
a3.sinks.k3.hdfs.roundUnit = hour
#是否使用本地时间戳
a3.sinks.k3.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a3.sinks.k3.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a3.sinks.k3.hdfs.fileType = DataStream
#多久生成一个新的文件
a3.sinks.k3.hdfs.rollInterval = 20
#设置每个文件的滚动大小大概是 128M
a3.sinks.k3.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a3.sinks.k3.hdfs.rollCount = 0
# Use a channel which buffers events in memory
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000
a3.channels.c3.transactionCapacity = 100
# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3

(2).启动flume

[root@lxm148 flume190]# bin/flume-ng agent -n a3 -c conf/ -f job/flume-taildir-hdfs.conf

(3).开始测试

[root@lxm148 flume190]# hdfs dfs -mkdir /flume/upload2 
# 上面这条命令也可以不写，因为创建好files目录以及该目录下的文件后，HDFS上会自动创建upload2目录[root@lxm148 flume190]# mkdir files[root@lxm148 flume190]# mkdir files2[root@lxm148 flume190]# cd ./files[root@lxm148 files]# touch file3.txt[root@lxm148 files]# echo hello>> file3.txt

文件上传到HDFS

(4).在设定时间内上传两个内容

[root@lxm148 files]# echo java >> file3.txt 
[root@lxm148 files]# echo hadoop >> file3.txt

flume会将两次追加的内容一起上传的HDFS同一个文件中

(5).修改文件名再次追加，会将文件全部再次上传的解决办法

[root@lxm148 files]# mv file3.txt file4.txt 
[root@lxm148 files]# echo mysql >> file4.txt

此时HDFS上新上传的文件，会包含之前上传文件的所有内容，并没有断开上传，且文件大小约等于之前上传文件之和。

(注：我这里比6+12=18大很多是因为我在之前追加了好几次内容，你的应该是20左右)

更正上述问题的方法——修改flume源文件：

(1)在官网上下载Flume1.9的源码

http://archive.apache.org/dist/flume/1.9.0/

(2)找到下面的目录文件

E:\flume\apache-flume-1.9.0-src\flume-ng-sources\flume-taildir-source

(3)在IDEA中打开这个项目，修改代码，打包编译

按照下图中两个类中的红框内容修改代码，打包编译

(4)删除/opt/soft/flume190/lib/目录下原有的flume-taildir-source-1.9.0.jar包

(5)将新打包好的jar包放到/opt/soft/flume190/lib/目录下

(6)重新启动Flume
[root@lxm148 flume190]# bin/flume-ng agent -n a3 -c conf/ -f job/flume-taildir-hdfs.cf f
(7)再进行追加内容到文件中，文件就会断开上传，不会重复上传之前的内容

五、Flume进阶

(一)Flume事务

(二)Flume Agent内部原理

DefaultSinkProcessor对应的是单个的Sink；
LoadBalancingSinkProcessor对应Sink Group，实现负载均衡的功能；
FailoverSinkProcessor对应Sink Group，实现错误恢复的功能。

(三)Flume拓扑结构

1.简单串联

2.复制和多路复用

(1)案例需求

使用 Flume-1 监控文件变动，Flume-1 将变动内容传递给 Flume-2，Flume-2 负责存储

到 HDFS。同时 Flume-1 将变动内容传递给 Flume-3，Flume-3 负责输出到 Local

FileSystem。

(2)实现步骤

1)准备工作，创建文件夹

[root@lxm148 job]# mkdir group1[root@lxm148 job]# mkdir -p /opt/soft/datas/flume3[root@lxm148 job]# cd group1

2)在group1目录下创建文件

[root@lxm148 group1]# vim flume-file-flume.conf# Name the components on this agent
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
# 将数据流复制给所有 channel
a1.sources.r1.selector.type = replicating
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /opt/soft/hive312/logs/hive.log
a1.sources.r1.shell = /bin/bash -c
# Describe the sink
# sink 端的 avro 是一个数据发送者
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = lxm148
a1.sinks.k1.port = 4141
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = lxm148
a1.sinks.k2.port = 4142
# Describe the channel
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
# Bind the source and sink to the channel
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

[root@lxm148 group1]# vim flume-flume-hdfs.conf# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1
# Describe/configure the source
# source 端的 avro 是一个数据接收服务
a2.sources.r1.type = avro
a2.sources.r1.bind = lxm148
a2.sources.r1.port = 4141
# Describe the sink
a2.sinks.k1.type = hdfs
a2.sinks.k1.hdfs.path = hdfs://lxm148:9000/flume2/%Y%m%d/%H
#上传文件的前缀
a2.sinks.k1.hdfs.filePrefix = flume2-
#是否按照时间滚动文件夹
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 = 30
#设置每个文件的滚动大小大概是 128M
a2.sinks.k1.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a2.sinks.k1.hdfs.rollCount = 0
# 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

[root@lxm148 group1]# vim flume-flume-dir.conf# 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 = lxm148
a3.sources.r1.port = 4142
# Describe the sink
a3.sinks.k1.type = file_roll
a3.sinks.k1.sink.directory = /opt/soft/datas/flume3# 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

3)执行配置文件

[root@lxm148 group1]#  bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group1/flume-flume-dir.conf[root@lxm148 group1]#  bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group1/flume-flume-hdfs.conf[root@lxm148 group1]#  bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group1/flume-file-flume.conf

4)启动hadoop和hive