當(dāng)前位置：首頁(yè) > 编程资源 > 编程问答 >内容正文

编程问答

storm 简介及单机版安装指南

發(fā)布時(shí)間：2024/4/14 编程问答 37 豆豆

生活随笔收集整理的這篇文章主要介紹了 storm 简介及单机版安装指南小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

2019獨(dú)角獸企業(yè)重金招聘Python工程師標(biāo)準(zhǔn)>>>

轉(zhuǎn)載：http://www.open-open.com/lib/view/open1374979211233.html

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

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

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

1、準(zhǔn)備工作

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

2、一個(gè)Storm集群的基本組件

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

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

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

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

? ? ? ? ? ? ? ? ? ? ? ? ? Storm VS MapReduce

Nimbus和Supervisor之間的所有協(xié)調(diào)工作都是通過一個(gè)Zookeeper集群來(lái)完成。并且，nimbus進(jìn)程和supervisor都是快速失敗（fail-fast)和無(wú)狀態(tài)的。所有的狀態(tài)要么在Zookeeper里面，要么在本地磁盤上。這也就意味著你可以用kill -9來(lái)殺死nimbus和supervisor進(jìn)程，然后再重啟它們，它們可以繼續(xù)工作，就好像什么都沒有發(fā)生過似的。這個(gè)設(shè)計(jì)使得storm不可思議的穩(wěn)定。

3、Topologies

為了在storm上面做實(shí)時(shí)計(jì)算，你要去建立一些topologies。一個(gè)topology就是一個(gè)計(jì)算節(jié)點(diǎn)所組成的圖。Topology里面的每個(gè)處理節(jié)點(diǎn)都包含處理邏輯，而節(jié)點(diǎn)之間的連接則表示數(shù)據(jù)流動(dòng)的方向。

運(yùn)行一個(gè)Topology是很簡(jiǎn)單的。首先，把你所有的代碼以及所依賴的jar打進(jìn)一個(gè)jar包。然后運(yùn)行類似下面的這個(gè)命令。

1	strom jar all-your-code.jar backtype.storm.MyTopology arg1 arg2

這個(gè)命令會(huì)運(yùn)行主類: backtype.strom.MyTopology, 參數(shù)是arg1, arg2。這個(gè)類的main函數(shù)定義這個(gè)topology并且把它提交給Nimbus。storm jar負(fù)責(zé)連接到nimbus并且上傳jar文件。

因?yàn)閠opology的定義其實(shí)就是一個(gè)Thrift結(jié)構(gòu)并且nimbus就是一個(gè)Thrift服務(wù)，有可以用任何語(yǔ)言創(chuàng)建并且提交topology。上面的方面是用JVM
-based語(yǔ)言提交的最簡(jiǎn)單的方法, 看一下文章:?在生產(chǎn)集群上運(yùn)行topology去看看怎么啟動(dòng)以及停止topologies。

4、Stream

Stream是storm里面的關(guān)鍵抽象。一個(gè)stream是一個(gè)沒有邊界的tuple序列。storm提供一些原語(yǔ)來(lái)分布式地、可靠地把一個(gè)stream傳輸進(jìn)一個(gè)新的stream。比如：你可以把一個(gè)tweets流傳輸?shù)綗衢T話題的流。

storm提供的最基本的處理stream的原語(yǔ)是spout和bolt。你可以實(shí)現(xiàn)Spout和Bolt對(duì)應(yīng)的接口以處理你的應(yīng)用的邏輯。

spout的流的源頭。比如一個(gè)spout可能從Kestrel隊(duì)列里面讀取消息并且把這些消息發(fā)射成一個(gè)流。又比如一個(gè)spout可以調(diào)用twitter的一個(gè)api并且把返回的tweets發(fā)射成一個(gè)流。

通常Spout會(huì)從外部數(shù)據(jù)源（隊(duì)列、數(shù)據(jù)庫(kù)等）讀取數(shù)據(jù)，然后封裝成Tuple形式，之后發(fā)送到Stream中。Spout是一個(gè)主動(dòng)的角色，在接口內(nèi)部有個(gè)nextTuple函數(shù)，Storm框架會(huì)不停的調(diào)用該函數(shù)。

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

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

spout和bolt所組成一個(gè)網(wǎng)絡(luò)會(huì)被打包成topology， topology是storm里面最高一級(jí)的抽象（類似 Job），你可以把topology提交給storm的集群來(lái)運(yùn)行。topology的結(jié)構(gòu)在Topology那一段已經(jīng)說過了，這里就不再贅述了。

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

topology里面的每一個(gè)節(jié)點(diǎn)都是并行運(yùn)行的。在你的topology里面，你可以指定每個(gè)節(jié)點(diǎn)的并行度， storm則會(huì)在集群里面分配那么多線程來(lái)同時(shí)計(jì)算。

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

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

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

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

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

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

