04_ Flume采集文件到HDFS案例

时间：2021-11-03 10:39:08

采集需求：比如业务系统使用log4j生成的日志，日志内容不断增加，需要把追加到日志文件中的数据实时采集到hdfs

根据需求，首先定义以下3大要素

　　采集源，即source——监控文件内容更新 :  exec  ‘tail -F file’

　　下沉目标，即sink——HDFS文件系统  :  hdfs sink

　　Source和sink之间的传递通道——channel，可用file channel 也可以用内存channel

1.配置采集方案：

# Name the components on this agent

a1.sources = r1

a1.sinks = k1

a1.channels = c1

#exec 指的是命令

# Describe/configure the source

a1.sources.r1.type = exec

#F根据文件名追中, f根据文件的nodeid追中
#mkdir /home/hadoop/log   touch /home/hadoop/log/test.log

a1.sources.r1.command = tail -F /home/hadoop/log/test.log

a1.sources.r1.channels = c1

# Describe the sink

#下沉目标

a1.sinks.k1.type = hdfs

a1.sinks.k1.channel = c1

#指定目录, flum帮做目的替换

a1.sinks.k1.hdfs.path = /flume2/events/%y-%m-%d/%H%M/

#文件的命名, 前缀

a1.sinks.k1.hdfs.filePrefix = events-

#10 分钟就改目录

a1.sinks.k1.hdfs.round = true

a1.sinks.k1.hdfs.roundValue = 10

a1.sinks.k1.hdfs.roundUnit = minute

#文件滚动之前的等待时间(秒)

a1.sinks.k1.hdfs.rollInterval = 3

#文件滚动的大小限制(bytes)

a1.sinks.k1.hdfs.rollSize = 500

#写入多少个event数据后滚动文件(事件个数)

a1.sinks.k1.hdfs.rollCount = 20

#5个事件就往里面写入

a1.sinks.k1.hdfs.batchSize = 5

#用本地时间格式化目录

a1.sinks.k1.hdfs.useLocalTimeStamp = true

#下沉后, 生成的文件类型，默认是Sequencefile，可用DataStream，则为普通文本

a1.sinks.k1.hdfs.fileType = DataStream

# 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

启动命令：

　　bin/flume-ng agent -c conf -f conf/tail-hdfs.conf -n a1

不断向文件中写入日志，并采集到hdfs中, hdfs文件中的目录不用自己建的

04_ Flume采集文件到HDFS案例

前端页面查看下, shizhan2:50070, 文件目录: /flum/events...

04_ Flume采集文件到HDFS案例

Flume支持众多的source和sink类型，详细手册可参考官方文档:http://flume.apache.org/FlumeUserGuide.html

　　

相关文章