Flume之安装与入门案例分析

Flume入门 1. Flume 安装部署 1.1 安装地址

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

1.2 安装部署

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

   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 监听一个端口，收集该端口数据，并打印到控制台

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，并添加如下内容，参考http://flume.apache.org/FlumeUserGuide.html#a-simple-example

   # 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连接起来
   a1.sources.r1.channels = c1
   # 将c1与k1连接起来
   a1.sinks.k1.channel = c1
   

4. 先开启 flume 监听端口

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

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

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

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

2.2 实时监控单个追加文件 2.2.1 案例需求

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

2.2.2 需求分析

2.2.3 实现步骤

1. Flume 要想将数据输出到 HDFS，须持有 Hadoop 相关 jar 包，将下列包拷贝到/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 文件并添加如下内容，参考http://flume.apache.org/FlumeUserGuide.html#exec-source

   # Name the components on this agent
   a2.sources = r2
   a2.sinks = k2
   a2.channels = c2
   
   # Describe/configure the source
   a2.sources.r2.type = exec
   a2.sources.r2.command = tail -F /opt/module/hive-3.1.2/logs/hive.log
   
   # Describe the sink
   a2.sinks.k2.type = hdfs
   a2.sinks.k2.hdfs.path = hdfs://hadoop102:8020/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
   #多久生成一个新的文件
   a2.sinks.k2.hdfs.rollInterval = 60
   #设置每个文件的滚动大小
   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
   

3. 运行 Flume

   [codecat@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 监听整个目录的文件(不包含临时文件)，并上传至 HDFS

2.3.2 需求分析

2.3.3 实现步骤

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

   a3.sources = r3
   a3.sinks = k3
   a3.channels = c3
   
   # Describe/configure the source
   a3.sources.r3.type = spooldir
   a3.sources.r3.spoolDir = /opt/module/flume-1.9.0/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://hadoop102:8020/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
   

2. 启动 Flume

   [codecat@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@hadoop102 flume-1.9.0]$ mkdir upload
     

   • 向 upload 文件夹中添加文件

     [codecat@hadoop102 upload]$ touch 1.txt
     [codecat@hadoop102 upload]$ touch 1.tmp
     [codecat@hadoop102 upload]$ touch 1.log
     

4. 查看 HDFS 上的数据
   

5. 再次查询 upload 文件夹
   

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

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

2.4.1 案例需求

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

2.4.2 需求分析

2.4.3 实现步骤

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

   a3.sources = r3
   a3.sinks = k3
   a3.channels = c3
   
   # Describe/configure the source
   a3.sources.r3.type = TAILDIR
   a3.sources.r3.positionFile = /opt/module/flume-1.9.0/tail_dir.json
   a3.sources.r3.filegroups = f1
   a3.sources.r3.filegroups.f1 = /opt/module/flume-1.9.0/files/file.*
   
   # Describe the sink
   a3.sinks.k3.type = hdfs
   a3.sinks.k3.hdfs.path = hdfs://hadoop102:8020/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
   

2. 启动 Flume

   [codecat@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@hadoop102 flume-1.9.0]$ mkdir files
     

   • 向 files 文件夹中添加文件
     

4. 查看 HDFS 上的数据