一、需求

• 利用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