Flume之安装与入门案例分析

1. Flume 安装部署

1.1 安装地址

1. Flume 官网地址&＃xff1a;http://flume.apache.org/
2. 文档查看地址&＃xff1a;http://flume.apache.org/FlumeUserGuide.html
3. 下载地址&＃xff1a;http://archive.apache.org/dist/flume/

1.2 安装部署

1. 解压 apache-flume-1.9.0-bin.tar.gz 到/opt/module/目录下&＃xff0c;并修改名称

   tar -zxf /opt/software/apacheflume-1.9.0-bin.tar.gz -C /opt/module/
mv /opt/module/apache-flume-1.9.0-bin /opt/module/flume-1.9.0


2. 将 lib 文件夹下的 guava-11.0.2.jar 删除以兼容 Hadoop 3.1.3

   rm /opt/module/flume-1.9.0/lib/guava11.0.2.jar


2. Flume 入门案例

2.1 监控端口数据官方案例

2.1.1. 案例需求

使用 Flume 监听一个端口&＃xff0c;收集该端口数据&＃xff0c;并打印到控制台

2.1.2. 需求分析

2.1.3. 实现步骤

1. 安装 netcat 工具

   sudo yum install -y nc


2. 判断 44444 端口是否被占用

   sudo netstat -nlp | grep 44444


3. 创建 Flume Agent 配置文件 flume-netcat-logger.conf&＃xff0c;并添加如下内容&＃xff0c;参考http://flume.apache.org/FlumeUserGuide.html#a-simple-example

   # Name the components on this agent
# a1&＃xff1a;表示agent的名称
# r1&＃xff1a;表示a1的source名称
a1.sources &＃61; r1
# k1&＃xff1a;表示a1的sink名称
a1.sinks &＃61; k1
# c1&＃xff1a;表示a1的channel名称
a1.channels &＃61; c1# Describe/configure the source
# a1的输入源类型为netcat端口类型
a1.sources.r1.type &＃61; netcat
# a1的监听主机号
a1.sources.r1.bind &＃61; localhost
# a1的监听端口号
a1.sources.r1.port &＃61; 44444# Describe the sink
# a1的输出目的地是控制台logger类型
a1.sinks.k1.type &＃61; logger# Use a channel which buffers events in memory
# a1的channel类型为memory
a1.channels.c1.type &＃61; memory
# a1的channel总容量为1000个event
a1.channels.c1.capacity &＃61; 1000
# a1的channel传输是收集到100个event以后再去提交事务
a1.channels.c1.transactionCapacity &＃61; 100# Bind the source and sink to the channel
# 将r1与c1连接起来
a1.sources.r1.channels &＃61; c1
# 将c1与k1连接起来
a1.sinks.k1.channel &＃61; c1


4. 先开启 flume 监听端口

   // 第一种写法
[codecat&＃64;hadoop102 flume-1.9.0]$ flume-ng agent --conf conf/ --name a1 --conf-file jobs/flume-netcat-logger.conf -Dflume.root.logger&＃61;INFO,console// 第二种写法
[codecat&＃64;hadoop102 flume-1.9.0]$ flume-ng agent -c conf/ -n a1 -f jobs/flume-netcat-logger.conf -Dflume.root.logger&＃61;INFO,console


   • --conf/-c&＃xff1a;表示配置文件存储在 conf/目录
   • --name/-n&＃xff1a;表示agent 的名称
   • --conf-file/-f&＃xff1a;flume 本次启动读取的配置文件路径
   • -Dflume.root.logger&＃61;INFO,console &＃xff1a;-D 表示 flume 运行时动态修改 flume.root.logger参数属性值&＃xff0c;并将控制台日志打印级别设置为 INFO 级别
     

5. 使用 netcat 工具向本机的 44444 端口发送内容
   

6. 在 Flume 监听页面观察接收数据情况
   

2.2 实时监控单个追加文件

2.2.1 案例需求

实时监控 Hive 日志&＃xff0c;并上传到 HDFS 中

2.2.2 需求分析

2.2.3 实现步骤

1. Flume 要想将数据输出到 HDFS&＃xff0c;须持有 Hadoop 相关 jar 包&＃xff0c;将下列包拷贝到/opt/module/flume-1.9.0/lib下

   commons-configuration2-2.1.1.jar
commons-io-2.5.jar
hadoop-auth-3.1.3.jar
hadoop-common-3.1.3.jar
hadoop-hdfs-3.1.3.jar
htrace-core4-4.1.0-incubating.jar


2. 创建 flume-file-hdfs.conf 文件并添加如下内容&＃xff0c;参考http://flume.apache.org/FlumeUserGuide.html#exec-source

   # Name the components on this agent
