Flume-1.8.0_部署与常用案例
时间:2022-07-26
本文章向大家介绍Flume-1.8.0_部署与常用案例，主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项，具有一定的参考价值，需要的朋友可以参考一下。
该文章是基于 Hadoop2.7.6_01_部署 进行的
Flume官方文档：FlumeUserGuide
常见问题：记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 = c1flume启动
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 = c1flume启动
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 = c1flume启动
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 &可参见：Hadoop2.7.6_02_HDFS常用操作 ----- 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 &可参见：Hadoop2.7.6_02_HDFS常用操作 ----- 3.3. web日志模拟
浏览器查看flume下沉的数据
6. 更多source和sink组件
Flume支持众多的source和sink类型，详细手册可参考官方文档
- Spring Cloud实战小贴士：Feign的继承特性(伪RPC模式)
- 仿淘宝购买详情页购买缩小动画
- 高仿今日头条（1）
- android Native堆
- Spring Boot中Web应用的统一异常处理
- android bitmap的内存分配和优化
- Gitflow工作流程
- Android学习第一弹：Android通过用代码画虚线椭圆边框背景来学习一下shape的用法
- Spring Boot中使用Swagger2构建强大的RESTful API文档
- iOS iOS与html进行交互
- Android学习第三弹之Android图片颜色处理
- C#Winform使用扩展方法自定义富文本框(RichTextBox)字体颜色
- Android如何实现毛玻璃效果之Android高级模糊技术
- Android Loader详解
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- .Net Core微服务入门全纪录（八）——Docker Compose与容器网络
- PHP中少用但是很好用的方法
- 使用一维数据构造简单卷积神经网络
- .Net Core微服务入门全纪录（完结）——Ocelot与Swagger
- PHP使用GD库生成文件
- 反向传播算法从原理到实现
- 基于EntityFramework 6 Code First实现多租户的一种思路
- PHP使用GD库生成柱状图
- PHP生成Mysql数据字典
- .Net Core in Docker极简入门（上篇）
- PHP一个比较完善的树形结构代码
- .Net Core in Docker极简入门（下篇）
- snoopy（强大的PHP采集类） 详细介绍
- 十分钟搭建自己的私有NuGet服务器-BaGet
- PHP遍历文件夹下的所有文件和文件夹