flume监控文件sink到hdfs

flume监控文件sink到hdfs

启动命令建议全路径

/home/mjxt/apache-flume-1.9.0-bin/bin/flume-ng agent -n a1 -f /home/mjxt/test_flume/flume-conf.properties -Dflume.root.logger=DEBUG,Console

配置如下

# 定义agent
a1.sources = s1
a1.channels = c1
a1.sinks = k1
# sources类型
a1.sources.s1.type  =  taildir
#存储读取文件数据最后位置
a1.sources.s1.positionFile  = /home/mjxt/test_flume/taildir_position.json
a1.sources.s1.filegroups  =  g1
# 监控目录 需要先创建
a1.sources.s1.filegroups.g1  = /home/mjxt/test_flume/txt/001.*.log
# sink到哪儿
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# hdfs的输出路径
a1.sinks.k1.hdfs.path = hdfs:///tmp/flume_test/
# 指定输出到hdfs之后的文件格式
a1.sinks.k1.hdfs.filePrefix = /%Y%m%d/log
a1.sinks.k1.hdfs.fileSuffix = .log
a1.sinks.k1.hdfs.minBlockReplicas = 1
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.rollSize = 0
a1.sinks.k1.hdfs.rollCount = 100
a1.sinks.k1.hdfs.idleTimeout = 0
# 设置通道channels
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /home/mjxt/test_flume/
a1.channels.c1.dataDirs =  /home/mjxt/test_flume/data
# 关联
a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1

监控目录的路径是代表是监控所有带001.log的文件

最终sink的路径是 /tmp/flume_test/