二、Flume 实例（一）

系统管理员 2023-06-01 14:58 91阅读 0赞

**目录**

一、监控端口数据

二、实时读取本地文件到 HDFS 案例

三、实时读取目录文件到 HDFS 案例

--------------------

## 一、监控端口数据 ##

案例需求：

*  首先， Flume 监控本机 44444 端口
 *  然后通过 telnet 工具向本机 44444 端口发送消息
 *  最后 Flume 将监听的数据实时显示在控制台

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70][]

**1、创建该案例的配置文件**

mkdir /opt/software/flume/job
    touch netcat-flume-logger.conf
    vim vim netcat-flume-logger.conf

**2、配置该案例的配置文件**

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    # Describe/configure the source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    # Describe the sink
    a1.sinks.k1.type = logger
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

**3、配置选项解析**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1][]

**4、先开启 flume 监听端口**

/opt/software/bin/flume-ng agent -c conf/ -f /opt/software/job/netcat-flume-logger.conf -n 
    a1 Dflume.root.logger=INFO,console

**参数说明：**

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

**--name a1 ：** 表示给 agent 起名为 a1

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

**-Dflume.root.logger==INFO,console ：** \-D 表示 flume 运 行 时 动 态 修 改flume.root.logger 参数属性值，并将控制台日志打印级别设置为 INFO 级别。日志级别包括:log、info、 warn、 error。

**5、向本机的 44444 端口发送内容**

nc localhost 44444

之后再开启的 nc 客户端向 Flume 发送数据，即可显示在 flume 中

## 二、实时读取本地文件到 HDFS 案例 ##

案例需求： 实时监控 Hive 日志， 并上传到 HDFS 中

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 2][]

**1、Flume 要想将数据输出到 HDFS， 必须持有 Hadoop 相关 jar 包**

commons-configuration-1.6.jar
    hadoop-auth-2.7.2.jar
    hadoop-common-2.7.2.jar
    hadoop-hdfs-2.7.2.jar 
    commons-io-2.4.jar # 1.99 版本 flume 必须引用的 jar
    htrace-core-3.1.0-incubating.jar # 1.99 版本 flume 必须引用的 jar

拷贝到 /opt/software/flume/lib 文件夹下。

**2、创建 flume-file-hdfs.conf 文件**

cd /opt/software/flume/job/
    touch flume-file-hdfs.conf

注： 要想读取 Linux 系统中的文件， 就得按照 Linux 命令的规则执行命令。 由于 hive日志在 Linux 系统中，所以读取文件的类型选择： **exec 即 execute 执行的意思。 表示执行 Linux命令来读取文件。**

**3、配置该案例的配置文件**

# 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/logs/hive.log
    a2.sources.r2.shell = /bin/bash -c
    # Describe the sink
    a2.sinks.k2.type = hdfs
    a2.sinks.k2.hdfs.path = hdfs://hadoop102:9000/flume/local_to_hdfs/
    #上传文件的前缀
    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 = 1000
    #设置文件类型， 可支持压缩
    a2.sinks.k2.hdfs.fileType = DataStream
    #多久生成一个新的文件
    a2.sinks.k2.hdfs.rollInterval = 600
    #设置每个文件的滚动大小
    a2.sinks.k2.hdfs.rollSize = 134217700
    #文件的滚动与 Event 数量无关
    a2.sinks.k2.hdfs.rollCount = 0
    #最小冗余数
    a2.sinks.k2.hdfs.minBlockReplicas = 1
    # 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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 3][]

**4、执行监控配置**

bin/flume-ng agent --conf conf/ --name a2 --conf-file job/flume-file-hdfs.conf

**5、在指定的sink目录查看是否有生成的文件**

hdfs dfs -ls hdfs://hadoop102:9000/flume/local_to_hdfs/

## 三、实时读取目录文件到 HDFS 案例 ##

案例需求： 使用 flume 监听整个目录的文件

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 4][]

1、创建配置文件 flume-dir-hdfs.conf

touch flume-dir-hdfs.conf
    vim flume-dir-hdfs.conf

2、**配置该案例的配置文件**

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/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: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 = 600
    #设置每个文件的滚动大小大概是 128M
    a3.sinks.k3.hdfs.rollSize = 134217700
    #文件的滚动与 Event 数量无关
    a3.sinks.k3.hdfs.rollCount = 0
    #最小冗余数
    a3.sinks.k3.hdfs.minBlockReplicas = 1
    # 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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 5][]

**3、启动监控文件夹命令**

bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-dir-hdfs.conf

**注意：**在使用 Spooling Directory Source 时

1) 不要在监控目录中创建并持续修改文件

2) 上传完成的文件会以.COMPLETED 结尾

3) 被监控文件夹每 600 毫秒扫描一次文件变动

**4、向 upload 文件夹中添加文件**

在/opt/software/flume 目录下创建 upload 文件夹

mkdir upload
    touch atguigu.txt
    touch atguigu.tmp
    touch atguigu.log

**5、使用 hdfs dfs -ls dir 命令查看 HDFS 上的数据**

**6、等待 1s， 再次查询 upload 文件夹，使用 ls 或 ll 命令**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70]: /images/20230601/c30cf1d3482943bda68bc5c21e4b979e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1]: /images/20230601/c730fb160432449a9bbbd2d0513211b7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 2]: /images/20230601/e096604a656948278119db0f3b927444.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 3]: /images/20230601/8b0d4fba210247a59985188ebf8d51c3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 4]: /images/20230601/2bedbe5dde80436d91091e03a2dde02e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 5]: /images/20230601/91a7bb65f4d74cc293dd9ab46cba921e.png