storm如何實現(xiàn)單機版安裝

小編給大家分享一下storm如何實現(xiàn)單機版安裝，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！

Storm是一個分布式的、高容錯的實時計算系統(tǒng)。
Storm對于實時計算的的意義相當于Hadoop對于批處理的意義。Hadoop為我們提供了Map和Reduce原語，使我們對數(shù)據(jù)進行批處理變的非常的簡單和優(yōu)美。同樣，Storm也對數(shù)據(jù)的實時計算提供了簡單Spout和Bolt原語。
Storm適用的場景：
1、流數(shù)據(jù)處理：Storm可以用來用來處理源源不斷的消息，并將處理之后的結(jié)果保存到持久化介質(zhì)中。
2、分布式RPC：由于Storm的處理組件都是分布式的，而且處理延遲都極低，所以可以Storm可以做為一個通用的分布式RPC框架來使用。

在這個教程里面我們將學習如何創(chuàng)建Topologies, 并且把topologies部署到storm的集群里面去。Java將是我們主要的示范語言， 個別例子會使用python以演示storm的多語言特性。

1、準備工作

這個教程使用storm-starter項目里面的例子。我推薦你們下載這個項目的代碼并且跟著教程一起做。先讀一下：配置storm開發(fā)環(huán)境和新建一個strom項目這兩篇文章把你的機器設(shè)置好。

2、一個Storm集群的基本組件

storm的集群表面上看和hadoop的集群非常像。但是在Hadoop上面你運行的是MapReduce的Job, 而在Storm上面你運行的是Topology。它們是非常不一樣的 — 一個關(guān)鍵的區(qū)別是： 一個MapReduce Job最終會結(jié)束， 而一個Topology運永遠運行（除非你顯式的殺掉他）。

在Storm的集群里面有兩種節(jié)點： 控制節(jié)點(master node)和工作節(jié)點(worker node)?？刂乒?jié)點上面運行一個后臺程序： Nimbus， 它的作用類似Hadoop里面的JobTracker。Nimbus負責在集群里面分布代碼，分配工作給機器， 并且監(jiān)控狀態(tài)。

每一個工作節(jié)點上面運行一個叫做Supervisor的節(jié)點（類似 TaskTracker）。Supervisor會監(jiān)聽分配給它那臺機器的工作，根據(jù)需要 啟動/關(guān)閉工作進程。每一個工作進程執(zhí)行一個Topology（類似 Job）的一個子集；一個運行的Topology由運行在很多機器上的很多工作進程 Worker（類似 Child）組成。

                                                  storm topology結(jié)構(gòu)

bolt可以接收任意多個輸入stream， 作一些處理， 有些bolt可能還會發(fā)射一些新的stream。一些復雜的流轉(zhuǎn)換， 比如從一些tweet里面計算出熱門話題， 需要多個步驟， 從而也就需要多個bolt。 Bolt可以做任何事情: 運行函數(shù)， 過濾tuple, 做一些聚合， 做一些合并以及訪問數(shù)據(jù)庫等等。

Bolt處理輸入的Stream，并產(chǎn)生新的輸出Stream。Bolt可以執(zhí)行過濾、函數(shù)操作、Join、操作數(shù)據(jù)庫等任何操作。Bolt是一個被動的角色，其接口中有一個execute(Tuple input)方法，在接收到消息之后會調(diào)用此函數(shù)，用戶可以在此方法中執(zhí)行自己的處理邏輯。

topology結(jié)構(gòu)

topology里面的每一個節(jié)點都是并行運行的。 在你的topology里面， 你可以指定每個節(jié)點的并行度， storm則會在集群里面分配那么多線程來同時計算。

一個topology會一直運行直到你顯式停止它。storm自動重新分配一些運行失敗的任務， 并且storm保證你不會有數(shù)據(jù)丟失， 即使在一些機器意外停機并且消息被丟掉的情況下。

5、數(shù)據(jù)模型(Data Model)

storm使用tuple來作為它的數(shù)據(jù)模型。每個tuple是一堆值，每個值有一個名字，并且每個值可以是任何類型， 在我的理解里面一個tuple可以看作一個沒有方法的java對象?？傮w來看，storm支持所有的基本類型、字符串以及字節(jié)數(shù)組作為tuple的值類型。你也可以使用你自己定義的類型來作為值類型， 只要你實現(xiàn)對應的序列化器(serializer)。

一個Tuple代表數(shù)據(jù)流中的一個基本的處理單元，例如一條cookie日志，它可以包含多個Field，每個Field表示一個屬性。

