Flume-1.8.0_部署与常用案例
程序员文章站
2022-06-18 15:45:01
该文章是基于 Hadoop2.7.6_01_部署 进行的 Flume官方文档：FlumeUserGuide 常见问题：记flume部署过程中遇到的问题以及解决方法（持续更新） 1. 前言 在一个完整的大数据处理系统中，除了hdfs+mapreduce+hive组成分析系统的核心之外，还需要数据采集、 ......
该文章是基于 进行的
Flume官方文档：
常见问题：
1. 前言
在一个完整的大数据处理系统中，除了hdfs+mapreduce+hive组成分析系统的核心之外，还需要数据采集、结果数据导出、任务调度等不可或缺的辅助系统，而这些辅助工具在hadoop生态体系中都有便捷的开源框架，如图所示：
2. Flume介绍
2.1. 概述
- Flume是一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统。
- Flume可以采集文件，socket数据包等各种形式源数据，又可以将采集到的数据输出到HDFS、hbase、hive、kafka等众多外部存储系统中
- 一般的采集需求，通过对flume的简单配置即可实现
- Flume针对特殊场景也具备良好的自定义扩展能力，因此，flume可以适用于大部分的日常数据采集场景
2.2. 运行机制
1、 Flume分布式系统中最核心的角色是agent，flume采集系统就是由一个个agent所连接起来形成
2、 每一个agent相当于一个数据传递员，内部有三个组件：
注意：Source 到 Channel 到 Sink之间传递数据的形式是Event事件；Event事件是一个数据流单元。
a) Source：采集源，用于跟数据源对接，以获取数据
b) Sink：下沉地，采集数据的传送目的，用于往下一级agent传递数据或者往最终存储系统传递数据
c)Channel：angent内部的数据传输通道，用于从source将数据传递到sink
3. Flume采集系统结构图
3.1. 简单结构
单个agent采集数据
3.2. 复杂结构
多级agent之间串联
4. Flume的安装部署
4.1. 软件部署
1 [yun@mini01 software]$ pwd 2 /app/software 3 [yun@mini01 software]$ tar xf apache-flume-1.8.0-bin.tar.gz 4 [yun@mini01 software]$ mv apache-flume-1.8.0-bin /app/flume-1.8.0 5 [yun@mini01 software]$ cd /app/ 6 [yun@mini01 ~]$ ln -s flume-1.8.0 flume # 建立软连接 7 [yun@mini01 ~]$ ll 8 total 28 9 lrwxrwxrwx 1 yun yun 11 Jul 25 21:54 flume -> flume-1.8.0 10 drwxrwxr-x 7 yun yun 187 Jul 25 21:53 flume-1.8.0 11 ………………
4.2. 环境变量
1 [root@mini01 profile.d]# pwd 2 /etc/profile.d 3 [root@mini01 profile.d]# cat flume.sh 4 export FLUME_HOME="/app/flume" 5 export PATH=$FLUME_HOME/bin:$PATH 6 [root@mini01 profile.d]# logout 7 [yun@mini01 ~]$ source /etc/profile # 环境变量生效
5. 采集案例
5.1. 简单案例——从网络端口接收数据下沉到logger
配置文件
1 [yun@mini01 conf]$ pwd 2 /app/flume/conf 3 [yun@mini01 conf]$ ll 4 total 20 5 -rw-r--r-- 1 yun yun 1661 Sep 15 2017 flume-conf.properties.template 6 -rw-r--r-- 1 yun yun 1455 Sep 15 2017 flume-env.ps1.template 7 -rw-r--r-- 1 yun yun 1568 Sep 15 2017 flume-env.sh.template 8 -rw-r--r-- 1 yun yun 3107 Sep 15 2017 log4j.properties 9 -rw-rw-r-- 1 yun yun 741 Jul 25 22:12 netcat-logger.conf 10 [yun@mini01 conf]$ cat netcat-logger.conf 11 # Name the components on this agent 12 a1.sources = r1 13 a1.sinks = k1 14 a1.channels = c1 15 16 # Describe/configure the source 17 a1.sources.r1.type = netcat 18 a1.sources.r1.bind = localhost 19 a1.sources.r1.port = 44444 20 # bind = localhost 绑定的是本地端口 21 22 # Describe the sink 23 a1.sinks.k1.type = logger 24 25 # Use a channel which buffers events in memory 26 #下沉的时候是一批一批的, 下沉的时候是一个个eventChannel参数解释： 27 #capacity：默认该通道中最大的可以存储的event数量 28 #trasactionCapacity：每次最大可以从source中拿到或者送到sink中的event数量 29 a1.channels.c1.type = memory 30 a1.channels.c1.capacity = 1000 31 a1.channels.c1.transactionCapacity = 100 32 33 # Bind the source and sink to the channel 34 a1.sources.r1.channels = c1 35 a1.sinks.k1.channel = c1
flume启动
1 [yun@mini01 conf]$ pwd 2 /app/flume/conf 3 # 其中--conf-file 指定的配置文件可以为相对路径也可以是绝对路径 4 [yun@mini01 conf]$ flume-ng agent --conf conf --conf-file netcat-logger.conf --name a1 -Dflume.root.logger=INFO,console 5 ……………… 6 18/07/25 22:19:23 INFO node.Application: Starting Channel c1 7 18/07/25 22:19:23 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: CHANNEL, name: c1: Successfully registered new MBean. 8 18/07/25 22:19:23 INFO instrumentation.MonitoredCounterGroup: Component type: CHANNEL, name: c1 started 9 18/07/25 22:19:23 INFO node.Application: Starting Sink k1 10 18/07/25 22:19:23 INFO node.Application: Starting Source r1 11 18/07/25 22:19:23 INFO source.NetcatSource: Source starting 12 18/07/25 22:19:23 INFO source.NetcatSource: Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]
source端的Telnet输入
1 [yun@mini01 ~]$ telnet localhost 44444 2 Trying ::1... 3 telnet: connect to address ::1: Connection refused 4 Trying 127.0.0.1... 5 Connected to localhost. 6 Escape character is '^]'. 7 111 8 OK 9 222 10 OK 11 334334634geg 12 OK 13 gwegweg 14 OK 15 ^] 16 telnet> quit 17 Connection closed.
当在Telnet端输入，logger显示
1 18/07/25 22:20:09 INFO sink.LoggerSink: Event: { headers:{} body: 31 31 31 0D 111. }
2 18/07/25 22:20:10 INFO sink.LoggerSink: Event: { headers:{} body: 32 32 32 0D 222. }
3 18/07/25 22:20:13 INFO sink.LoggerSink: Event: { headers:{} body: 33 33 34 33 33 34 36 33 34 67 65 67 0D 334334634geg. }
4 18/07/25 22:20:14 INFO sink.LoggerSink: Event: { headers:{} body: 67 77 65 67 77 65 67 0D gwegweg. }
5.2. 监视文件夹——下沉到logger
配置文件
1 [yun@mini01 conf]$ pwd 2 /app/flume/conf 3 [yun@mini01 conf]$ ll 4 total 20 5 -rw-r--r-- 1 yun yun 1661 Sep 15 2017 flume-conf.properties.template 6 -rw-r--r-- 1 yun yun 1455 Sep 15 2017 flume-env.ps1.template 7 -rw-r--r-- 1 yun yun 1568 Sep 15 2017 flume-env.sh.template 8 -rw-r--r-- 1 yun yun 3107 Sep 15 2017 log4j.properties 9 -rw-rw-r-- 1 yun yun 741 Jul 25 22:12 netcat-logger.conf 10 -rw-rw-r-- 1 yun yun 597 Jul 25 22:29 spooldir-logger.conf 11 [yun@mini01 conf]$ cat spooldir-logger.conf 12 # Name the components on this agent 13 a1.sources = r1 14 a1.sinks = k1 15 a1.channels = c1 16 17 # Describe/configure the source 18 #监听目录,spoolDir指定目录, fileHeader要不要给文件加前缀名 19 a1.sources.r1.type = spooldir 20 a1.sources.r1.spoolDir = /app/software/flume 21 a1.sources.r1.fileHeader = true 22 23 # Describe the sink 24 a1.sinks.k1.type = logger 25 26 # Use a channel which buffers events in memory 27 a1.channels.c1.type = memory 28 a1.channels.c1.capacity = 1000 29 a1.channels.c1.transactionCapacity = 100 30 31 # Bind the source and sink to the channel 32 a1.sources.r1.channels = c1 33 a1.sinks.k1.channel = c1
flume启动
1 [yun@mini01 conf]$ pwd 2 /app/flume/conf 3 [yun@mini01 conf]$ flume-ng agent --conf conf --conf-file spooldir-logger.conf --name a1 -Dflume.root.logger=INFO,console 4 ……………… 5 18/07/25 23:01:04 INFO instrumentation.MonitoredCounterGroup: Component type: CHANNEL, name: c1 started 6 18/07/25 23:01:04 INFO node.Application: Starting Sink k1 7 18/07/25 23:01:04 INFO node.Application: Starting Source r1 8 18/07/25 23:01:04 INFO source.SpoolDirectorySource: SpoolDirectorySource source starting with directory: /app/software/flume 9 18/07/25 23:01:04 INFO instrumentation.MonitoredCounterGroup: Monitored counter group for type: SOURCE, name: r1: Successfully registered new MBean. 10 18/07/25 23:01:04 INFO instrumentation.MonitoredCounterGroup: Component type: SOURCE, name: r1 started
往/app/software/flume目录加入文件
1 # 原文件目录 2 [yun@mini01 hive]$ pwd 3 /app/software/hive 4 [yun@mini01 hive]$ ll 5 total 48 6 -rw-rw-r-- 1 yun yun 71 Jul 12 21:53 t_sz01.dat 7 -rw-rw-r-- 1 yun yun 71 Jul 12 21:53 t_sz01.dat2 8 -rw-rw-r-- 1 yun yun 79 Jul 12 22:15 t_sz02_ext.dat 9 -rw-rw-r-- 1 yun yun 52 Jul 12 23:09 t_sz03_20180711.dat1 10 -rw-rw-r-- 1 yun yun 52 Jul 12 23:09 t_sz03_20180711.dat2 11 -rw-rw-r-- 1 yun yun 52 Jul 12 23:09 t_sz03_20180712.dat1 12 -rw-rw-r-- 1 yun yun 52 Jul 12 23:09 t_sz03_20180712.dat2 13 -rw-rw-r-- 1 yun yun 753 Jul 14 10:36 t_sz05_buck.dat 14 -rw-rw-r-- 1 yun yun 507 Jul 14 10:07 t_sz05_buck.dat.bak 15 [yun@mini01 hive]$ cp -a t_access_times.dat t_sz01.dat t_sz01.dat2 ../flume/ 16 [yun@mini01 hive]$ cp -a t_sz05_buck.dat t_sz05_buck.dat2 ../flume/ 17 ############################################ 18 # 对应的flume目录 注意文件名不能重复，否则flume会报错，也不能是一个目录 19 [yun@mini01 flume]$ pwd 20 /app/software/flume 21 [yun@mini01 flume]$ ll 22 total 20 23 -rw-rw-r-- 1 yun yun 288 Jul 18 14:20 t_access_times.dat.COMPLETED 24 -rw-rw-r-- 1 yun yun 71 Jul 12 21:53 t_sz01.dat2.COMPLETED 25 -rw-rw-r-- 1 yun yun 71 Jul 12 21:53 t_sz01.dat.COMPLETED 26 -rw-rw-r-- 1 yun yun 753 Jul 14 10:36 t_sz05_buck.dat 27 -rw-rw-r-- 1 yun yun 753 Jul 14 10:36 t_sz05_buck.dat.COMPLETED
有fileHeader配置
有fileHeader配置
5.3. 用tail命令获取数据，下沉到HDFS
配置文件
1 [yun@mini01 conf]$ pwd 2 /app/flume/conf 3 [yun@mini01 conf]$ ll 4 total 28 5 -rw-r--r-- 1 yun yun 1661 Sep 15 2017 flume-conf.properties.template 6 -rw-r--r-- 1 yun yun 1455 Sep 15 2017 flume-env.ps1.template 7 -rw-r--r-- 1 yun yun 1568 Sep 15 2017 flume-env.sh.template 8 -rw-r--r-- 1 yun yun 3107 Sep 15 2017 log4j.properties 9 -rw-rw-r-- 1 yun yun 741 Jul 25 22:12 netcat-logger.conf 10 -rw-rw-r-- 1 yun yun 593 Jul 25 22:30 spooldir-logger.conf 11 -rw-rw-r-- 1 yun yun 1275 Jul 25 23:29 tail-hdfs.conf 12 [yun@mini01 conf]$ cat tail-hdfs.conf 13 # Name the components on this agent 14 a1.sources = r1 15 a1.sinks = k1 16 a1.channels = c1 17 18 # Describe/configure the source 19 a1.sources.r1.type = exec 20 a1.sources.r1.command = tail -F /app/webservice/logs/access.log 21 a1.sources.r1.channels = c1 22 23 # Describe the sink 24 a1.sinks.k1.type = hdfs 25 a1.sinks.k1.channel = c1 26 a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/ 27 a1.sinks.k1.hdfs.filePrefix = events- 28 # 以下3项表示每隔10分钟切换目录存储 29 a1.sinks.k1.hdfs.round = true 30 a1.sinks.k1.hdfs.roundValue = 10 31 a1.sinks.k1.hdfs.roundUnit = minute 32 # 滚动当前文件前等待的秒数 33 a1.sinks.k1.hdfs.rollInterval = 30 34 # 文件大小以字节为单位触发滚动 35 a1.sinks.k1.hdfs.rollSize = 1024 36 # 在滚动之前写入文件的事件数 37 a1.sinks.k1.hdfs.rollCount = 500 38 # 在它被刷新到HDFS之前写入文件的事件数量。100个事件为一个批次 39 a1.sinks.k1.hdfs.batchSize = 100 40 a1.sinks.k1.hdfs.useLocalTimeStamp = true 41 #生成的文件类型，默认是Sequencefile，可用DataStream，则为普通文本 42 a1.sinks.k1.hdfs.fileType = DataStream 43 44 # Use a channel which buffers events in memory 45 a1.channels.c1.type = memory 46 a1.channels.c1.capacity = 1000 47 a1.channels.c1.transactionCapacity = 100 48 49 # Bind the source and sink to the channel 50 a1.sources.r1.channels = c1 51 a1.sinks.k1.channel = c1
flume启动
1 [yun@mini01 conf]$ flume-ng agent -c conf -f tail-hdfs.conf -n a1
启动jar包打印日志
1 [yun@mini01 webservice]$ pwd 2 /app/webservice 3 [yun@mini01 webservice]$ java -jar testlog.jar &
可参见： ----- 3.3. web日志模拟
浏览器查看flume下沉的数据
5.4. 级联下沉到HDFS
由mini01 的flume发送数据到mini02的flume，然后由mini02的flume下沉到HDFS。
其中mini02的flume安装过程略。
配置文件mini01
1 [yun@mini01 conf]$ pwd 2 /app/flume/conf 3 [yun@mini01 conf]$ ll 4 total 32 5 -rw-r--r-- 1 yun yun 1661 Sep 15 2017 flume-conf.properties.template 6 -rw-r--r-- 1 yun yun 1455 Sep 15 2017 flume-env.ps1.template 7 -rw-r--r-- 1 yun yun 1568 Sep 15 2017 flume-env.sh.template 8 -rw-r--r-- 1 yun yun 3107 Sep 15 2017 log4j.properties 9 -rw-rw-r-- 1 yun yun 741 Jul 25 22:12 netcat-logger.conf 10 -rw-rw-r-- 1 yun yun 593 Jul 25 22:30 spooldir-logger.conf 11 -rw-rw-r-- 1 yun yun 789 Jul 26 22:31 tail-avro-avro-logger.conf 12 -rw-rw-r-- 1 yun yun 1283 Jul 25 23:41 tail-hdfs.conf 13 [yun@mini01 conf]$ cat tail-avro-avro-logger.conf 14 # Name the components on this agent 15 a1.sources = r1 16 a1.sinks = k1 17 a1.channels = c1 18 19 # Describe/configure the source 20 a1.sources.r1.type = exec 21 a1.sources.r1.command = tail -F /app/webservice/logs/access.log 22 a1.sources.r1.channels = c1 23 24 # Describe the sink 25 #绑定的不是本机, 是另外一台机器的服务地址, sink端的avro是一个发送端, avro的客户端, 往mini02这个机器上发 26 a1.sinks = k1 27 a1.sinks.k1.type = avro 28 a1.sinks.k1.channel = c1 29 a1.sinks.k1.hostname = mini02 30 a1.sinks.k1.port = 4141 31 a1.sinks.k1.batch-size = 2 32 33 # Use a channel which buffers events in memory 34 a1.channels.c1.type = memory 35 a1.channels.c1.capacity = 1000 36 a1.channels.c1.transactionCapacity = 100 37 38 # Bind the source and sink to the channel 39 a1.sources.r1.channels = c1 40 a1.sinks.k1.channel = c1
配置文件mini02
1 [yun@mini02 conf]$ pwd 2 /app/flume/conf 3 [yun@mini02 conf]$ ll 4 total 20 5 -rw-rw-r-- 1 yun yun 1357 Jul 26 22:39 avro-hdfs.conf 6 -rw-r--r-- 1 yun yun 1661 Sep 15 2017 flume-conf.properties.template 7 -rw-r--r-- 1 yun yun 1455 Sep 15 2017 flume-env.ps1.template 8 -rw-r--r-- 1 yun yun 1568 Sep 15 2017 flume-env.sh.template 9 -rw-r--r-- 1 yun yun 3107 Sep 15 2017 log4j.properties 10 [yun@mini02 conf]$ cat avro-hdfs.conf 11 # Name the components on this agent 12 a1.sources = r1 13 a1.sinks = k1 14 a1.channels = c1 15 16 # Describe/configure the source 17 #source中的avro组件是接收者服务, 绑定本机 18 a1.sources.r1.type = avro 19 a1.sources.r1.channels = c1 20 a1.sources.r1.bind = 0.0.0.0 21 a1.sources.r1.port = 4141 22 23 # Describe the sink 24 a1.sinks.k1.type = hdfs 25 a1.sinks.k1.channel = c1 26 a1.sinks.k1.hdfs.path = /flume/new-events/%y-%m-%d/%H%M/ 27 a1.sinks.k1.hdfs.filePrefix = events- 28 # 以下3项表示每隔10分钟切换目录存储 29 a1.sinks.k1.hdfs.round = true 30 a1.sinks.k1.hdfs.roundValue = 10 31 a1.sinks.k1.hdfs.roundUnit = minute 32 # 滚动当前文件前等待的秒数 33 a1.sinks.k1.hdfs.rollInterval = 30 34 # 文件大小以字节为单位触发滚动 35 a1.sinks.k1.hdfs.rollSize = 204800 36 # 在滚动之前写入文件的事件数 37 a1.sinks.k1.hdfs.rollCount = 500 38 # 在它被刷新到HDFS之前写入文件的事件数量，每批次事件最大数 39 a1.sinks.k1.hdfs.batchSize = 100 40 a1.sinks.k1.hdfs.useLocalTimeStamp = true 41 #生成的文件类型，默认是Sequencefile，可用DataStream，则为普通文本 42 a1.sinks.k1.hdfs.fileType = DataStream 43 44 # Use a channel which buffers events in memory 45 a1.channels.c1.type = memory 46 a1.channels.c1.capacity = 1000 47 a1.channels.c1.transactionCapacity = 100 48 49 # Bind the source and sink to the channel 50 a1.sources.r1.channels = c1 51 a1.sinks.k1.channel = c1
启动flume
1 # 启动mini02的flume 2 3 4 # 启动mini01的flume
启动jar包打印日志
1 [yun@mini01 webservice]$ pwd 2 /app/webservice 3 [yun@mini01 webservice]$ java -jar testlog.jar &
可参见： ----- 3.3. web日志模拟
浏览器查看flume下沉的数据
6. 更多source和sink组件
Flume支持众多的source和sink类型，详细手册可参考官方文档