storm 原理簡介及單機版安裝指南
本文翻譯自： https://github.com/nathanmarz/storm/wiki/Tutorial
Storm是一個分散式的、高容錯的實時計算系統。 Storm對於實時計算的的意義相當於Hadoop對於批處理的意義。Hadoop為我們提供了Map和Reduce原語，使我們對資料進行批處理變的非常的簡單和優美。同樣，Storm也對資料的實時計算提供了簡單Spout和Bolt原語。 Storm適用的場景： 1、流資料處理：Storm可以用來用來處理源源不斷的訊息，並將處理之後的結果儲存到持久化介質中。 2、分散式RPC：由於Storm的處理元件都是分散式的，而且處理延遲都極低，所以可以Storm可以做為一個通用的分散式RPC框架來使用。
在這個教程裡面我們將學習如何建立Topologies, 並且把topologies部署到storm的叢集裡面去。Java將是我們主要的示範語言， 個別例子會使用python以演示storm的多語言特性。
1、準備工作
這個教程使用storm-starter專案裡面的例子。我推薦你們下載這個專案的程式碼並且跟著教程一起做。先讀一下：配置storm開發環境和新建一個strom專案這兩篇文章把你的機器設定好。
2、一個Storm叢集的基本元件
storm的叢集表面上看和hadoop的叢集非常像。但是在Hadoop上面你執行的是MapReduce的Job, 而在Storm上面你執行的是Topology。它們是非常不一樣的 — 一個關鍵的區別是： 一個MapReduce Job最終會結束， 而一個Topology運永遠執行（除非你顯式的殺掉他）
在Storm的叢集裡面有兩種節點： 控制節點(master node)和工作節點(worker node)。控制節點上面執行一個後臺程式： Nimbus， 它的作用類似Hadoop裡面的JobTracker。Nimbus負責在叢集裡面分佈程式碼，分配工作給機器， 並且監控狀態。
每一個工作節點上面執行一個叫做Supervisor的節點（類似 TaskTracker）。Supervisor會監聽分配給它那臺機器的工作，根據需要 啟動/關閉工作程序。每一個工作程序執行一個Topology（類似 Job）的一個子集；一個執行的Topology由執行在很多機器上的很多工作程序 Worker（類似 Child）組成。
storm topology結構
Storm VS MapReduce
Nimbus和Supervisor之間的所有協調工作都是通過一個Zookeeper叢集來完成。並且，nimbus程序和supervisor都是快速失敗（fail-fast)和無狀態的。所有的狀態要麼在Zookeeper裡面， 要麼在本地磁碟上。這也就意味著你可以用kill -9來殺死nimbus和supervisor程序， 然後再重啟它們，它們可以繼續工作， 就好像什麼都沒有發生過似的。這個設計使得storm不可思議的穩定。
3、Topologies
為了在storm上面做實時計算， 你要去建立一些topologies。一個topology就是一個計算節點所組成的圖。Topology裡面的每個處理節點都包含處理邏輯， 而節點之間的連線則表示資料流動的方向。
執行一個Topology是很簡單的。首先，把你所有的程式碼以及所依賴的jar打進一個jar包。然後執行類似下面的這個命令。
strom jar all-your-code.jar backtype.storm.MyTopology arg1 arg2這個命令會執行主類: backtype.strom.MyTopology, 引數是arg1, arg2。這個類的main函式定義這個topology並且把它提交給Nimbus。storm jar負責連線到nimbus並且上傳jar檔案。
因為topology的定義其實就是一個Thrift結構並且nimbus就是一個Thrift服務， 有可以用任何語言建立並且提交topology。上面的方面是用JVM -based語言提交的最簡單的方法, 看一下文章: 在生產叢集上執行topology去看看怎麼啟動以及停止topologies。
4、Stream
Stream是storm裡面的關鍵抽象。一個stream是一個沒有邊界的tuple序列。storm提供一些原語來分散式地、可靠地把一個stream傳輸進一個新的stream。比如： 你可以把一個tweets流傳輸到熱門話題的流。
storm提供的最基本的處理stream的原語是spout和bolt。你可以實現Spout和Bolt對應的介面以處理你的應用的邏輯。
spout的流的源頭。比如一個spout可能從Kestrel佇列裡面讀取訊息並且把這些訊息發射成一個流。又比如一個spout可以呼叫twitter的一個api並且把返回的tweets發射成一個流。
通常Spout會從外部資料來源（佇列、資料庫等）讀取資料，然後封裝成Tuple形式，之後傳送到Stream中。Spout是一個主動的角色，在介面內部有個nextTuple函式，Storm框架會不停的呼叫該函式。
bolt可以接收任意多個輸入stream， 作一些處理， 有些bolt可能還會發射一些新的stream。一些複雜的流轉換， 比如從一些tweet裡面計算出熱門話題， 需要多個步驟， 從而也就需要多個bolt。 Bolt可以做任何事情: 執行函式， 過濾tuple, 做一些聚合， 做一些合併以及訪問資料庫等等。
Bolt處理輸入的Stream，併產生新的輸出Stream。Bolt可以執行過濾、函式操作、Join、操作資料庫等任何操作。Bolt是一個被動的角色，其介面中有一個execute(Tuple input)方法，在接收到訊息之後會呼叫此函式，使用者可以在此方法中執行自己的處理邏輯。
spout和bolt所組成一個網路會被打包成topology， topology是storm裡面最高一級的抽象（類似 Job）， 你可以把topology提交給storm的叢集來執行。topology的結構在Topology那一段已經說過了，這裡就不再贅述了。
topology結構
topology裡面的每一個節點都是並行執行的。 在你的topology裡面， 你可以指定每個節點的並行度， storm則會在叢集裡面分配那麼多執行緒來同時計算。
一個topology會一直執行直到你顯式停止它。storm自動重新分配一些執行失敗的任務， 並且storm保證你不會有資料丟失， 即使在一些機器意外停機並且訊息被丟掉的情況下。
5、資料模型(Data Model)
storm使用tuple來作為它的資料模型。每個tuple是一堆值，每個值有一個名字，並且每個值可以是任何型別， 在我的理解裡面一個tuple可以看作一個沒有方法的java物件。總體來看，storm支援所有的基本型別、字串以及位元組陣列作為tuple的值型別。你也可以使用你自己定義的型別來作為值型別， 只要你實現對應的序列化器(serializer)。
一個Tuple代表資料流中的一個基本的處理單元，例如一條cookie日誌，它可以包含多個Field，每個Field表示一個屬性。
Tuple本來應該是一個Key-Value的Map，由於各個元件間傳遞的tuple的欄位名稱已經事先定義好了，所以Tuple只需要按序填入各個Value，所以就是一個Value List。
一個沒有邊界的、源源不斷的、連續的Tuple序列就組成了Stream。
topology裡面的每個節點必須定義它要發射的tuple的每個欄位。 比如下面這個bolt定義它所發射的tuple包含兩個欄位，型別分別是: double和triple。
publicclassDoubleAndTripleBoltimplementsIRichBolt {
privateOutputCollectorBase _collector;
@Override
publicvoidprepare(Map conf, TopologyContext context, OutputCollectorBase collector) {
_collector = collector;
}
@Override
publicvoidexecute(Tuple input) {
intval = input.getInteger(0);
_collector.emit(input,newValues(val*2, val*3));
_collector.ack(input);
}
@Override
publicvoidcleanup() {
}
@Override
publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(newFields("double","triple"));
}
}declareOutputFields方法定義要輸出的欄位 ： ["double", "triple"]。這個bolt的其它部分我們接下來會解釋。
6、一個簡單的Topology
讓我們來看一個簡單的topology的例子， 我們看一下storm-starter裡面的ExclamationTopology:
TopologyBuilder builder =newTopologyBuilder();
builder.setSpout(1,newTestWordSpout(),10);
builder.setBolt(2,newExclamationBolt(),3)
.shuffleGrouping(1);
builder.setBolt(3,newExclamationBolt(),2)
.shuffleGrouping(2);這個Topology包含一個Spout和兩個Bolt。Spout發射單詞， 每個bolt在每個單詞後面加個”!!!”。這三個節點被排成一條線: spout發射單詞給第一個bolt， 第一個bolt然後把處理好的單詞發射給第二個bolt。如果spout發射的單詞是["bob"]和["john"], 那麼第二個bolt會發射["bolt!!!!!!"]和["john!!!!!!"]出來。
我們使用setSpout和setBolt來定義Topology裡面的節點。這些方法接收我們指定的一個id， 一個包含處理邏輯的物件(spout或者bolt), 以及你所需要的並行度。
這個包含處理的物件如果是spout那麼要實現IRichSpout的介面， 如果是bolt，那麼就要實現IRichBolt介面. 最後一個指定並行度的引數是可選的。它表示叢集裡面需要多少個thread來一起執行這個節點。如果你忽略它那麼storm會分配一個執行緒來執行這個節點。
setBolt方法返回一個InputDeclarer物件， 這個物件是用來定義Bolt的輸入。 這裡第一個Bolt宣告它要讀取spout所發射的所有的tuple — 使用shuffle grouping。而第二個bolt宣告它讀取第一個bolt所發射的tuple。shuffle grouping表示所有的tuple會被隨機的分發給bolt的所有task。給task分發tuple的策略有很多種，後面會介紹。
如果你想第二個bolt讀取spout和第一個bolt所發射的所有的tuple， 那麼你應該這樣定義第二個bolt:
builder.setBolt(3,newExclamationBolt(),5)
.shuffleGrouping(1)
.shuffleGrouping(2);讓我們深入地看一下這個topology裡面的spout和bolt是怎麼實現的。Spout負責發射新的tuple到這個topology裡面來。TestWordSpout從["nathan", "mike", "jackson", "golda", "bertels"]裡面隨機選擇一個單詞發射出來。TestWordSpout裡面的nextTuple()方法是這樣定義的：
publicvoidnextTuple() {
Utils.sleep(100);
finalString[] words =newString[] {"nathan","mike",
"jackson","golda","bertels"};
finalRandom rand =newRandom();
finalString word = words[rand.nextInt(words.length)];
_collector.emit(newValues(word));
}可以看到，實現很簡單。
ExclamationBolt把”!!!”拼接到輸入tuple後面。我們來看下ExclamationBolt的完整實現。
publicstaticclassExclamationBoltimplementsIRichBolt {
OutputCollector _collector;
publicvoidprepare(Map conf, TopologyContext context,
OutputCollector collector) {
_collector = collector;
}
publicvoidexecute(Tuple tuple) {
_collector.emit(tuple,newValues(tuple.getString(0) +"!!!"));
_collector.ack(tuple);
}
publicvoidcleanup() {
}
publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(newFields("word"));
}
}prepare方法提供給bolt一個Outputcollector用來發射tuple。Bolt可以在任何時候發射tuple — 在prepare, execute或者cleanup方法裡面, 或者甚至在另一個執行緒裡面非同步發射。這裡prepare方法只是簡單地把OutputCollector作為一個類欄位儲存下來給後面execute方法使用。
execute方法從bolt的一個輸入接收tuple(一個bolt可能有多個輸入源). ExclamationBolt獲取tuple的第一個欄位，加上”!!!”之後再發射出去。如果一個bolt有多個輸入源，你可以通過呼叫Tuple#getSourceComponent方法來知道它是來自哪個輸入源的。
execute方法裡面還有其它一些事情值得一提： 輸入tuple被作為emit方法的第一個引數，並且輸入tuple在最後一行被ack。這些呢都是Storm可靠性API的一部分，後面會解釋。
cleanup方法在bolt被關閉的時候呼叫， 它應該清理所有被開啟的資源。但是叢集不保證這個方法一定會被執行。比如執行task的機器down掉了，那麼根本就沒有辦法來呼叫那個方法。cleanup設計的時候是被用來在local mode的時候才被呼叫(也就是說在一個程序裡面模擬整個storm叢集), 並且你想在關閉一些topology的時候避免資源洩漏。
最後，declareOutputFields定義一個叫做”word”的欄位的tuple。
以local mode執行ExclamationTopology 讓我們看看怎麼以local mode執行ExclamationToplogy。
storm的執行有兩種模式: 本地模式和分散式模式. 在本地模式中， storm用一個程序裡面的執行緒來模擬所有的spout和bolt. 本地模式對開發和測試來說比較有用。 你執行storm-starter裡面的topology的時候它們就是以本地模式執行的， 你可以看到topology裡面的每一個元件在發射什麼訊息。
在分散式模式下， storm由一堆機器組成。當你提交topology給master的時候， 你同時也把topology的程式碼提交了。master負責分發你的程式碼並且負責給你的topolgoy分配工作程序。如果一個工作程序掛掉了， master節點會把認為重新分配到其它節點。關於如何在一個叢集上面執行topology， 你可以看看Running topologies on a production cluster文章。
下面是以本地模式執行ExclamationTopology的程式碼:
Config conf =newConfig();
conf.setDebug(true);
conf.setNumWorkers(2);
LocalCluster cluster =newLocalCluster();
cluster.submitTopology("test", conf, builder.createTopology());
Utils.sleep(10000);
cluster.killTopology("test");
cluster.shutdown();首先， 這個程式碼定義通過定義一個LocalCluster物件來定義一個程序內的叢集。提交topology給這個虛擬的叢集和提交topology給分散式叢集是一樣的。通過呼叫submitTopology方法來提交topology， 它接受三個引數：要執行的topology的名字，一個配置物件以及要執行的topology本身。
topology的名字是用來唯一區別一個topology的，這樣你然後可以用這個名字來殺死這個topology的。前面已經說過了， 你必須顯式的殺掉一個topology， 否則它會一直執行。
Conf物件可以配置很多東西， 下面兩個是最常見的：
- TOPOLOGY_WORKERS(setNumWorkers) 定義你希望叢集分配多少個工作程序給你來執行這個topology. topology裡面的每個元件會被需要執行緒來執行。每個元件到底用多少個執行緒是通過setBolt和setSpout來指定的。這些執行緒都執行在工作程序裡面. 每一個工作程序包含一些節點的一些工作執行緒。比如， 如果你指定300個執行緒，60個程序， 那麼每個工作程序裡面要執行6個執行緒， 而這6個執行緒可能屬於不同的元件(Spout, Bolt)。你可以通過調整每個元件的並行度以及這些執行緒所在的程序數量來調整topology的效能。
- TOPOLOGY_DEBUG(setDebug), 當它被設定成true的話， storm會記錄下每個元件所發射的每條訊息。這在本地環境除錯topology很有用， 但是在線上這麼做的話會影響效能的。
感興趣的話可以去看看Conf物件的Javadoc去看看topology的所有配置。 可以看看建立一個新storm專案去看看怎麼配置開發環境以使你能夠以本地模式執行topology.
執行中的Topology主要由以下三個元件組成的：
Worker processes（程序）
Executors (threads)（執行緒）
Tasks
Spout或者Bolt的Task個數一旦指定之後就不能改變了，而Executor的數量可以根據情況來進行動態的調整。預設情況下# executor = #tasks即一個Executor中執行著一個Task
該拓撲一共有兩個工作程序(Worker)，2+2+6=10個執行器(Executor)，2+4+6=12個任務。因此，每個工作程序可以分配到10/2=5個執行器，12/2=6個任務。預設情況下，一個執行器執行一個任務，但是如果指定了任務的數目，則任務會平均分配到執行器中，因此，GreenBolt的例項"green-bolt"的一個執行器將會分配到4/2個任務。
1個worker程序執行的是1個topology的子集（注：不會出現1個worker為多個topology服務）。1個worker程序會啟動1個或多個executor執行緒來執行1個topology的component(spout或bolt)。因此，1個執行中的topology就是由叢集中多臺物理機上的多個worker程序組成的。 executor是1個被worker程序啟動的單獨執行緒。每個executor只會執行1個topology的1個component(spout或bolt)的task(set)（注：task可以是1個或多個，storm預設是1個component只生成1個task，executor執行緒裡會在每次迴圈裡順序呼叫所有task例項）。 task是最終執行spout或bolt中程式碼的單元（注：1個task即為spout或bolt的1個例項，executor執行緒在執行期間會呼叫該task的nextTuple或execute方法）。topology啟動後，1個component(spout或bolt)的task數目是固定不變的，但該component使用的executor執行緒數可以動態調整（例如：1個executor執行緒可以執行該component的1個或多個task例項）。這意味著，對於1個component存在這樣的條件：#threads<=#tasks（即：執行緒數小於等於task數目）。預設情況下task的數目等於executor執行緒數目，即1個executor執行緒只執行1個task。
7、流分組策略(Stream grouping)
流分組策略告訴topology如何在兩個元件之間傳送tuple。 要記住， spouts和bolts以很多task的形式在topology裡面同步執行。如果從task的粒度來看一個執行的topology， 它應該是這樣的:
從task角度來看topology
當Bolt A的一個task要傳送一個tuple給Bolt B， 它應該傳送給Bolt B的哪個task呢？
stream grouping專門回答這種問題的。在我們深入研究不同的stream grouping之前， 讓我們看一下storm-starter裡面的另外一個topology。WordCountTopology讀取一些句子， 輸出句子裡面每個單詞出現的次數.
TopologyBuilder builder =newTopologyBuilder();
builder.setSpout(1,newRandomSentenceSpout(),5);
builder.setBolt(2,newSplitSentence(),8)
.shuffleGrouping(1);
builder.setBolt(3,newWordCount(),12)
.fieldsGrouping(2,newFields("word"));SplitSentence對於句子裡面的每個單詞發射一個新的tuple, WordCount在記憶體裡面維護一個單詞->次數的mapping， WordCount每收到一個單詞， 它就更新記憶體裡面的統計狀態。
有好幾種不同的stream grouping:
- 最簡單的grouping是shuffle grouping, 它隨機發給任何一個task。上面例子裡面RandomSentenceSpout和SplitSentence之間用的就是shuffle grouping, shuffle grouping對各個task的tuple分配的比較均勻。
- 一種更有趣的grouping是fields grouping, SplitSentence和WordCount之間使用的就是fields grouping, 這種grouping機制保證相同field值的tuple會去同一個task， 這對於WordCount來說非常關鍵，如果同一個單詞不去同一個task， 那麼統計出來的單詞次數就不對了。
fields grouping是stream合併，stream聚合以及很多其它場景的基礎。在背後呢， fields grouping使用的一致性雜湊來分配tuple的。
還有一些其它型別的stream grouping. 你可以在Concepts一章裡更詳細的瞭解。
下面是一些常用的 “路由選擇” 機制：
Storm的Grouping即訊息的Partition機制。當一個Tuple被髮送時，如何確定將它傳送個某個（些）Task來處理？？
l ShuffleGrouping：隨機選擇一個Task來發送。
l FiledGrouping：根據Tuple中Fields來做一致性hash，相同hash值的Tuple被髮送到相同的Task。
l AllGrouping：廣播發送，將每一個Tuple傳送到所有的Task。
l GlobalGrouping：所有的Tuple會被髮送到某個Bolt中的id最小的那個Task。
l NoneGrouping：不關心Tuple傳送給哪個Task來處理，等價於ShuffleGrouping。
l DirectGrouping：直接將Tuple傳送到指定的Task來處理。
8、使用別的語言來定義Bolt
Bolt可以使用任何語言來定義。用其它語言定義的bolt會被當作子程序(subprocess)來執行， storm使用JSON訊息通過stdin/stdout來和這些subprocess通訊。這個通訊協議是一個只有100行的庫， storm團隊給這些庫開發了對應的Ruby, Python和Fancy版本。
下面是WordCountTopology裡面的SplitSentence的定義:
publicstaticclassSplitSentenceextendsShellBoltimplementsIRichBolt {
publicSplitSentence() {
super("python","splitsentence.py");
}
publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(newFields("word"));
}
}SplitSentence繼承自ShellBolt並且宣告這個Bolt用python來執行，並且引數是: splitsentence.py。下面是splitsentence.py的定義:
importstorm
classSplitSentenceBolt(storm.BasicBolt):
defprocess(self, tup):
words=tup.values[0].split(" ")
forwordinwords:
storm.emit([word])
SplitSentenceBolt().run()更多有關用其它語言定義Spout和Bolt的資訊， 以及用其它語言來建立topology的 資訊可以參見: Using non-JVM languages with Storm.
9、可靠的訊息處理
在這個教程的前面，我們跳過了有關tuple的一些特徵。這些特徵就是storm的可靠性API： storm如何保證spout發出的每一個tuple都被完整處理。看看《storm如何保證訊息不丟失》以更深入瞭解storm的可靠性API.
Storm允許使用者在Spout中發射一個新的源Tuple時為其指定一個MessageId，這個MessageId可以是任意的Object物件。多個源Tuple可以共用同一個MessageId，表示這多個源Tuple對使用者來說是同一個訊息單元。Storm的可靠性是指Storm會告知使用者每一個訊息單元是否在一個指定的時間內被完全處理。完全處理的意思是該MessageId繫結的源Tuple以及由該源Tuple衍生的所有Tuple都經過了Topology中每一個應該到達的Bolt的處理。
ack機制即， spout傳送的每一條訊息，
- 在規定的時間內，spout收到Acker的ack響應，即認為該tuple 被後續bolt成功處理
- 在規定的時間內，沒有收到Acker的ack響應tuple，就觸發fail動作，即認為該tuple處理失敗，
- 或者收到Acker傳送的fail響應tuple，也認為失敗，觸發fail動作
另外Ack機制還常用於限流作用： 為了避免spout傳送資料太快，而bolt處理太慢，常常設定pending數，當spout有等於或超過pending數的tuple沒有收到ack或fail響應時，跳過執行nextTuple， 從而限制spout傳送資料。
通過conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, pending);設定spout pend數。
在Spout中由message 1繫結的tuple1和tuple2分別經過bolt1和bolt2的處理，然後生成了兩個新的Tuple，並最終流向了bolt3。當bolt3處理完之後，稱message 1被完全處理了。
Storm中的每一個Topology中都包含有一個Acker元件。Acker元件的任務就是跟蹤從Spout中流出的每一個messageId所繫結的Tuple樹中的所有Tuple的處理情況。如果在使用者設定的最大超時時間內這些Tuple沒有被完全處理，那麼Acker會告訴Spout該訊息處理失敗，相反則會告知Spout該訊息處理成功。
那麼Acker是如何記錄Tuple的處理結果呢？？
A xor A = 0.
A xor B…xor B xor A = 0，其中每一個操作數出現且僅出現兩次。
在Spout中，Storm系統會為使用者指定的MessageId生成一個對應的64位的整數，作為整個Tuple Tree的RootId。RootId會被傳遞給Acker以及後續的Bolt來作為該訊息單元的唯一標識。同時，無論Spout還是Bolt每次新生成一個Tuple時，都會賦予該Tuple一個唯一的64位整數的Id。
當Spout發射完某個MessageId對應的源Tuple之後，它會告訴Acker自己發射的RootId以及生成的那些源Tuple的Id。而當Bolt處理完一個輸入Tuple併產生出新的Tuple時，也會告知Acker自己處理的輸入Tuple的Id以及新生成的那些Tuple的Id。Acker只需要對這些Id進行異或運算，就能判斷出該RootId對應的訊息單元是否成功處理完成了。
如何使用Ack機制
- spout 在傳送資料的時候帶上msgid
- 設定acker數至少大於0；
Config.setNumAckers(conf, ackerParal); - 在bolt中完成處理tuple時，執行OutputCollector.ack(tuple), 當失敗處理時，執行OutputCollector.fail(tuple); ** 推薦使用IBasicBolt， 因為IBasicBolt 自動封裝了OutputCollector.ack(tuple), 處理失敗時，請丟擲FailedException，則自動執行OutputCollector.fail(tuple)
如何關閉Ack機制
有2種途徑
- spout傳送資料是不帶上msgid
- 設定acker數等於0
10、單機版安裝指南
環境：centos 6.4
安裝步驟請參考：http://blog.sina.com.cn/s/blog_546abd9f0101cce8.html
要注意上面的本地模式執行WordCount其實並沒有使用到上述安裝的工具，只是一個storm的虛擬環境下測試demo。那我們怎樣將程式執行在剛剛搭建的單機版的環境裡面呢， 很簡單，官方的例子： 注意看官方例項中WordCountTopology類如果不帶引數其實是執行的本地模式，也就是剛說的虛擬的環境，帶上引數就是將jar傳送到了storm執行了。 首先弄好環境： 啟動zookeeper： /usr/local/zookeeper/bin/zkServer.sh 單機版直接啟動，不用修改什麼配置，如叢集就需要修改zoo.cfg另一篇文章會講到。 配置storm： 檔案在/usr/local/storm/conf/storm.yaml 內容： storm.zookeeper.servers: - 127.0.0.1 storm.zookeeper.port: 2181 nimbus.host: "127.0.0.1" storm.local.dir: "/tmp/storm" supervisor.slots.ports: - 6700 - 6701 - 6702 - 6703 這個指令碼檔案寫的不咋地，所以在配置時一定注意在每一項的開始時要加空格，冒號後也必須要加空格，否則storm就不認識這個配置檔案了。 說明一下：storm.local.dir表示storm需要用到的本地目錄。nimbus.host表示那一臺機器是master機器，即nimbus。storm.zookeeper.servers表示哪幾臺機器是zookeeper伺服器。storm.zookeeper.port表示zookeeper的埠號，這裡一定要與zookeeper配置的埠號一致，否則會出現通訊錯誤，切記切記。當然你也可以配superevisor.slot.port，supervisor.slots.ports表示supervisor節點的槽數，就是最多能跑幾個worker程序（每個sprout或bolt預設只啟動一個worker，但是可以通過conf修改成多個）。 執行： # bin/storm nimbus（啟動主節點） # bin/storm supervisor（啟動從節點） 執行命令：# storm jar StormStarter.jar storm.starter.WordCountTopology test 此命令的作用就是用storm將jar傳送給storm去執行，後面的test是定義的toplogy名稱。 搞定，任務就傳送到storm上執行起來了，還可以通過命令： # bin/storm ui
然後執行 jps 會看到 3 個程序：zookeeper 、nimbus、 supervisor 啟動ui，可以通過瀏覽器， ip:8080/ 檢視執行i情況。 配置後，執行 storm jar sm.jar main.java.TopologyMain words.txt
也許會報：java.lang.NoClassDefFoundError: clojure.core.protocols$seq_reduce
這是由於我使用了 oracle JDK 1.7 的緣故，換成 open JDK 1.6 就正常了，
su -c "yum install java-1.6.0-openjdk-devel"具體參考：https://github.com/technomancy/leiningen/issues/676
測試程式碼：
https://github.com/storm-book/examples-ch02-getting_started
執行結果：
storm jar sm.jar main.java.TopologyMain words.txt
...
6020 [main] INFO backtype.storm.messaging.loader - Shutdown receiving-thread: [Getting-Started-Toplogie-1-1374946750, 4]
6020 [main] INFO backtype.storm.daemon.worker - Shut down receive thread
6020 [main] INFO backtype.storm.daemon.worker - Terminating zmq context
6020 [main] INFO backtype.storm.daemon.worker - Shutting down executors
OK:is
6021 [main] INFO backtype.storm.daemon.executor - Shutting down executor word-counter:[2 2]
OK:an
OK:storm
OK:simple
6023 [Thread-16] INFO backtype.storm.util - Async loop interrupted!
OK:application
OK:but
OK:very
OK:powerfull
OK:really
OK:
OK:StOrm
OK:is
OK:great
6038 [Thread-15] INFO backtype.storm.util - Async loop interrupted!
-- Word Counter [word-counter-2] --
really: 1
but: 1
application: 1
is: 2
great: 2
are: 1
test: 1
simple: 1
an: 1
powerfull: 1
storm: 3
very: 1
6043 [main] INFO backtype.storm.daemon.executor - Shut down executor word-counter:[2 2]
6044 [main] INFO backtype.storm.daemon.executor - Shutting down executor word-normalizer:[3 3]
6045 [Thread-18] INFO backtype.storm.util - Async loop interrupted!
6052 [Thread-17] INFO backtype.storm.util - Async loop interrupted!
6056 [main] INFO backtype.storm.daemon.executor - Shut down executor word-normalizer:[3 3]
6056 [main] INFO backtype.storm.daemon.executor - Shutting down executor word-reader:[4 4]
6058 [Thread-19] INFO backtype.storm.util - Async loop interrupted!
...其它參考地址：
https://github.com/philipgao/storm-demo
http://blog.sina.com.cn/s/blog_8ae7b3fe010124mr.html
http://blog.jobbole.com/48595/ Storm：最火的流式處理框架
http://www.searchtb.com/2012/09/introduction-to-storm.html storm簡介
https://www.ibm.com/developerworks/cn/opensource/os-twitterstorm/#list1 使用 Twitter Storm 處理實時的大資料
http://blog.csdn.net/tntzbzc/article/details/19974515 storm 計算 CCU 的小例子
分散式安裝指南：
http://hitina.lofter.com/post/a8c5e_136579#
注：本文主體部分來源於 徐明明同學 翻譯的 storm wiki 教程，
http://xumingming.sinaapp.com/138/twitter-storm%E5%85%A5%E9%97%A8/
flume+kafka+storm+mysql架構設計
http://blog.csdn.net/mylittlered/article/details/20810265
實時計算在“大眾點評”網