一直以來都想接觸Storm實時計算這塊的東西，最近在群裡看到上海一哥們羅寶寫的Flume+Kafka+Storm的實時日志流系統的搭建文檔，自己也跟着整了一遍，之前羅寶的文章中有一些要注意點沒提到的，以後一些寫錯的點，在這邊我會做修正；内容應該說絕大部分引用羅寶的文章的，這裡要謝謝羅寶兄弟，還有寫這篇文章@晨色星空J2EE也給了我很大幫助，這裡也謝謝@晨色星空J2EE
之前在弄這個的時候，跟群裡的一些人讨論過，有的人說，直接用storm不就可以做實時處理了，用不着那麼麻煩；其實不然，做軟體開發的都知道子產品化思想，這樣設計的原因有兩方面：
一方面是可以子產品化，功能劃分更加清晰，從“資料采集--資料接入--流失計算--資料輸出/存儲”
1）.資料采集
負責從各節點上實時采集資料，選用cloudera的flume來實作
2）.資料接入
由于采集資料的速度和資料處理的速度不一定同步，是以添加一個消息中間件來作為緩沖，選用apache的kafka
3）.流式計算
對采集到的資料進行實時分析，選用apache的storm
4）.資料輸出
對分析後的結果持久化，暫定用mysql
另一方面是子產品化之後，加入當Storm挂掉了之後，資料采集和資料接入還是繼續在跑着，資料不會丢失，storm起來之後可以繼續進行流式計算；
那麼接下來我們來看下整體的架構圖
詳細介紹各個元件及安裝配置：
作業系統：ubuntu
Flume
Flume是Cloudera提供的一個分布式、可靠、和高可用的海量日志采集、聚合和傳輸的日志收集系統，支援在日志系統中定制各類資料發送方，用于收集資料;同時，Flume提供對資料進行簡單處理，并寫到各種資料接受方(可定制)的能力。
下圖為flume典型的體系結構：
Flume資料源以及輸出方式:
Flume提供了從console(控制台)、RPC(Thrift-RPC)、text(檔案)、tail(UNIX
tail)、syslog(syslog日志系統，支援TCP和UDP等2種模式)，exec(指令執行)等資料源上收集資料的能力,在我們的系統中目前使用exec方式進行日志采集。
Flume的資料接受方，可以是console(控制台)、text(檔案)、dfs(HDFS檔案)、RPC(Thrift-RPC)和syslogTCP(TCP
syslog日志系統)等。在我們系統中由kafka來接收。
Flume下載下傳及文檔：
<a href="http://flume.apache.org/">http://flume.apache.org/</a>
Flume安裝：
$tar zxvf apache-flume-1.4.0-bin.tar.gz/usr/local
Flume啟動指令：
$bin/flume-ng agent --conf conf --conf-file conf/flume-conf.properties --name producer -Dflume.root.logger=INFO,console
Kafka
kafka是一種高吞吐量的分布式釋出訂閱消息系統，她有如下特性：
通過O(1)的磁盤資料結構提供消息的持久化，這種結構對于即使數以TB的消息存儲也能夠保持長時間的穩定性能。
高吞吐量：即使是非常普通的硬體kafka也可以支援每秒數十萬的消息。
支援通過kafka伺服器和消費機叢集來分區消息。
支援Hadoop并行資料加載。
kafka的目的是提供一個釋出訂閱解決方案，它可以處理消費者規模的網站中的所有動作流資料。 這種動作（網頁浏覽，搜尋和其他使用者的行動）是在現代網絡上的許多社會功能的一個關鍵因素。 這些資料通常是由于吞吐量的要求而通過處理日志和日志聚合來解決。 對于像Hadoop的一樣的日志資料和離線分析系統，但又要求實時處理的限制，這是一個可行的解決方案。kafka的目的是通過Hadoop的并行加載機制來統一線上和離線的消息處理，也是為了通過叢集機來提供實時的消費。
kafka分布式訂閱架構如下圖：--取自Kafka官網
羅寶兄弟文章上的架構圖是這樣的
其實兩者沒有太大差別，官網的架構圖隻是把Kafka簡潔的表示成一個Kafka Cluster，而羅寶兄弟的架構圖就相對詳細一些；
Kafka版本：0.8.0
Kafka下載下傳及文檔：http://kafka.apache.org/
Kafka安裝：
> tar xzf kafka-<VERSION>.tgz
> cd kafka-<VERSION>
> ./sbt update
> ./sbt package
> ./sbt assembly-package-dependency
啟動及測試指令：
（1） start server
> bin/zookeeper-server-start.shconfig/zookeeper.properties
> bin/kafka-server-start.shconfig/server.properties
這裡是官網上的教程，kafka本身有内置zookeeper，但是我自己在實際部署中是使用單獨的zookeeper叢集，是以第一行指令我就沒執行，這裡隻是些出來給大家看下。
配置獨立的zookeeper叢集需要配置server.properties檔案，講zookeeper.connect修改為獨立叢集的IP和端口
zookeeper.connect=nutch1:2181
（2）Create a topic
> bin/kafka-create-topic.sh --zookeeper localhost:2181 --replica 1 --partition 1 --topic test
> bin/kafka-list-topic.sh --zookeeperlocalhost:2181
（3）Send some messages
> bin/kafka-console-producer.sh--broker-list localhost:9092 --topic test
（4）Start a consumer
> bin/kafka-console-consumer.sh--zookeeper localhost:2181 --topic test --from-beginning
kafka-console-producer.sh和kafka-console-cousumer.sh隻是系統提供的指令行工具。這裡啟動是為了測試是否能正常生産消費；驗證流程正确性
在實際開發中還是要自行開發自己的生産者與消費者；
kafka的安裝也可以參考我之前寫的文章：http://blog.csdn.net/weijonathan/article/details/18075967
Storm
Storm的主要特點如下：
可以使用各種程式設計語言。你可以在Storm之上使用各種程式設計語言。預設支援Clojure、Java、Ruby和Python。要增加對其他語言的支援，隻需實作一個簡單的Storm通信協定即可。
容錯性。Storm會管理工作程序和節點的故障。
水準擴充。計算是在多個線程、程序和伺服器之間并行進行的。
可靠的消息處理。Storm保證每個消息至少能得到一次完整處理。任務失敗時，它會負責從消息源重試消息。
本地模式。Storm有一個“本地模式”，可以在處理過程中完全模拟Storm叢集。這讓你可以快速進行開發和單元測試。
由于篇幅問題，具體的安裝步驟可以參考我之前寫的文章：http://blog.csdn.net/weijonathan/article/details/17762477
接下來重頭戲開始拉！那就是架構之間的整合啦
flume和kafka整合
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下：
注：這裡的flumeng-kafka-plugin.jar這個包，後面在github項目中已經移動到package目錄了。找不到的童鞋可以到package目錄擷取。
完成上面的步驟之後，我們來測試下flume+kafka這個流程有沒有走通；
我們先啟動flume，然後再啟動kafka，啟動步驟按之前的步驟執行；接下來我們使用kafka的kafka-console-consumer.sh腳本檢視是否有flume有沒有往Kafka傳輸資料；
以上這個是我的test.log檔案通過flume抓取傳到kafka的資料；說明我們的flume和kafka流程走通了；
大家還記得剛開始我們的流程圖麼，其中有一步是通過flume到kafka，還有一步是到hdfs的；而我們這邊還沒有提到如何存入kafka且同時存如hdfs；
flume是支援資料同步複制，同步複制流程圖如下，取自于flume官網，官網使用者指南位址：http://flume.apache.org/FlumeUserGuide.html
怎麼設定同步複制呢，看下面的配置：
#2個channel和2個sink的配置檔案 這裡我們可以設定兩個sink，一個是kafka的，一個是hdfs的；
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
具體配置大夥根據自己的需求去設定，這裡就不具體舉例了
kafka和storm的整合
2.使用maven package進行編譯，得到storm-kafka-0.8-plus-0.3.0-SNAPSHOT.jar包
--有轉載的童鞋注意下，這裡的包名之前寫錯了，現在改正确了！不好意思！
3.将該jar包及kafka_2.9.2-0.8.0-beta1.jar、metrics-core-2.2.0.jar、scala-library-2.9.2.jar
(這三個jar包在kafka項目中能找到)
備注：如果開發的項目需要其他jar，記得也要放進storm的Lib中比如用到了mysql就要添加mysql-connector-java-5.1.22-bin.jar到storm的lib下
那麼接下來我們把storm也重新開機下；
完成以上步驟之後，我們還有一件事情要做，就是使用kafka-storm0.8插件，寫一個自己的Storm程式；
這裡我給大夥附上一個我弄的storm程式，百度網盤分享位址：http://pan.baidu.com/s/1mgp0LLY
先稍微看下程式的建立Topology代碼
資料操作主要在WordCounter類中，這裡隻是使用簡單JDBC進行插入處理
這裡隻需要輸入一個參數作為Topology名稱就可以了！我們這裡使用本地模式，是以不輸入參數，直接看流程是否走通；
storm-0.9.0.1/bin/storm jar storm-start-demo-0.0.1-SNAPSHOT.jar com.storm.topology.MyTopology
先看下日志，這裡列印出來了往資料庫裡面插入資料了
然後我們檢視下資料庫；插入成功了！
到這裡我們的整個整合就完成了！
但是這裡還有一個問題，不知道大夥有沒有發現。這個也是@晨色星空J2EE跟我說的，其實我也應該想到的；
由于我們使用storm進行分布式流式計算，那麼分布式最需要注意的是資料一緻性以及避免髒資料的産生；是以我提供的測試項目隻能用于測試，正式開發不能這樣處理；
@晨色星空J2EE給的建議是建立一個zookeeper的分布式全局鎖，保證資料一緻性，避免髒資料錄入！
zookeeper用戶端架構大夥可以使用Netflix Curator來完成，由于這塊我還沒去看，是以隻能寫到這裡了！
這裡在一次謝謝羅寶和@晨色星空！
轉載的話請注明來源位址：http://blog.csdn.net/weijonathan/article/details/18301321 和 http://www.51studyit.com/html/notes/20140312/14.html