Tuple本來應該是一個Key-Value的Map，由于各個組件間傳遞的tuple的字段名稱已經(jīng)事先定義好了，所以Tuple只需要按序填入各個Value，所以就是一個Value List。

一個沒有邊界的、源源不斷的、連續(xù)的Tuple序列就組成了Stream。

topology里面的每個節(jié)點必須定義它要發(fā)射的tuple的每個字段。 比如下面這個bolt定義它所發(fā)射的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發(fā)射單詞， 每個bolt在每個單詞后面加個”!!!”。這三個節(jié)點被排成一條線: spout發(fā)射單詞給第一個bolt， 第一個bolt然后把處理好的單詞發(fā)射給第二個bolt。如果spout發(fā)射的單詞是["bob"]和["john"], 那么第二個bolt會發(fā)射["bolt!!!!!!"]和["john!!!!!!"]出來。

我們使用setSpout和setBolt來定義Topology里面的節(jié)點。這些方法接收我們指定的一個id， 一個包含處理邏輯的對象(spout或者bolt), 以及你所需要的并行度。

這個包含處理的對象如果是spout那么要實現(xiàn)IRichSpout的接口， 如果是bolt，那么就要實現(xiàn)IRichBolt接口.
最后一個指定并行度的參數(shù)是可選的。它表示集群里面需要多少個thread來一起執(zhí)行這個節(jié)點。如果你忽略它那么storm會分配一個線程來執(zhí)行這個節(jié)點。

setBolt方法返回一個InputDeclarer對象， 這個對象是用來定義Bolt的輸入。 這里第一個Bolt聲明它要讀取spout所發(fā)射的所有的tuple — 使用shuffle grouping。而第二個bolt聲明它讀取第一個bolt所發(fā)射的tuple。shuffle grouping表示所有的tuple會被隨機的分發(fā)給bolt的所有task。給task分發(fā)tuple的策略有很多種，后面會介紹。

如果你想第二個bolt讀取spout和第一個bolt所發(fā)射的所有的tuple， 那么你應該這樣定義第二個bolt:

builder.setBolt(3,newExclamationBolt(),5)
            .shuffleGrouping(1)
            .shuffleGrouping(2);

讓我們深入地看一下這個topology里面的spout和bolt是怎么實現(xiàn)的。Spout負責發(fā)射新的tuple到這個topology里面來。TestWordSpout從["nathan", "mike", "jackson", "golda", "bertels"]里面隨機選擇一個單詞發(fā)射出來。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));
}

可以看到，實現(xiàn)很簡單。

ExclamationBolt把”!!!”拼接到輸入tuple后面。我們來看下ExclamationBolt的完整實現(xiàn)。

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用來發(fā)射tuple。Bolt可以在任何時候發(fā)射tuple — 在prepare, execute或者cleanup方法里面, 或者甚至在另一個線程里面異步發(fā)射。這里prepare方法只是簡單地把OutputCollector作為一個類字段保存下來給后面execute方法使用。

execute方法從bolt的一個輸入接收tuple(一個bolt可能有多個輸入源). ExclamationBolt獲取tuple的第一個字段，加上”!!!”之后再發(fā)射出去。如果一個bolt有多個輸入源，你可以通過調(diào)用Tuple#getSourceComponent方法來知道它是來自哪個輸入源的。

execute方法里面還有其它一些事情值得一提： 輸入tuple被作為emit方法的第一個參數(shù)，并且輸入tuple在最后一行被ack。這些呢都是Storm可靠性API的一部分，后面會解釋。

cleanup方法在bolt被關(guān)閉的時候調(diào)用， 它應該清理所有被打開的資源。但是集群不保證這個方法一定會被執(zhí)行。比如執(zhí)行task的機器down掉了，那么根本就沒有辦法來調(diào)用那個方法。cleanup設(shè)計的時候是被用來在local mode的時候才被調(diào)用(也就是說在一個進程里面模擬整個storm集群), 并且你想在關(guān)閉一些topology的時候避免資源泄漏。

最后，declareOutputFields定義一個叫做”word”的字段的tuple。

以local mode運行ExclamationTopology
讓我們看看怎么以local mode運行ExclamationToplogy。

storm的運行有兩種模式: 本地模式和分布式模式. 在本地模式中， storm用一個進程里面的線程來模擬所有的spout和bolt. 本地模式對開發(fā)和測試來說比較有用。 你運行storm-starter里面的topology的時候它們就是以本地模式運行的， 你可以看到topology里面的每一個組件在發(fā)射什么消息。

