一、需求
- 利用flume采集Linux下的文件信息，并且传入到kafka集群当中。
- 环境准备zookeeper集群和kafka集群安装好。
二、配置flume
- 官网下载flume。博主自己这里使用的是flume1.6.0。
- 官网地址http://flume.apache.org/download.html
- 解压缩。tar -zxvf apache-flume-1.6.0-bin.tar.gz -C /usr/apps/
- 创建flume配置文件。
- cd /usr/apps/flume/apache-flume-1.6.0-bin/conf
- vim exec.conf flume配置文件如下
?1 a1.sources = r1 ?2 a1.channels = c1 ?3 a1.sinks = k1 ?4 a1.sources.r1.type = exec ?5 a1.sources.r1.command = tail -F /usr/test/click_log/1.log ?6 a1.sources.r1.channels = c1 ?7 a1.channels.c1.type=memory ?8 a1.channels.c1.capacity=10000 ?9 a1.channels.c1.transactionCapacity=100 10 #固定的不需要修改 11 a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink 12 #消息的主题 13 a1.sinks.k1.topic = logMq 14 #kafka broker集群 15 a1.sinks.k1.brokerList = bhz136:9092,bhz137:9092,bhz139:9092 16 a1.sinks.k1.requiredAcks = 1 17 a1.sinks.k1.batchSize = 20 18 a1.sinks.k1.channel = c1
三、启动flume连接到kafka
- ./bin/flume-ng agent -n a1 -c conf -f conf/exec.conf -Dflume.root.logger=INFO,console
四、启动kafka消费者消费数据
- kafka-console-consumer.sh --zookeeper bhz136:2181 --from-beginning --topic logMq
五、运行测试程序。测试结果如下：
注：1.log文件是通过以下脚本生成的测试文件
?1 for((i=0;i<=50000;i++)); ?2 ????????do echo "message-"+$i >>/usr/test/click_log/1.log; ?3 done
flume整合kafka
原文地址：https://www.cnblogs.com/gemine/p/9127431.html