topology里面的每個(gè)節(jié)點(diǎn)必須定義它要發(fā)射的tuple的每個(gè)字段。比如下面這個(gè)bolt定義它所發(fā)射的tuple包含兩個(gè)字段，類型分別是: double和triple。

01	publicclassDoubleAndTripleBoltimplementsIRichBolt {

02	????privateOutputCollectorBase _collector;

04	????@Override

05	????publicvoidprepare(Map conf, TopologyContext context, OutputCollectorBase collector) {

06	????????_collector = collector;

????}

09	????@Override

10	????publicvoidexecute(Tuple input) {

11	????????intval = input.getInteger(0);

12	????????_collector.emit(input,newValues(val2, val3));

13	????????_collector.ack(input);

????}

16	????@Override

17	????publicvoidcleanup() {

????}

20	????@Override

21	????publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {

22	????????declarer.declare(newFields("double","triple"));

????}

}

declareOutputFields方法定義要輸出的字段： ["double", "triple"]。這個(gè)bolt的其它部分我們接下來(lái)會(huì)解釋。

6、一個(gè)簡(jiǎn)單的Topology

讓我們來(lái)看一個(gè)簡(jiǎn)單的topology的例子，我們看一下storm-starter里面的ExclamationTopology:

1	TopologyBuilder builder =newTopologyBuilder();

2	builder.setSpout(1,newTestWordSpout(),10);

3	builder.setBolt(2,newExclamationBolt(),3)

4	????????.shuffleGrouping(1);

5	builder.setBolt(3,newExclamationBolt(),2)

6	????????.shuffleGrouping(2);

這個(gè)Topology包含一個(gè)Spout和兩個(gè)Bolt。Spout發(fā)射單詞，每個(gè)bolt在每個(gè)單詞后面加個(gè)”!!!”。這三個(gè)節(jié)點(diǎn)被排成一條線: spout發(fā)射單詞給第一個(gè)bolt，第一個(gè)bolt然后把處理好的單詞發(fā)射給第二個(gè)bolt。如果spout發(fā)射的單詞是["bob"]和["john"], 那么第二個(gè)bolt會(huì)發(fā)射["bolt!!!!!!"]和["john!!!!!!"]出來(lái)。

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

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

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

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

1	builder.setBolt(3,newExclamationBolt(),5)

2	????????????.shuffleGrouping(1)

3	????????????.shuffleGrouping(2);

讓我們深入地看一下這個(gè)topology里面的spout和bolt是怎么實(shí)現(xiàn)的。Spout負(fù)責(zé)發(fā)射新的tuple到這個(gè)topology里面來(lái)。TestWordSpout從["nathan", "mike", "jackson", "golda", "bertels"]里面隨機(jī)選擇一個(gè)單詞發(fā)射出來(lái)。TestWordSpout里面的nextTuple()方法是這樣定義的：

1	publicvoidnextTuple() {

2	????Utils.sleep(100);

3	????finalString[] words =newString[] {"nathan","mike",

4	?????????????????????"jackson","golda","bertels"};

5	????finalRandom rand =newRandom();

6	????finalString word = words[rand.nextInt(words.length)];

7	????_collector.emit(newValues(word));

}

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

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

01	publicstaticclassExclamationBoltimplementsIRichBolt {

02	????OutputCollector _collector;

04	????publicvoidprepare(Map conf, TopologyContext context,

05	????????????????????????OutputCollector collector) {

06	????????_collector = collector;

????}

09	????publicvoidexecute(Tuple tuple) {

10	????????_collector.emit(tuple,newValues(tuple.getString(0) +"!!!"));

11	????????_collector.ack(tuple);

????}

14	????publicvoidcleanup() {

????}

17	????publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {

18	????????declarer.declare(newFields("word"));

????}

}

prepare方法提供給bolt一個(gè)Outputcollector用來(lái)發(fā)射tuple。Bolt可以在任何時(shí)候發(fā)射tuple — 在prepare, execute或者cleanup方法里面, 或者甚至在另一個(gè)線程里面異步發(fā)射。這里prepare方法只是簡(jiǎn)單地把OutputCollector作為一個(gè)類字段保存下來(lái)給后面execute方法使用。

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

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

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

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

以local mode運(yùn)行ExclamationTopology
讓我們看看怎么以local mode運(yùn)行ExclamationToplogy。

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

在分布式模式下， storm由一堆機(jī)器組成。當(dāng)你提交topology給master的時(shí)候，你同時(shí)也把topology的代碼提交了。master負(fù)責(zé)分發(fā)你的代碼并且負(fù)責(zé)給你的topolgoy分配工作進(jìn)程。如果一個(gè)工作進(jìn)程掛掉了， master節(jié)點(diǎn)會(huì)把認(rèn)為重新分配到其它節(jié)點(diǎn)。關(guān)于如何在一個(gè)集群上面運(yùn)行topology，你可以看看Running topologies on a production cluster文章。