在分布式模式下， storm由一堆機器組成。當你提交topology給master的時候， 你同時也把topology的代碼提交了。master負責分發(fā)你的代碼并且負責給你的topolgoy分配工作進程。如果一個工作進程掛掉了， master節(jié)點會把認為重新分配到其它節(jié)點。關(guān)于如何在一個集群上面運行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對象來定義一個進程內(nèi)的集群。提交topology給這個虛擬的集群和提交topology給分布式集群是一樣的。通過調(diào)用submitTopology方法來提交topology， 它接受三個參數(shù)：要運行的topology的名字，一個配置對象以及要運行的topology本身。

topology的名字是用來唯一區(qū)別一個topology的，這樣你然后可以用這個名字來殺死這個topology的。前面已經(jīng)說過了， 你必須顯式的殺掉一個topology， 否則它會一直運行。

Conf對象可以配置很多東西， 下面兩個是最常見的：

1. TOPOLOGY_WORKERS(setNumWorkers) 定義你希望集群分配多少個工作進程給你來執(zhí)行這個topology. topology里面的每個組件會被需要線程來執(zhí)行。每個組件到底用多少個線程是通過setBolt和setSpout來指定的。這些線程都運行在工作進程里面. 每一個工作進程包含一些節(jié)點的一些工作線程。比如， 如果你指定300個線程，60個進程， 那么每個工作進程里面要執(zhí)行6個線程， 而這6個線程可能屬于不同的組件(Spout, Bolt)。你可以通過調(diào)整每個組件的并行度以及這些線程所在的進程數(shù)量來調(diào)整topology的性能。

2. TOPOLOGY_DEBUG(setDebug), 當它被設(shè)置成true的話， storm會記錄下每個組件所發(fā)射的每條消息。這在本地環(huán)境調(diào)試topology很有用， 但是在線上這么做的話會影響性能的。

感興趣的話可以去看看Conf對象的Javadoc去看看topology的所有配置。
可以看看創(chuàng)建一個新storm項目去看看怎么配置開發(fā)環(huán)境以使你能夠以本地模式運行topology.

運行中的Topology主要由以下三個組件組成的：

Worker processes（進程）

Executors (threads)（線程）

Tasks

Spout或者Bolt的Task個數(shù)一旦指定之后就不能改變了，而Executor的數(shù)量可以根據(jù)情況來進行動態(tài)的調(diào)整。默認情況下# executor = #tasks即一個Executor中運行著一個Task

1個worker進程執(zhí)行的是1個topology的子集（注：不會出現(xiàn)1個worker為多個topology服務）。1個worker進程會啟動1個或多個executor線程來執(zhí)行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線程里會在每次循環(huán)里順序調(diào)用所有task實例）。

task是最終運行spout或bolt中代碼的單元（注：1個task即為spout或bolt的1個實例，executor線程在執(zhí)行期間會調(diào)用該task的nextTuple或execute方法）。topology啟動后，1個component(spout或bolt)的task數(shù)目是固定不變的，但該component使用的executor線程數(shù)可以動態(tài)調(diào)整（例如：1個executor線程可以執(zhí)行該component的1個或多個task實例）。這意味著，對于1個component存在這樣的條件：#threads<=#tasks（即：線程數(shù)小于等于task數(shù)目）。默認情況下task的數(shù)目等于executor線程數(shù)目，即1個executor線程只運行1個task。

從task角度來看topology

當Bolt A的一個task要發(fā)送一個tuple給Bolt B， 它應該發(fā)送給Bolt B的哪個task呢？

stream grouping專門回答這種問題的。在我們深入研究不同的stream grouping之前， 讓我們看一下storm-starter里面的另外一個topology。WordCountTopology讀取一些句子， 輸出句子里面每個單詞出現(xiàn)的次數(shù).

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對于句子里面的每個單詞發(fā)射一個新的tuple, WordCount在內(nèi)存里面維護一個單詞->次數(shù)的mapping， WordCount每收到一個單詞， 它就更新內(nèi)存里面的統(tǒng)計狀態(tài)。

有好幾種不同的stream grouping:

1. 最簡單的grouping是shuffle grouping, 它隨機發(fā)給任何一個task。上面例子里面RandomSentenceSpout和SplitSentence之間用的就是shuffle grouping, shuffle grouping對各個task的tuple分配的比較均勻。

