Flume+Kafka整合

腳本生產數據---->flume採集數據----->kafka消費數據------->storm集羣處理數據

 

日誌文件使用log4j生成，滾動生成！

 

當前正在寫入的文件在知足必定的數量閾值以後，須要重命名！！！ 

 

flume+Kafka整合步驟及相關配置：（先安裝好zookeeper集羣和Kafka集羣）

配置flume：

一、下載flume

二、解壓flume安裝包

　　cd  /export/servers/

　　tar  -zxvf  apache-flume-1.6.0-bin.tar.gz

　　ln  -s  apache-flume-1.6.0-bin  flume

三、建立flume配置文件

　　cd  /export/servers/flume/conf/

　　mkdir  myconf

　　vi  exec.conf

　　輸入一下內容：

　　a1.sources=r1

　　a1.channels=c1

　　a1.sinks=k1

 

　　a1.sources.r1.type=exec

　　a1.sources.r1.command=tail -F /export/data/flume_sources/click_log/1.log

　　a1.sources.r1.channels=c1

 

　　a1.channels.c1.type=memory

　　a1.channels.c1.capacity=10000

　　a1.channels.c1.transactionCapacity=100

 

　　a1.sinks.k1.type=org.apache.flume.sink.kafka.KafkaSink

　　a1.sinks.k1.topic=test

　　a1.sinks.k1.brokerList=kafka01:9092

　　a1.sinks.k1.requiredAcks=1

　　a1.sinks.k1.batchSize=20

　　a1.sinks.k1.channel=c1

四、準備目標數據的目錄

　　mkdir  -p  /export/data/flume_sources/click_log

五、經過腳本建立目標文件並生產數據

　　for((i=0;i<=50000;i++));

　　do  echo "message-" + $i >> /export/data/flume_sources/click_log/1.log;

　　done

注：腳本名稱爲click_log_out.sh，須要使用root用戶賦權，chmod  +x  click_log_out.sh

六、開始打通全部流程

　　一：啓動Kafka集羣

　　　　kafka-server-start.sh  /export/servers/kafka/config/server.properties

　　二：建立一個topic並開啓consumer

　　　　kafka-console-consumer.sh  --topic=test  --zookeeper  zk01:2181

　　三：執行數據生產的腳本

　　　　sh  click_log_out.sh

　　四：啓動flume客戶端

　　　　./bin/flume_ng  agent  -n  a1  -c  conf  -f  conf/myconf/exec.conf  -Dflume.root.logger=INFO,console

　　五：在第三步啓動的kafka consumer窗口查看效果