说明

监控一个目录的文件，当文件发生变化是，将内容发送到 HDFS

检查 hadoop 环境

Flume 将数据输出到 HDFS，依赖 Hadoop 相关 jar 包

如果 flume 与 hadoop 安装在一台服务器上，集成较为简单

检查 /etc/profile.d/bigdata_env.sh 文件，确认 Hadoop 和 Java 环境变量配置正确

# 配置JAVA_HOME
export JAVA_HOME=/program/jdk1.8.0_202
export PATH=${JAVA_HOME}/bin:$PATH

# 配置 HADOOP_HOME
export HADOOP_HOME=/program/hadoop-3.0.3
export PATH=${HADOOP_HOME}/bin:$PATH
export PATH=${HADOOP_HOME}/sbin:$PATH

Taildir Source

参见：
https://www.malaoshi.top/show_1IX24u2g3aa2.html

HDFS

官方文档：
http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#hdfs-sink

其他文档：
https://flume.liyifeng.org/#hdfs-sink

关键参数解释：

a1.sinks.k1.type = hdfs

# HDFS保存文件目录，由年月日组成
a1.sinks.k1.hdfs.path = hdfs://hadoop1:8020/flume/logs/%Y%m%d
# 路径配置时间后，此处必须为true
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# 不压缩
a1.sinks.k1.hdfs.fileType = DataStream
# 文件后缀
a1.sinks.k1.hdfs.fileSuffix = .log
# 间隔 30s 后，如果有数据，就新建文件
a1.sinks.k1.hdfs.rollInterval = 180
# 如果写入内容达到120m（要低于hadoop的128m），就新建文件
a1.sinks.k1.hdfs.rollSize = 125829120

创建测试目录

/test_flume

编写配置文件

cd /program/flume-1.9.0/conf

vim test_hdfs.conf

内容如下：

# 定义agent名字为a1，相当于变量名，可任意起
# 设置 sources、sinks、channels的名字
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置source类型为taildir
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1 f2

# 监控该文件
a1.sources.r1.filegroups.f1 = /test_flume/1.log
# 监控该目录下所有 .log 后缀的文件
a1.sources.r1.filegroups.f2 = /test_flume/logs/.*log

# 配置 sink 
a1.sinks.k1.type = hdfs
# HDFS保存文件目录，由年月日组成
a1.sinks.k1.hdfs.path = hdfs://hadoop1:8020/flume/logs/%Y%m%d/
# 文件夹配置时间后，此处必须为true
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# 不压缩
a1.sinks.k1.hdfs.fileType = DataStream
# 文件后缀
a1.sinks.k1.hdfs.fileSuffix = .log
# 间隔 60s 后，如果有数据，就新建文件
a1.sinks.k1.hdfs.rollInterval = 180
# 如果写入内容达到120m（要低于hadoop的128m），就新建文件
a1.sinks.k1.hdfs.rollSize = 125829120


# 配置 channel 类型为内存
a1.channels.c1.type = memory

# 内存队列最大为1000，可以存储1000个event
a1.channels.c1.capacity = 1000

# 一个事务中从source接收的events数量或发送给sink的events数量最大为100
a1.channels.c1.transactionCapacity = 100

# 将source 和 sink 绑定到channel上
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

启动

进入 flume 根目录：

cd /program/flume-1.9.0/

执行下面命令（必须在上面目录中才能执行）：

bin/flume-ng agent -n a1 -c conf -f conf/test_hdfs.conf -Dflume.root.logger=INFO,console

解释：
https://www.malaoshi.top/show_1IX258AKgb2F.html

执行结果

会打印一大堆信息，看到下面字样，表示启动成功：

测试：监控 1.log 文件

修改 /test_flume/1.log 文件：

进入目录：

cd /test_flume

执行下面命令：

echo lucy >> 1.log

注意： 如果使用 vim ，文件权限的 other 要有 w权限，否则保存后，inode 值会改变

控制台打印

2022-11-21 13:05:00,134 (PollableSourceRunner-TaildirSource-r1) [INFO - org.apache.flume.source.taildir.ReliableTaildirEventReader.openFile(ReliableTaildirEventReader.java:290)] Opening file: /test_flume/logs/a.log, inode: 51427501, pos: 220
2022-11-21 13:05:00,134 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.hdfs.HDFSDataStream.configure(HDFSDataStream.java:57)] Serializer = TEXT, UseRawLocalFileSystem = false
2022-11-21 13:05:00,143 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.hdfs.BucketWriter.open(BucketWriter.java:246)] Creating hdfs://hadoop1:8020/flume/logs/20221121//FlumeData.1669007100135.log.tmp

说明 flume 成功监控到 1.log 文件内容的编号

查看 HDFS 文件

访问：http://hadoop1:9870/explorer.html#/flume/logs ，可以查看 根据当前年月日创建的目录

进入该目录，可以看到创建的文件

原文出处：/show_1IX258Tc7AHi.html

说明