2. 一種更有趣的grouping是fields grouping, SplitSentence和WordCount之間使用的就是fields grouping, 這種grouping機制保證相同field值的tuple會去同一個task， 這對于WordCount來說非常關(guān)鍵，如果同一個單詞不去同一個task， 那么統(tǒng)計出來的單詞次數(shù)就不對了。

fields grouping是stream合并，stream聚合以及很多其它場景的基礎(chǔ)。在背后呢， fields grouping使用的一致性哈希來分配tuple的。

還有一些其它類型的stream grouping. 你可以在Concepts一章里更詳細的了解。

下面是一些常用的 “路由選擇” 機制：

Storm的Grouping即消息的Partition機制。當一個Tuple被發(fā)送時，如何確定將它發(fā)送個某個（些）Task來處理？？

l ShuffleGrouping：隨機選擇一個Task來發(fā)送。

l FiledGrouping：根據(jù)Tuple中Fields來做一致性hash，相同hash值的Tuple被發(fā)送到相同的Task。

l AllGrouping：廣播發(fā)送，將每一個Tuple發(fā)送到所有的Task。

l GlobalGrouping：所有的Tuple會被發(fā)送到某個Bolt中的id最小的那個Task。

l NoneGrouping：不關(guān)心Tuple發(fā)送給哪個Task來處理，等價于ShuffleGrouping。

l DirectGrouping：直接將Tuple發(fā)送到指定的Task來處理。

8、使用別的語言來定義Bolt

Bolt可以使用任何語言來定義。用其它語言定義的bolt會被當作子進程(subprocess)來執(zhí)行， storm使用JSON消息通過stdin/stdout來和這些subprocess通信。這個通信協(xié)議是一個只有100行的庫， storm團隊給這些庫開發(fā)了對應的Ruby, Python和Fancy版本。

下面是WordCountTopology里面的SplitSentence的定義:

publicstaticclassSplitSentenceextendsShellBoltimplementsIRichBolt {
    publicSplitSentence() {
        super("python","splitsentence.py");
    }
 
    publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(newFields("word"));
    }
}

SplitSentence繼承自ShellBolt并且聲明這個Bolt用python來運行，并且參數(shù)是: splitsentence.py。下面是splitsentence.py的定義:

importstorm
 
classSplitSentenceBolt(storm.BasicBolt):
    defprocess(self, tup):
        words=tup.values[0].split(" ")
        forwordinwords:
          storm.emit([word])
 
SplitSentenceBolt().run()

更多有關(guān)用其它語言定義Spout和Bolt的信息， 以及用其它語言來創(chuàng)建topology的 信息可以參見: Using non-JVM languages with Storm.

9、可靠的消息處理

在這個教程的前面，我們跳過了有關(guān)tuple的一些特征。這些特征就是storm的可靠性API： storm如何保證spout發(fā)出的每一個tuple都被完整處理?？纯础秙torm如何保證消息不丟失》以更深入了解storm的可靠性API.

Storm允許用戶在Spout中發(fā)射一個新的源Tuple時為其指定一個MessageId，這個MessageId可以是任意的Object對象。多個源Tuple可以共用同一個MessageId，表示這多個源Tuple對用戶來說是同一個消息單元。Storm的可靠性是指Storm會告知用戶每一個消息單元是否在一個指定的時間內(nèi)被完全處理。完全處理的意思是該MessageId綁定的源Tuple以及由該源Tuple衍生的所有Tuple都經(jīng)過了Topology中每一個應該到達的Bolt的處理。

ack機制即， spout發(fā)送的每一條消息，

• 在規(guī)定的時間內(nèi)，spout收到Acker的ack響應，即認為該tuple 被后續(xù)bolt成功處理

• 在規(guī)定的時間內(nèi)，沒有收到Acker的ack響應tuple，就觸發(fā)fail動作，即認為該tuple處理失敗，

• 或者收到Acker發(fā)送的fail響應tuple，也認為失敗，觸發(fā)fail動作

另外Ack機制還常用于限流作用： 為了避免spout發(fā)送數(shù)據(jù)太快，而bolt處理太慢，常常設(shè)置pending數(shù)，當spout有等于或超過pending數(shù)的tuple沒有收到ack或fail響應時，跳過執(zhí)行nextTuple， 從而限制spout發(fā)送數(shù)據(jù)。

通過conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, pending);設(shè)置spout pend數(shù)。

在Spout中由message 1綁定的tuple1和tuple2分別經(jīng)過bolt1和bolt2的處理，然后生成了兩個新的Tuple，并最終流向了bolt3。當bolt3處理完之后，稱message 1被完全處理了。

