flume是一個實時消息收集系統，它定義了多種的source、channel、sink，可以根據實際情況選擇。
Flume下載下傳及文檔：
http://flume.apache.org/
Kafka
kafka是一種高吞吐量的分布式釋出訂閱消息系統，她有如下特性：
通過O(1)的磁盤資料結構提供消息的持久化，這種結構對于即使數以TB的消息存儲也能夠保持長時間的穩定性能。
高吞吐量：即使是非常普通的硬體kafka也可以支援每秒數十萬的消息。
支援通過kafka伺服器和消費機叢集來分區消息。
支援Hadoop并行資料加載。
kafka的目的是提供一個釋出訂閱解決方案，它可以處理消費者規模的網站中的所有動作流資料。 這種動作（網頁浏覽，搜尋和其他使用者的行動）是在現代網絡上的許多社會功能的一個關鍵因素。 這些資料通常是由于吞吐量的要求而通過處理日志和日志聚合來解決。 對于像Hadoop的一樣的日志資料和離線分析系統，但又要求實時處理的限制，這是一個可行的解決方案。kafka的目的是通過Hadoop的并行加載機制來統一線上和離線的消息處理，也是為了通過叢集機來提供實時的消費。
kafka分布式訂閱架構如下圖：--取自Kafka官網
配置kafka的配置檔案 server.properties ，其它可根據自己的情況修改。
啟動kafka，啟動之前先啟動zookeeper，zookeeper的配置不再叙述。
# bin/kafka-server-start.sh config/server.properties
Create a topic
# bin/kafka-topics.sh --create --zookeeper localhost:2181 ----replication-factor 1 --partitions 1 --topic test
檢視topic
# bin/kafka-topics.sh --list --zookeeper localhost:2181
測試是否能正常生産消費；驗證流程正确性
# bin/kafka-console-producer.sh--broker-list localhost:9092 --topic test
# bin/kafka-console-consumer.sh--zookeeper localhost:2181 --topic test --from-beginning
接下來是架構之間的整合
flume和kafka整合
1.下載下傳flume-kafka-plus:https://github.com/beyondj2ee/flumeng-kafka-plugin
2.提取插件中的flume-conf.properties檔案
修改該檔案：#source section
producer.sources.s.type = exec
producer.sources.s.command = tail -f -n+1 /mnt/hgfs/vmshare/test.log
producer.sources.s.channels = c
修改所有topic的值改為test
将改後的配置檔案放進flume/conf目錄下
在該項目中提取以下jar包放入環境中flume的lib下：
還有package目錄的flumeng-kafka-plugin.jar包一并放到flume的lib目錄下。
附上flume的配置檔案
############################################
# producer config
###########################################
#agent section
producer.sources = s
producer.channels = c
producer.sinks = r
#source section
producer.sources.s.channels = c
producer.sources.s.command = tail -f /var/log/messages
#producer.sources.s.type=spooldir
#producer.sources.s.spoolDir=/home/xiaojie.li
#producer.sources.s.fileHeader=false
#producer.sources.s.type=syslogtcp
#producer.sources.s.port=5140
#producer.sources.s.host=localhost
# Each sink's type must be defined
producer.sinks.r.type = org.apache.flume.plugins.KafkaSink
producer.sinks.r.metadata.broker.list=10.10.10.127:9092
producer.sinks.r.zk.connect=10.10.10.127:2181
producer.sinks.r.partition.key=0
producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartition
producer.sinks.r.serializer.class=kafka.serializer.StringEncoder
producer.sinks.r.request.required.acks=0
producer.sinks.r.max.message.size=1000000
producer.sinks.r.producer.type=sync
producer.sinks.r.custom.encoding=UTF-8
producer.sinks.r.custom.topic.name=test
#Specify the channel the sink should use
producer.sinks.r.channel = c
# Each channel's type is defined.
producer.channels.c.type = memory
producer.channels.c.capacity = 1000
producer.channels.c.transactionCapacity=100
#producer.channels.c.type=file
#producer.channels.c.checkpointDir=/home/checkdir
#producer.channels.c.dataDirs=/home/datadir
驗證flume和kafka組合
前面kafka已經啟動，這裡直接啟動flume
# bin/flume-ng agent -c conf -f conf/master.properties -n producer -Dflume.root.logger=INFO,console
使用kafka的kafka-console-consumer.sh腳本檢視是否有flume有沒有往Kafka傳輸資料；
可以看到tail /var/log/messages已經通過flume傳到kafka裡，說明flume+kafka組合已經成功了。
日志最終需要儲存在hdfs裡
還需要自己開發插件去實作，這裡不再多說。