flume消费kafka数据配置

大家好，欢迎来到IT知识分享网。

（1）在hadoop104的/opt/module/flume/conf目录下创建kafka-flume-hdfs.conf文件

[mars@hadoop104 conf]$ vim kafka-flume-hdfs.conf

在文件配置如下内容

## 组件

a1.sources=r1 r2

a1.channels=c1 c2

a1.sinks=k1 k2



## source1
##类型为kafka
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource

#批次拉取的大小
a1.sources.r1.batchSize = 5000
#延迟时间，如不足5000，也进行拉取
a1.sources.r1.batchDurationMillis = 2000

a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092

#topic为"topic_start"
a1.sources.r1.kafka.topics=topic_start



## source2
a1.sources.r2.type = org.apache.flume.source.kafka.KafkaSource


a1.sources.r2.batchSize = 5000

a1.sources.r2.batchDurationMillis = 2000

a1.sources.r2.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092

a1.sources.r2.kafka.topics=topic_event



## channel1

a1.channels.c1.type = file

a1.channels.c1.checkpointDir = /opt/module/flume/checkpoint/behavior1

a1.channels.c1.dataDirs = /opt/module/flume/data/behavior1/

a1.channels.c1.maxFileSize = 2146435071

a1.channels.c1.capacity = 1000000

a1.channels.c1.keep-alive = 6



## channel2

a1.channels.c2.type = file

a1.channels.c2.checkpointDir = /opt/module/flume/checkpoint/behavior2

a1.channels.c2.dataDirs = /opt/module/flume/data/behavior2/

a1.channels.c2.maxFileSize = 2146435071

a1.channels.c2.capacity = 1000000

a1.channels.c2.keep-alive = 6



## sink1 

a1.sinks.k1.type = hdfs

a1.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_start/%Y-%m-%d

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



##sink2

a1.sinks.k2.type = hdfs

a1.sinks.k2.hdfs.path = /origin_data/gmall/log/topic_event/%Y-%m-%d

a1.sinks.k2.hdfs.filePrefix = logevent-



## 不要产生大量小文件（重要）

##滚动时间，生产根据要求进行设置，单位S
a1.sinks.k1.hdfs.rollInterval = 3600

##读取数据大小，达到阀值写入文件，128m
a1.sinks.k1.hdfs.rollSize = 134217728

##读取条数：0为不参考
a1.sinks.k1.hdfs.rollCount = 0



a1.sinks.k2.hdfs.rollInterval = 3600

a1.sinks.k2.hdfs.rollSize = 134217728

a1.sinks.k2.hdfs.rollCount = 0



## 控制输出文件是压缩文件。

a1.sinks.k1.hdfs.fileType = CompressedStream

a1.sinks.k2.hdfs.fileType = CompressedStream



a1.sinks.k1.hdfs.codeC = lzop

a1.sinks.k2.hdfs.codeC = lzop



## 拼装

a1.sources.r1.channels = c1

a1.sinks.k1.channel= c1



a1.sources.r2.channels = c2

a1.sinks.k2.channel= c2