Storm中的每一個Topology中都包含有一個Acker組件。Acker組件的任務就是跟蹤從Spout中流出的每一個messageId所綁定的Tuple樹中的所有Tuple的處理情況。如果在用戶設(shè)置的最大超時時間內(nèi)這些Tuple沒有被完全處理，那么Acker會告訴Spout該消息處理失敗，相反則會告知Spout該消息處理成功。

那么Acker是如何記錄Tuple的處理結(jié)果呢？？

A xor A = 0.

A xor B…xor B xor A = 0，其中每一個操作數(shù)出現(xiàn)且僅出現(xiàn)兩次。

在Spout中，Storm系統(tǒng)會為用戶指定的MessageId生成一個對應的64位的整數(shù)，作為整個Tuple Tree的RootId。RootId會被傳遞給Acker以及后續(xù)的Bolt來作為該消息單元的唯一標識。同時，無論Spout還是Bolt每次新生成一個Tuple時，都會賦予該Tuple一個唯一的64位整數(shù)的Id。

當Spout發(fā)射完某個MessageId對應的源Tuple之后，它會告訴Acker自己發(fā)射的RootId以及生成的那些源Tuple的Id。而當Bolt處理完一個輸入Tuple并產(chǎn)生出新的Tuple時，也會告知Acker自己處理的輸入Tuple的Id以及新生成的那些Tuple的Id。Acker只需要對這些Id進行異或運算，就能判斷出該RootId對應的消息單元是否成功處理完成了。

如何使用Ack機制

• spout 在發(fā)送數(shù)據(jù)的時候帶上msgid

• 設(shè)置acker數(shù)至少大于0；Config.setNumAckers(conf, ackerParal);

• 在bolt中完成處理tuple時，執(zhí)行OutputCollector.ack(tuple), 當失敗處理時，執(zhí)行OutputCollector.fail(tuple); ** 推薦使用IBasicBolt， 因為IBasicBolt 自動封裝了OutputCollector.ack(tuple), 處理失敗時，請拋出FailedException，則自動執(zhí)行OutputCollector.fail(tuple)

如何關(guān)閉Ack機制

有2種途徑

• spout發(fā)送數(shù)據(jù)是不帶上msgid

• 設(shè)置acker數(shù)等于0

10、單機版安裝指南

環(huán)境：centos 6.4

安裝步驟請參考：http://blog.sina.com.cn/s/blog_546abd9f0101cce8.html

要注意上面的本地模式運行WordCount其實并沒有使用到上述安裝的工具，只是一個storm的虛擬環(huán)境下測試demo。那我們怎樣將程序運行在剛剛搭建的單機版的環(huán)境里面呢，
很簡單，官方的例子：
注意看官方實例中WordCountTopology類如果不帶參數(shù)其實是執(zhí)行的本地模式，也就是剛說的虛擬的環(huán)境，帶上參數(shù)就是將jar發(fā)送到了storm執(zhí)行了。
首先弄好環(huán)境：
啟動zookeeper：
/usr/local/zookeeper/bin/zkServer.sh  單機版直接啟動，不用修改什么配置，如集群就需要修改zoo.cfg另一篇文章會講到。
配置storm：
文件在/usr/local/storm/conf/storm.yaml
內(nèi)容：
 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配置的端口號一致，否則會出現(xiàn)通信錯誤，切記切記。當然你也可以配superevisor.slot.port，supervisor.slots.ports表示supervisor節(jié)點的槽數(shù)，就是最多能跑幾個worker進程（每個sprout或bolt默認只啟動一個worker，但是可以通過conf修改成多個）。
執(zhí)行：
# bin/storm nimbus（啟動主節(jié)點）
# bin/storm supervisor（啟動從節(jié)點）
執(zhí)行命令：# storm jar StormStarter.jar storm.starter.WordCountTopology test
此命令的作用就是用storm將jar發(fā)送給storm去執(zhí)行，后面的test是定義的toplogy名稱。
搞定，任務就發(fā)送到storm上運行起來了，還可以通過命令：
# bin/storm ui

然后執(zhí)行 jps 會看到 3 個進程：zookeeper 、nimbus、 supervisor
啟動ui，可以通過瀏覽器， ip:8080/ 查看運行i情況。
配置后，執(zhí)行 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

運行結(jié)果：

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!
...

以上是“storm如何實現(xiàn)單機版安裝”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對大家有所幫助，如果還想學習更多知識，歡迎關(guān)注億速云行業(yè)資訊頻道！