storm 原理簡介及單機版安裝指南

時間 2019-11-06

標籤 storm 原理簡介單機安裝指南欄目 Storm 简体版

原文原文鏈接

本文翻譯自： https://github.com/nathanmarz/storm/wiki/Tutorialhtml

Storm是一個分佈式的、高容錯的實時計算系統。
Storm對於實時計算的的意義至關於Hadoop對於批處理的意義。Hadoop爲咱們提供了Map和Reduce原語，使咱們對數據進行批處理變的很是的簡單和優美。一樣，Storm也對數據的實時計算提供了簡單Spout和Bolt原語。
Storm適用的場景：
一、流數據處理：Storm能夠用來用來處理源源不斷的消息，並將處理以後的結果保存到持久化介質中。
二、分佈式RPC：因爲Storm的處理組件都是分佈式的，並且處理延遲都極低，因此能夠Storm能夠作爲一個通用的分佈式RPC框架來使用。java

在這個教程裏面咱們將學習如何建立Topologies, 而且把topologies部署到storm的集羣裏面去。Java將是咱們主要的示範語言，個別例子會使用python以演示storm的多語言特性。node

一、準備工做

這個教程使用storm-starter項目裏面的例子。我推薦大家下載這個項目的代碼而且跟着教程一塊兒作。先讀一下：配置storm開發環境和新建一個strom項目這兩篇文章把你的機器設置好。python

二、一個Storm集羣的基本組件

storm的集羣表面上看和hadoop的集羣很是像。可是在Hadoop上面你運行的是MapReduce的Job, 而在Storm上面你運行的是Topology。它們是很是不同的 — 一個關鍵的區別是：一個MapReduce Job最終會結束，而一個Topology運永遠運行（除非你顯式的殺掉他）。mysql

在Storm的集羣裏面有兩種節點：控制節點(master node)和工做節點(worker node)。控制節點上面運行一個後臺程序： Nimbus，它的做用相似Hadoop裏面的JobTracker。Nimbus負責在集羣裏面分佈代碼，分配工做給機器，而且監控狀態。git

每個工做節點上面運行一個叫作Supervisor的節點（相似 TaskTracker）。Supervisor會監聽分配給它那臺機器的工做，根據須要啓動/關閉工做進程。每個工做進程執行一個Topology（相似 Job）的一個子集；一個運行的Topology由運行在不少機器上的不少工做進程 Worker（相似 Child）組成。github

storm topology結構sql

Storm VS MapReduce數據庫

Nimbus和Supervisor之間的全部協調工做都是經過一個Zookeeper集羣來完成。而且，nimbus進程和supervisor都是快速失敗（fail-fast)和無狀態的。全部的狀態要麼在Zookeeper裏面，要麼在本地磁盤上。這也就意味着你能夠用kill -9來殺死nimbus和supervisor進程，而後再重啓它們，它們能夠繼續工做，就好像什麼都沒有發生過似的。這個設計使得storm難以想象的穩定。centos

三、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。

四、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保證你不會有數據丟失，即便在一些機器意外停機而且消息被丟掉的狀況下。

五、數據模型(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的其它部分咱們接下來會解釋。

六、一個簡單的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。

七、流分組策略(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來處理。

八、使用別的語言來定義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.

九、可靠的消息處理

在這個教程的前面，咱們跳過了有關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

十、單機版安裝指南

環境：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://tianhailong.com/%E6%9C%AC%E5%9C%B0%E6%A8%A1%E5%BC%8F%E8%BF%90%E8%A1%8Cstorm%E7%9A%84demo.html/%E8%BF%90%E8%A1%8C%E6%97%A5%E5%BF%97

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/