flume框架案例之抽取文件

a3.sources = r3

a3.sinks = k3

a3.channels = c3

# Describe/configure the source

# 源是某个目录使用spooldir

a3.sources.r3.type = spooldir

# 抽取的目录 $ cp -r ${HADOOP_HOME}/logs/* /home/user01/logs ->执行完之后，会在原文件后面自动加一个.COMPLETED，所以换个目录比较合适

a3.sources.r3.spoolDir = /home/user01/logs

# 抽取该目录下符合包含.log结尾的文件 用正则表达式去除不符合要求的文件a3.sources.r3.ignorePattern = ^.*out.*

# 设置channel类型是file

a3.channels.c3.type = file

# 设置检查点目录，记录已经获取哪些文件，一些元数据信息

a3.channels.c3.checkpointDir = /opt/modules/cdh/flume-1.5.0-cdh5.3.6/checkpoint

#设置缓存的数据存储目录

a3.channels.c3.dataDirs = /opt/modules/cdh/flume-1.5.0-cdh5.3.6/bufferdata

# 设置sinks的输出类型

a3.sinks.k3.type = hdfs

# 启用设置多级目录，这里按年/月/日/时 2级目录，每个小时生成一个文件夹

a3.sinks.k3.hdfs.path = hdfs://bigdata.ibeifeng.com:8020/flume2/%Y%m%d/%H

# 设置HDFS生成文件的的前缀

a3.sinks.k3.hdfs.filePrefix = accesslog

#启用按时间生成文件夹

a3.sinks.k3.hdfs.round = true

#设置round单位:小时

a3.sinks.k3.hdfs.roundValue = 1

a3.sinks.k3.hdfs.roundUnit = hour

#使用本地时间戳

a3.sinks.k3.hdfs.useLocalTimeStamp = true

# 设置每次写入的DFS的event的个数为100个

a3.sinks.k3.hdfs.batchSize = 100

# 写入HDFS的方式

a3.sinks.k3.hdfs.fileType = DataStream

# 写入HDFS的文件格式

a3.sinks.k3.hdfs.writeFormat = Text

#设置解决文件过多过小问题

#每600秒生成一个文件

a3.sinks.k3.hdfs.rollInterval = 60

#当达到128000000bytes时，创建新文件 127*1024*1024

#实际环境中如果按照128M回顾文件,那么这里设置一般设置成127M

a3.sinks.k3.hdfs.rollSize = 128000000

#设置文件的生成不和events数相关，与时间和大小相关

a3.sinks.k3.hdfs.rollCount = 0

#设置成1，否则当有副本复制时就重新生成文件，上面三条则没有效果

a3.sinks.k3.hdfs.minBlockReplicas =1

//channel 连接sources和sinks

# Bind the source and sink to the channel

a3.sources.r3.channels = c3

a3.sinks.k3.channel = c3

===================================

2.启动Flume-agent a3

$ bin/flume-ng agent --conf conf --conf-file conf/a3.conf --name a3 -Dflume.root.logger=INFO,console

bin/flume-ng agent --conf conf --conf-file conf/a3.conf --name a3 -Dflume.root.logger=INFO,console