下面是以本地模式運(yùn)行ExclamationTopology的代碼:

1	Config conf =newConfig();

2	conf.setDebug(true);

3	conf.setNumWorkers(2);

5	LocalCluster cluster =newLocalCluster();

6	cluster.submitTopology("test", conf, builder.createTopology());

7	Utils.sleep(10000);

8	cluster.killTopology("test");

9	cluster.shutdown();

首先，這個(gè)代碼定義通過定義一個(gè)LocalCluster對(duì)象來(lái)定義一個(gè)進(jìn)程內(nèi)的集群。提交topology給這個(gè)虛擬的集群和提交topology給分布式集群是一樣的。通過調(diào)用submitTopology方法來(lái)提交topology，它接受三個(gè)參數(shù)：要運(yùn)行的topology的名字，一個(gè)配置對(duì)象以及要運(yùn)行的topology本身。

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

Conf對(duì)象可以配置很多東西，下面兩個(gè)是最常見的：

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

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

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

運(yùn)行中的Topology主要由以下三個(gè)組件組成的：

Worker processes（進(jìn)程） Executors (threads)（線程） Tasks

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

7、流分組策略(Stream grouping)

流分組策略告訴topology如何在兩個(gè)組件之間發(fā)送tuple。要記住， spouts和bolts以很多task的形式在topology里面同步執(zhí)行。如果從task的粒度來(lái)看一個(gè)運(yùn)行的topology，它應(yīng)該是這樣的:

從task角度來(lái)看topology

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

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

1	TopologyBuilder builder =newTopologyBuilder();

3	builder.setSpout(1,newRandomSentenceSpout(),5);

4	builder.setBolt(2,newSplitSentence(),8)

5	????????.shuffleGrouping(1);

6	builder.setBolt(3,newWordCount(),12)

7	????????.fieldsGrouping(2,newFields("word"));

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

有好幾種不同的stream grouping:

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

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

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

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

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

Storm的Grouping即消息的Partition機(jī)制。當(dāng)一個(gè)Tuple被發(fā)送時(shí)，如何確定將它發(fā)送個(gè)某個(gè)（些）Task來(lái)處理？？

l ShuffleGrouping：隨機(jī)選擇一個(gè)Task來(lái)發(fā)送。 l FiledGrouping：根據(jù)Tuple中Fields來(lái)做一致性hash，相同hash值的Tuple被發(fā)送到相同的Task。 l AllGrouping：廣播發(fā)送，將每一個(gè)Tuple發(fā)送到所有的Task。 l GlobalGrouping：所有的Tuple會(huì)被發(fā)送到某個(gè)Bolt中的id最小的那個(gè)Task。 l NoneGrouping：不關(guān)心Tuple發(fā)送給哪個(gè)Task來(lái)處理，等價(jià)于ShuffleGrouping。 l DirectGrouping：直接將Tuple發(fā)送到指定的Task來(lái)處理。

8、使用別的語(yǔ)言來(lái)定義Bolt

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

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

1	publicstaticclassSplitSentenceextendsShellBoltimplementsIRichBolt {

2	????publicSplitSentence() {

3	????????super("python","splitsentence.py");

????}

6	????publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {

7	????????declarer.declare(newFields("word"));

????}

}

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

1	importstorm

3	classSplitSentenceBolt(storm.BasicBolt):

4	????defprocess(self, tup):

5	????????words=tup.values[0].split(" ")

6	????????forwordinwords:

7	??????????storm.emit([word])

9	SplitSentenceBolt().run()

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

9、可靠的消息處理

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

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

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

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

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

A xor A = 0.

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

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

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

10、單機(jī)版安裝指南

http://blog.sina.com.cn/s/blog_546abd9f0101cce8.html