a2.sources &＃61; r2
a2.sinks &＃61; k2
a2.channels &＃61; c2# Describe/configure the source
a2.sources.r2.type &＃61; exec
a2.sources.r2.command &＃61; tail -F /opt/module/hive-3.1.2/logs/hive.log# Describe the sink
a2.sinks.k2.type &＃61; hdfs
a2.sinks.k2.hdfs.path &＃61; hdfs://hadoop102:8020/flume/%Y%m%d/%H
#上传文件的前缀
a2.sinks.k2.hdfs.filePrefix &＃61; logs-
#是否按照时间滚动文件夹
a2.sinks.k2.hdfs.round &＃61; true
#多少时间单位创建一个新的文件夹
a2.sinks.k2.hdfs.roundValue &＃61; 1
#重新定义时间单位
a2.sinks.k2.hdfs.roundUnit &＃61; hour
#是否使用本地时间戳
a2.sinks.k2.hdfs.useLocalTimeStamp &＃61; true
#积攒多少个 Event 才 flush 到 HDFS 一次
a2.sinks.k2.hdfs.batchSize &＃61; 100
#设置文件类型&＃xff0c;可支持压缩
a2.sinks.k2.hdfs.fileType &＃61; DataStream
#多久生成一个新的文件
a2.sinks.k2.hdfs.rollInterval &＃61; 60
#设置每个文件的滚动大小
a2.sinks.k2.hdfs.rollSize &＃61; 134217700
#文件的滚动与 Event 数量无关
a2.sinks.k2.hdfs.rollCount &＃61; 0# Use a channel which buffers events in memory
a2.channels.c2.type &＃61; memory
a2.channels.c2.capacity &＃61; 1000
a2.channels.c2.transactionCapacity &＃61; 100# Bind the source and sink to the channel
a2.sources.r2.channels &＃61; c2
a2.sinks.k2.channel &＃61; c2


3. 运行 Flume

   [codecat&＃64;hadoop102 flume-1.9.0]$ flume-ng agent -c conf/ -n a2 -f jobs/flume-file-hdfs.conf


4. 开启 Hive
   

5. 在 HDFS 上查看文件
   

2.3 实时监控目录下多个新文件

2.3.1 案例需求

使用 Flume 监听整个目录的文件(不包含临时文件)&＃xff0c;并上传至 HDFS

2.3.2 需求分析

2.3.3 实现步骤

1. 创建配置文件 flume-dir-hdfs.conf并添加如下内容&＃xff0c;参考&＃xff1a;http://flume.apache.org/FlumeUserGuide.html#spooling-directory-source

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


2. 启动 Flume

   [codecat&＃64;hadoop102 flume-1.9.0]$ flume-ng agent -c conf/ -n a3 -f jobs/flume-dir-hdfs.conf


3. 向 upload 文件夹中添加文件

   • 在/opt/module/flume-1.9.0目录下创建 upload 文件夹

     [codecat&＃64;hadoop102 flume-1.9.0]$ mkdir upload


   • 向 upload 文件夹中添加文件

     [codecat&＃64;hadoop102 upload]$ touch 1.txt
[codecat&＃64;hadoop102 upload]$ touch 1.tmp
[codecat&＃64;hadoop102 upload]$ touch 1.log


4. 查看 HDFS 上的数据
   

5. 再次查询 upload 文件夹
   

2.4 实时监控目录下的多个追加文件

Exec source 适用于监控一个实时追加的文件&＃xff0c;但不能保证数据不丢失&＃xff1b;Spooldir Source 能够保证数据不丢失&＃xff0c;且能够实现断点续传&＃xff0c;但延迟较高&＃xff0c;不能实时监控&＃xff1b;而 Taildir Source 既能够实现断点续传&＃xff0c;又可以保证数据不丢失&＃xff0c;还能够进行实时监控

2.4.1 案例需求

使用 Flume 监听整个目录的实时追加文件&＃xff0c;并上传至 HDFS

2.4.2 需求分析

2.4.3 实现步骤

1. 创建配置文件 flume-taildir-hdfs.conf 并添加如下内容&＃xff0c;参考&＃xff1a;http://flume.apache.org/FlumeUserGuide.html#taildir-source

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


2. 启动 Flume

   [codecat&＃64;hadoop102 flume-1.9.0]$ flume-ng agent -c conf/ -n a3 -f jobs/flume-taildir-hdfs.conf


3. 向 files 文件夹中追加内容

   • 在/opt/module/flume-1.9.0 目录下创建 files文件夹

     [codecat&＃64;hadoop102 flume-1.9.0]$ mkdir files


   • 向 files 文件夹中添加文件
     

4. 查看 HDFS 上的数据