flume從Kafka消費數據到HDFS

#source的名字
agent.sources = kafkaSource
# channels的名字，建議按照type來命名
agent.channels = memoryChannel
# sink的名字，建議按照目標來命名
agent.sinks = hdfsSink

# 指定source使用的channel名字
agent.sources.kafkaSource.channels = memoryChannel
# 指定sink需要使用的channel的名字,注意這里是channel
agent.sinks.hdfsSink.channel = memoryChannel

#-------- kafkaSource相關配置-----------------
# 定義消息源類型
agent.sources.kafkaSource.type = org.apache.flume.source.kafka.KafkaSource
# 定義kafka所在zk的地址
#
# 這里特別注意: 是kafka的zookeeper的地址
#
agent.sources.kafkaSource.zookeeperConnect = 10.255.29.30:2181
# 配置消費的kafka topic
agent.sources.kafkaSource.topic = topic_start
# 配置消費者組的id，不能跟spark streaming中group id一樣
agent.sources.kafkaSource.groupId = kafka2hdfs
# 消費超時時間,參照如下寫法可以配置其他所有kafka的consumer選項。注意格式從kafka.xxx開始是consumer的配置屬性
agent.sources.kafkaSource.kafka.consumer.timeout.ms = 100

#------- memoryChannel相關配置-------------------------
# channel類型
agent.channels.memoryChannel.type = memory
# channel存儲的事件容量
agent.channels.memoryChannel.capacity=10000
# 事務容量
agent.channels.memoryChannel.transactionCapacity=1000

#---------hdfsSink 相關配置------------------
agent.sinks.hdfsSink.type = hdfs
# 注意, 我們輸出到下面一個子文件夾datax中
agent.sinks.hdfsSink.hdfs.path = hdfs://app02:8020/tmp/mes/%Y%m%d%H
agent.sinks.hdfsSink.hdfs.writeFormat = Text
agent.sinks.hdfsSink.hdfs.fileType = DataStream
#為了防止小文件
agent.sinks.hdfsSink.hdfs.rollSize = 0
agent.sinks.hdfsSink.hdfs.rollCount = 0
#agent.sinks.hdfsSink.hdfs.rollInterval = 60

#配置前綴和后綴
agent.sinks.hdfsSink.hdfs.filePrefix=test
agent.sinks.hdfsSink.hdfs.fileSuffix=.data

#避免文件在關閉前使用臨時文件
agent.sinks.hdfsSink.hdfs.inUserPrefix=_
agent.sinks.hdfsSink.hdfs.inUserSuffix=

#自定義攔截器
#agent.sources.kafkaSource.interceptors=i1
#agent.sources.kafkaSource.interceptors.i1.type=com.hadoop.flume.FormatInterceptor$Builder

#為了防止小文件
agent.sinks.hdfsSink.hdfs.rollSize = 0
agent.sinks.hdfsSink.hdfs.rollCount = 0
#agent.sinks.hdfsSink.hdfs.rollInterval = 60
但是只在一定程度上減少了小文件，跟kafak數據量大小有關

所以需要將這些小文件進行歸檔。可以同時和spark streaming一起消費同一批數據