要注意上面的本地模式運(yùn)行WordCount其實(shí)并沒有使用到上述安裝的工具，只是一個(gè)storm的虛擬環(huán)境下測(cè)試demo。那我們?cè)鯓訉⒊绦蜻\(yùn)行在剛剛搭建的單機(jī)版的環(huán)境里面呢，
很簡(jiǎn)單，官方的例子：
注意看官方實(shí)例中WordCountTopology類如果不帶參數(shù)其實(shí)是執(zhí)行的本地模式，也就是剛說的虛擬的環(huán)境，帶上參數(shù)就是將jar發(fā)送到了storm執(zhí)行了。
首先弄好環(huán)境：
啟動(dòng)zookeeper：
/usr/local/zookeeper/bin/zkServer.sh ?單機(jī)版直接啟動(dòng)，不用修改什么配置，如集群就需要修改zoo.cfg另一篇文章會(huì)講到。
配置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
這個(gè)腳本文件寫的不咋地，所以在配置時(shí)一定注意在每一項(xiàng)的開始時(shí)要加空格，冒號(hào)后也必須要加空格，否則storm就不認(rèn)識(shí)這個(gè)配置文件了。
說明一下：storm.local.dir表示storm需要用到的本地目錄。nimbus.host表示那一臺(tái)機(jī)器是master機(jī)器，即nimbus。storm.zookeeper.servers表示哪幾臺(tái)機(jī)器是zookeeper服務(wù)器。storm.zookeeper.port表示zookeeper的端口號(hào)，這里一定要與zookeeper配置的端口號(hào)一致，否則會(huì)出現(xiàn)通信錯(cuò)誤，切記切記。當(dāng)然你也可以配superevisor.slot.port，supervisor.slots.ports表示supervisor節(jié)點(diǎn)的槽數(shù)，就是最多能跑幾個(gè)worker進(jìn)程（每個(gè)sprout或bolt默認(rèn)只啟動(dòng)一個(gè)worker，但是可以通過conf修改成多個(gè)）。
執(zhí)行：
# bin/storm nimbus（啟動(dòng)主節(jié)點(diǎn)）
# bin/storm supervisor（啟動(dòng)從節(jié)點(diǎn)）
執(zhí)行命令：# storm jar StormStarter.jar storm.starter.WordCountTopology test
此命令的作用就是用storm將jar發(fā)送給storm去執(zhí)行，后面的test是定義的toplogy名稱。
搞定，任務(wù)就發(fā)送到storm上運(yùn)行起來(lái)了，還可以通過命令：
# bin/storm ui

然后執(zhí)行 jps 會(huì)看到 3 個(gè)進(jìn)程：zookeeper 、nimbus、?supervisor
啟動(dòng)ui，可以通過瀏覽器， ip:8080/ 查看運(yùn)行i情況。
配置后，執(zhí)行?storm ?jar sm.jar main.java.TopologyMain words.txt

也許會(huì)報(bào)：java.lang.NoClassDefFoundError: clojure.core.protocols$seq_reduce

這是由于我使用了 oracle JDK 1.7 的緣故，換成 open JDK 1.6 就正常了，

1	su -c "yum install java-1.6.0-openjdk"

具體參考：https://github.com/technomancy/leiningen/issues/676

測(cè)試代碼：

https://github.com/storm-book/examples-ch02-getting_started

運(yùn)行結(jié)果：

01	storm? jar sm.jar main.java.TopologyMain words.txt?

...

03	6020 [main] INFO? backtype.storm.messaging.loader? - Shutdown receiving-thread: [Getting-Started-Toplogie-1-1374946750, 4]

04	6020 [main] INFO? backtype.storm.daemon.worker? - Shut down receive thread

05	6020 [main] INFO? backtype.storm.daemon.worker? - Terminating zmq context

06	6020 [main] INFO? backtype.storm.daemon.worker? - Shutting down executors

OK:is

08	6021 [main] INFO? backtype.storm.daemon.executor? - Shutting down executor word-counter:[2 2]

OK:an

OK:storm

OK:simple

12	6023 [Thread-16] INFO? backtype.storm.util? - Async loop interrupted!

13	OK:application

OK:but

OK:very

16	OK:powerfull

OK:really

OK:

OK:StOrm

OK:is

OK:great

22	6038 [Thread-15] INFO? backtype.storm.util? - Async loop interrupted!

23	-- Word Counter [word-counter-2] --

really: 1

but: 1

26	application: 1

is: 2

great: 2

are: 1

test: 1

simple: 1

an: 1

33	powerfull: 1

storm: 3

very: 1

36	6043 [main] INFO? backtype.storm.daemon.executor? - Shut down executor word-counter:[2 2]

37	6044 [main] INFO? backtype.storm.daemon.executor? - Shutting down executor word-normalizer:[3 3]

38	6045 [Thread-18] INFO? backtype.storm.util? - Async loop interrupted!

39	6052 [Thread-17] INFO? backtype.storm.util? - Async loop interrupted!

40	6056 [main] INFO? backtype.storm.daemon.executor? - Shut down executor word-normalizer:[3 3]

41	6056 [main] INFO? backtype.storm.daemon.executor? - Shutting down executor word-reader:[4 4]

42	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://hitina.lofter.com/post/a8c5e_136579#

注：本文主體部分來(lái)源于?徐明明同學(xué) 翻譯的 storm wiki 教程，

http://xumingming.sinaapp.com/138/twitter-storm%E5%85%A5%E9%97%A8/

轉(zhuǎn)載于:https://my.oschina.net/u/2326085/blog/391241

總結(jié)

以上是生活随笔為你收集整理的storm 简介及单机版安装指南的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇：第三周项目5-数组作数据成员
下一篇： Jenkins学习七：Jenkins的授