flume hdfs 大数据 数据采集

以下是关于一个agent的exec-hdfs.properties采集配置文件

# 定义这个agent中各组件的名字

a1.sources = r1

a1.sinks = k1

a1.channels = c1

#【Source】

a1.sources.r1.type = exec

a1.sources.r1.command = tail -F /root/flumespool/access.log

a1.sources.r1.shell = /bin/bash -c

#【Channel】

#表示a1的channel类型是memory内存型

a1.channels.c1.type = memory

#表示通道channel中能够存储的最大事件数

a1.channels.c1.capacity = 1000

#表示每次从source获取的事件数或给到sink的事件数

a1.channels.c1.transactionCapacity = 100

#【Sink】

# Describe the sink

a1.sinks.k1.type = hdfs

a1.sinks.k1.hdfs.path = hdfs://mini1:9000/flume/%Y%m%d/%H

#上传文件的前缀

a1.sinks.k1.hdfs.filePrefix = daoke360.events.%M

#上传文件的后缀

a1.sinks.k1.hdfs.fileSuffix=.lzo

#设置文件类型，DataStream不会压缩输出文件，默认为SequenceFile

#当使用DataStream时候，文件不会被压缩，不需要设置hdfs.codeC;

#a1.sinks.k1.hdfs.fileType = DataStream

#当使用CompressedStream时候，文件会被压缩，需要设置hdfs.codeC;

a1.sinks.k1.hdfs.fileType = CompressedStream

#文件压缩格式，包括：gzip, bzip2, lzo, lzop, snAppy

a1.sinks.k1.hdfs.codeC = lzo

#writeFormat 序列文件记录的格式。在使用Flume创建数据文件之前设置为Text，否则 Apache Impala 或Apache Hive无法读取这些文件。

#a1.sinks.k1.hdfs.fileType = SequenceFile

#写sequence文件的格式。包含：Text, Writable（默认）

#a1.sinks.k1.hdfs.writeFormat = Text

#指定每个HDFS块的最小副本数。如果未指定，则它来自类路径中的默认Hadoop配置。

a1.sinks.k1.hdfs.minBlockReplicas = 1

#积攒多少个Event才flush到HDFS一次

a1.sinks.k1.hdfs.batchSize = 100

#设置每个文件的滚动大小，单位byte

#rollsize的配置表示到2G大小的时候回滚到下一个文件，也就是到了这个时间 hdfs就会rename正在写的文件到已经写完

a1.sinks.k1.hdfs.rollSize = 2048000000

#文件的滚动与Event数量无关

a1.sinks.k1.hdfs.rollCount = 0

#多久生成一个新的文件,单位秒,这里配置的是每个6小时回滚到下一个文件

#这个值不能设置为0，否则flume在连接hdfs时会报超时连接的error，导致程序反复在重连

a1.sinks.k1.hdfs.rollInterval = 21600

#是否应舍入时间戳

a1.sinks.k1.hdfs.round = true

#舍入时间戳为小时，有三个值可配置(second, minute , hour)

a1.sinks.k1.hdfs.roundUnit = hour

#按1小时进行舍入

a1.sinks.k1.hdfs.roundValue = 1

#是否使用本地时间戳

a1.sinks.k1.hdfs.useLocalTimeStamp = true

#当目前被打开的临时文件在该参数指定的时间（秒）内，没有任何数据写入，则将该临时文件关闭并重命名成目标文件；默认值：0

a1.sinks.k1.hdfs.idleTimeout = 0

#执行HDFS操作的超时时间（单位：毫秒），默认值：10000

a1.sinks.k1.hdfs.callTimeout = 10000

#【source channel sink】

a1.sources.r1.channels = c1

a1.sinks.k1.channel = c1