storm入門教程第一章storm概述

1.1實時流計算互聯(lián)網(wǎng)從誕生的第一時間起，對世界的最大的改變就是讓信息能夠實時交互，從而大大加速了各個環(huán)節(jié)的效率。正因為大家對信息實時響應、實時交互的需求，軟件行業(yè)除了個人操作系統(tǒng)之外，數(shù)據(jù)庫(更精確的說是關系型數(shù)據(jù)庫)應該是軟件行業(yè)發(fā)展最快、收益最為豐厚的產(chǎn)品了。記得十年前，很多銀行別說實時轉賬，連實時查詢都做不到，但是數(shù)據(jù)庫和高速網(wǎng)絡改變了這個情況。

隨著互聯(lián)網(wǎng)的更進一步發(fā)展，從Portal信息瀏覽型到Search信息搜索型到SNS關系交互傳遞型，以及電子商務、互聯(lián)網(wǎng)旅游生活產(chǎn)品等將生活中的流通環(huán)節(jié)在線化。對效率的要求讓大家對于實時性的要求進一步提升，而信息的交互和溝通正在從點對點往信息鏈甚至信息網(wǎng)的方向發(fā)展，這樣必然帶來數(shù)據(jù)在各個維度的交叉關聯(lián)，數(shù)據(jù)爆炸已不可避免。因此流式處理加NoSQL產(chǎn)品應運而生，分別解決實時框架和數(shù)據(jù)大規(guī)模存儲計算的問題。

早在7、8年前諸如UC伯克利、斯坦福等大學就開始了對流式數(shù)據(jù)處理的研究，但是由于更多的關注于金融行業(yè)的業(yè)務場景或者互聯(lián)網(wǎng)流量監(jiān)控的業(yè)務場景，以及當時互聯(lián)網(wǎng)數(shù)據(jù)場景的限制，造成了研究多是基于對傳統(tǒng)數(shù)據(jù)庫處理的流式化，對流式框架本身的研究偏少。目前這樣的研究逐漸沒有了聲音，工業(yè)界更多的精力轉向了實時數(shù)據(jù)庫。

2010年Yahoo！對S4的開源，2011年twitter對Storm的開源，改變了這個情況。以前互聯(lián)網(wǎng)的開發(fā)人員在做一個實時應用的時候，除了要關注應用邏輯計算處理本身，還要為了數(shù)據(jù)的實時流轉、交互、分布大傷腦筋。但是現(xiàn)在情況卻大為不同，以Storm為例，開發(fā)人員可以快速的搭建一套健壯、易用的實時流處理框架，配合SQL產(chǎn)品或者NoSQL產(chǎn)品或者MapReduce計算平臺，就可以低成本的做出很多以前很難想象的實時產(chǎn)品：比如一淘數(shù)據(jù)部的量子恒道品牌旗下的多個產(chǎn)品就是構建在實時流處理平臺上的。

本教程是一本對storm的基礎介紹手冊，但是我們也希望它不僅僅是一本storm的使用手冊，我們會在其中加入更多我們在實際數(shù)據(jù)生產(chǎn)過程的經(jīng)驗和應用的架構，最后的目的是幫助所有愿意使用實時流處理框架的技術同仁，同時也默默的改變這個世界。1.2Storm特點Storm是一個開源的分布式實時計算系統(tǒng)，可以簡單、可靠的處理大量的數(shù)據(jù)流。Storm有很多使用場景：如實時分析，在線機器學習，持續(xù)計算，分布式RPC，ETL等等。Storm支持水平擴展，具有高容錯性，保證每個消息都會得到處理，而且處理速度很快(在一個小集群中，每個結點每秒可以處理數(shù)以百萬計的消息)。Storm的部署和運維都很便捷，而且更為重要的是可以使用任意編程語言來開發(fā)應用。

Storm有如下特點：

編程模型簡單

在大數(shù)據(jù)處理方面相信大家對hadoop已經(jīng)耳熟能詳，基于GoogleMap/Reduce來實現(xiàn)的Hadoop為開發(fā)者提供了map、reduce原語，使并行批處理程序變得非常地簡單和優(yōu)美。同樣，Storm也為大數(shù)據(jù)的實時計算提供了一些簡單優(yōu)美的原語，這大大降低了開發(fā)并行實時處理的任務的復雜性，幫助你快速、高效的開發(fā)應用。

可擴展

在Storm集群中真正運行topology的主要有三個實體：工作進程、線程和任務。Storm集群中的每臺機器上都可以運行多個工作進程，每個工作進程又可創(chuàng)建多個線程，每個線程可以執(zhí)行多個任務，任務是真正進行數(shù)據(jù)處理的實體，我們開發(fā)的spout、bolt就是作為一個或者多個任務的方式執(zhí)行的。

因此，計算任務在多個線程、進程和服務器之間并行進行，支持靈活的水平擴展。

高可靠性

Storm可以保證spout發(fā)出的每條消息都能被“完全處理”，這也是直接區(qū)別于其他實時系統(tǒng)的地方，如S4。

請注意，spout發(fā)出的消息后續(xù)可能會觸發(fā)產(chǎn)生成千上萬條消息，可以形象的理解為一棵消息樹，其中spout發(fā)出的消息為樹根，Storm會跟蹤這棵消息樹的處理情況，只有當這棵消息樹中的所有消息都被處理了，Storm才會認為spout發(fā)出的這個消息已經(jīng)被“完全處理”。如果這棵消息樹中的任何一個消息處理失敗了，或者整棵消息樹在限定的時間內(nèi)沒有“完全處理”，那么spout發(fā)出的消息就會重發(fā)。

考慮到盡可能減少對內(nèi)存的消耗，Storm并不會跟蹤消息樹中的每個消息，而是采用了一些特殊的策略，它把消息樹當作一個整體來跟蹤，對消息樹中所有消息的唯一id進行異或計算，通過是否為零來判定spout發(fā)出的消息是否被“完全處理”，這極大的節(jié)約了內(nèi)存和簡化了判定邏輯，后面會對這種機制進行詳細介紹。

這種模式，每發(fā)送一個消息，都會同步發(fā)送一個ack/fail，對于網(wǎng)絡的帶寬會有一定的消耗，如果對于可靠性要求不高，可通過使用不同的emit接口關閉該模式。

上面所說的，Storm保證了每個消息至少被處理一次，但是對于有些計算場合，會嚴格要求每個消息只被處理一次，幸而Storm的0.7.0引入了事務性拓撲，解決了這個問題，后面會有詳述。

高容錯性

如果在消息處理過程中出了一些異常，Storm會重新安排這個出問題的處理單元。Storm保證一個處理單元永遠運行(除非你顯式殺掉這個處理單元)。

當然，如果處理單元中存儲了中間狀態(tài)，那么當處理單元重新被Storm啟動的時候，需要應用自己處理中間狀態(tài)的恢復。

支持多種編程語言

除了用java實現(xiàn)spout和bolt，你還可以使用任何你熟悉的編程語言來完成這項工作，這一切得益于Storm所謂的多語言協(xié)議。多語言協(xié)議是Storm內(nèi)部的一種特殊協(xié)議，允許spout或者bolt使用標準輸入和標準輸出來進行消息傳遞，傳遞的消息為單行文本或者是json編碼的多行。

Storm支持多語言編程主要是通過ShellBolt,ShellSpout和ShellProcess這些類來實現(xiàn)的，這些類都實現(xiàn)了IBolt和ISpout接口，以及讓shell通過java的ProcessBuilder類來執(zhí)行腳本或者程序的協(xié)議。

可以看到，采用這種方式，每個tuple在處理的時候都需要進行json的編解碼，因此在吞吐量上會有較大影響。

支持本地模式

Storm有一種“本地模式”，也就是在進程中模擬一個Storm集群的所有功能，以本地模式運行topology跟在集群上運行topology類似，這對于我們開發(fā)和測試來說非常有用。

高效

用ZeroMQ作為底層消息隊列,保證消息能快速被處理第二章Storm術語介紹及構建Topology2.1Storm基本概念在運行一個Storm任務之前，需要了解一些概念：TopologiesStreamsSpoutsBoltsStreamgroupingsReliabilityTasksWorkersConfigurationStorm集群和Hadoop集群表面上看很類似。但是Hadoop上運行的是MapReducejobs，而在Storm上運行的是拓撲(topology)，這兩者之間是非常不一樣的。一個關鍵的區(qū)別是：一個MapReducejob最終會結束，而一個topology永遠會運行(除非你手動kill掉)。在Storm的集群里面有兩種節(jié)點：控制節(jié)點(masternode)和工作節(jié)點(workernode)?？刂乒?jié)點上面運行一個叫Nimbus后臺程序，它的作用類似Hadoop里面的JobTracker。Nimbus負責在集群里面分發(fā)代碼，分配計算任務給機器，并且監(jiān)控狀態(tài)。每一個工作節(jié)點上面運行一個叫做Supervisor的節(jié)點。Supervisor會監(jiān)聽分配給它那臺機器的工作，根據(jù)需要啟動/關閉工作進程。每一個工作進程執(zhí)行一個topology的一個子集；一個運行的topology由運行在很多機器上的很多工作進程組成。Nimbus和Supervisor之間的所有協(xié)調(diào)工作都是通過Zookeeper集群完成。另外，Nimbus進程和Supervisor進程都是快速失敗(fail-fast)和無狀態(tài)的。所有的狀態(tài)要么在zookeeper里面，要么在本地磁盤上。這也就意味著你可以用kill-9來殺死Nimbus和Supervisor進程，然后再重啟它們，就好像什么都沒有發(fā)生過。這個設計使得Storm異常的穩(wěn)定。一個topology是spouts和bolts組成的圖，通過streamgroupings將圖中的spouts和bolts連接起來，如下圖：一個topology會一直運行直到你手動kill掉，Storm自動重新分配執(zhí)行失敗的任務，并且Storm可以保證你不會有數(shù)據(jù)丟失(如果開啟了高可靠性的話)。如果一些機器意外停機它上面的所有任務會被轉移到其他機器上。運行一個topology很簡單。首先，把你所有的代碼以及所依賴的jar打進一個jar包。然后運行類似下面的這個命令：stormjarall-my-code.jarbacktype.storm.MyTopologyarg1arg2復制代碼這個命令會運行主類:backtype.strom.MyTopology,參數(shù)是arg1,arg2。這個類的main函數(shù)定義這個topology并且把它提交給Nimbus。stormjar負責連接到Nimbus并且上傳jar包。Topology的定義是一個Thrift結構，并且Nimbus就是一個Thrift服務，你可以提交由任何語言創(chuàng)建的topology。上面的方面是用JVM-based語言提交的最簡單的方法。消息流stream是storm里的關鍵抽象。一個消息流是一個沒有邊界的tuple序列，而這些tuple序列會以一種分布式的方式并行地創(chuàng)建和處理。通過對stream中tuple序列中每個字段命名來定義stream。在默認的情況下，tuple的字段類型可以是：integer，long，short，byte，string，double，float，boolean和bytearray。你也可以自定義類型(只要實現(xiàn)相應的序列化器)。每個消息流在定義的時候會被分配給一個id，因為單向消息流使用的相當普遍，OutputFieldsDeclarer定義了一些方法讓你可以定義一個stream而不用指定這個id。在這種情況下這個stream會分配個值為‘default’默認的id。Storm提供的最基本的處理stream的原語是spout和bolt。你可以實現(xiàn)spout和bolt提供的接口來處理你的業(yè)務邏輯。消息源spout是Storm里面一個topology里面的消息生產(chǎn)者。一般來說消息源會從一個外部源讀取數(shù)據(jù)并且向topology里面發(fā)出消息：tuple。Spout可以是可靠的也可以是不可靠的。如果這個tuple沒有被storm成功處理，可靠的消息源spouts可以重新發(fā)射一個tuple，但是不可靠的消息源spouts一旦發(fā)出一個tuple就不能重發(fā)了。消息源可以發(fā)射多條消息流stream。使用OutputFieldsDeclarer.declareStream來定義多個stream，然后使用SpoutOutputCollector來發(fā)射指定的stream。Spout類里面最重要的方法是nextTuple。要么發(fā)射一個新的tuple到topology里面或者簡單的返回如果已經(jīng)沒有新的tuple。要注意的是nextTuple方法不能阻塞，因為storm在同一個線程上面調(diào)用所有消息源spout的方法。另外兩個比較重要的spout方法是ack和fail。storm在檢測到一個tuple被整個topology成功處理的時候調(diào)用ack，否則調(diào)用fail。storm只對可靠的spout調(diào)用ack和fail。所有的消息處理邏輯被封裝在bolts里面。Bolts可以做很多事情：過濾，聚合，查詢數(shù)據(jù)庫等等。Bolts可以簡單的做消息流的傳遞。復雜的消息流處理往往需要很多步驟，從而也就需要經(jīng)過很多bolts。比如算出一堆圖片里面被轉發(fā)最多的圖片就至少需要兩步：第一步算出每個圖片的轉發(fā)數(shù)量。第二步找出轉發(fā)最多的前10個圖片。(如果要把這個過程做得更具有擴展性那么可能需要更多的步驟)。Bolts可以發(fā)射多條消息流，使用OutputFieldsDeclarer.declareStream定義stream，使用OutputCollector.emit來選擇要發(fā)射的stream。Bolts的主要方法是execute,它以一個tuple作為輸入，bolts使用OutputCollector來發(fā)射tuple，bolts必須要為它處理的每一個tuple調(diào)用OutputCollector的ack方法，以通知Storm這個tuple被處理完成了，從而通知這個tuple的發(fā)射者spouts。一般的流程是：bolts處理一個輸入tuple,發(fā)射0個或者多個tuple,然后調(diào)用ack通知storm自己已經(jīng)處理過這個tuple了。storm提供了一個IBasicBolt會自動調(diào)用ack。定義一個topology的其中一步是定義每個bolt接收什么樣的流作為輸入。streamgrouping就是用來定義一個stream應該如果分配數(shù)據(jù)給bolts上面的多個tasks。Storm里面有7種類型的streamgroupingShuffleGrouping:隨機分組，隨機派發(fā)stream里面的tuple，保證每個bolt接收到的tuple數(shù)目大致相同。FieldsGrouping：按字段分組，比如按userid來分組，具有同樣userid的tuple會被分到相同的Bolts里的一個task，而不同的userid則會被分配到不同的bolts里的task。AllGrouping：廣播發(fā)送，對于每一個tuple，所有的bolts都會收到。GlobalGrouping：全局分組，這個tuple被分配到storm中的一個bolt的其中一個task。再具體一點就是分配給id值最低的那個task。NonGrouping：不分組，這個分組的意思是說stream不關心到底誰會收到它的tuple。目前這種分組和Shufflegrouping是一樣的效果，有一點不同的是storm會把這個bolt放到這個bolt的訂閱者同一個線程里面去執(zhí)行。DirectGrouping：直接分組，這是一種比較特別的分組方法，用這種分組意味著消息的發(fā)送者指定由消息接收者的哪個task處理這個消息。只有被聲明為DirectStream的消息流可以聲明這種分組方法。而且這種消息tuple必須使用emitDirect方法來發(fā)射。消息處理者可以通過TopologyContext來獲取處理它的消息的task的id(OutputCollector.emit方法也會返回task的id)。Localorshufflegrouping：如果目標bolt有一個或者多個task在同一個工作進程中，tuple將會被隨機發(fā)生給這些tasks。否則，和普通的ShuffleGrouping行為一致。Storm保證每個tuple會被topology完整的執(zhí)行。Storm會追蹤由每個spouttuple所產(chǎn)生的tuple樹(一個bolt處理一個tuple之后可能會發(fā)射別的tuple從而形成樹狀結構)，并且跟蹤這棵tuple樹什么時候成功處理完。每個topology都有一個消息超時的設置，如果storm在這個超時的時間內(nèi)檢測不到某個tuple樹到底有沒有執(zhí)行成功，那么topology會把這個tuple標記為執(zhí)行失敗，并且過一會兒重新發(fā)射這個tuple。為了利用Storm的可靠性特性，在你發(fā)出一個新的tuple以及你完成處理一個tuple的時候你必須要通知storm。這一切是由OutputCollector來完成的。通過emit方法來通知一個新的tuple產(chǎn)生了，通過ack方法通知一個tuple處理完成了。Storm的可靠性我們在Storm入門教程4會深入介紹。每一個spout和bolt會被當作很多task在整個集群里執(zhí)行。每一個executor對應到一個線程，在這個線程上運行多個task，而streamgrouping則是定義怎么從一堆task發(fā)射tuple到另外一堆task。你可以調(diào)用TopologyBuilder類的setSpout和setBolt來設置并行度(也就是有多少個task)。一個topology可能會在一個或者多個worker(工作進程)里面執(zhí)行，每個worker是一個物理JVM并且執(zhí)行整個topology的一部分。比如，對于并行度是300的topology來說，如果我們使用50個工作進程來執(zhí)行，那么每個工作進程會處理其中的6個tasks。Storm會盡量均勻的工作分配給所有的worker。Storm里面有一堆參數(shù)可以配置來調(diào)整Nimbus,Supervisor以及正在運行的topology的行為，一些配置是系統(tǒng)級別的，一些配置是topology級別的。default.yaml里面有所有的默認配置。你可以通過定義個storm.yaml在你的classpath里來覆蓋這些默認配置。并且你也可以在代碼里面設置一些topology相關的配置信息(使用StormSubmitter)。2.2構建Topology我們將設計一個topology，來實現(xiàn)對一個句子里面的單詞出現(xiàn)的頻率進行統(tǒng)計。這是一個簡單的例子，目的是讓大家對于topology快速上手，有一個初步的理解。在開始開發(fā)Storm項目的第一步，就是要設計topology。確定好你的數(shù)據(jù)處理邏輯，我們今天將的這個簡單的例子，topology也非常簡單。整個topology如下：

整個topology分為三個部分：KestrelSpout:數(shù)據(jù)源，負責發(fā)送sentenceSplitsentence:負責將sentence切分Wordcount:負責對單詞的頻率進行累加這個topology從kestrelqueue讀取句子,并把句子劃分成單詞,然后匯總每個單詞出現(xiàn)的次數(shù),一個tuple負責讀取句子,每一個tuple分別對應計算每一個單詞出現(xiàn)的次數(shù),大概樣子如下所示：

1)構建maven環(huán)境：為了開發(fā)stormtopology,你需要把storm相關的jar包添加到classpath里面去：要么手動添加所有相關的jar包，要么使用maven來管理所有的依賴。storm的jar包發(fā)布在Clojars(一個maven庫),如果你使用maven的話，把下面的配置添加在你項目的pom.xml里面。<repository><id></id><url>/repo</url></repository><dependency><groupId>storm</groupId><artifactId>storm</artifactId><version>0.5.3</version><scope>test</scope></dependency>2)定義topology：TopologyBuilderbuilder=newTopologyBuilder();builder.setSpout(1,newKestrelSpout(“”,22133,”sentence_queue”,newStringScheme()));builder.setBolt(2,newSplitSentence(),10).shuffleGrouping(1);builder.setBolt(3,newWordCount(),20).fieldsGrouping(2,newFields(“word”));這種topology的spout從句子隊列中讀取句子，在位于一個Kestrel的服務器端口22133。Spout用setSpout方法插入一個獨特的id到topology。Topology中的每個節(jié)點必須給予一個id，id是由其他bolts用于訂閱該節(jié)點的輸出流。KestrelSpout在topology中id為1。setBolt是用于在Topology中插入bolts。在topology中定義的第一個bolts是切割句子的bolts。這個bolts將句子流轉成成單詞流。讓我們看看SplitSentence實施：publicclassSplitSentenceimplementsIBasicBolt{publicvoidprepare(Mapconf,TopologyContextcontext){}publicvoidexecute(Tupletuple,BasicOutputCollectorcollector){Stringsentence=tuple.getString(0);for(Stringword:sentence.split(“”)){collector.emit(newValues(word));}}publicvoidcleanup(){}publicvoiddeclareOutputFields(OutputFieldsDeclarerdeclarer){declarer.declare(newFields(“word”));}關鍵的方法是execute方法。正如你可以看到，它將句子拆分成單詞，并發(fā)出每個單詞作為一個新的元組。另一個重要的方法是declareOutputFields，其中宣布bolts輸出元組的架構。在這里宣布，它發(fā)出一個域為word的元組setBolt的最后一個參數(shù)是你想為bolts的并行量。SplitSentencebolts是10個并發(fā)，這將導致在storm集群中有十個線程并行執(zhí)行。你所要做的的是增加bolts的并行量在遇到topology的瓶頸時。setBolt方法返回一個對象，用來定義bolts的輸入。例如，SplitSentence螺栓訂閱組件“1”使用隨機分組的輸出流。“1”是指已經(jīng)定義KestrelSpout。我將解釋在某一時刻的隨機分組的一部分。到目前為止，最要緊的是，SplitSentencebolts會消耗KestrelSpout發(fā)出的每一個元組。下面在讓我們看看wordcount的實現(xiàn)：publicclassWordCountimplementsIBasicBolt{privateMap<String,Integer>_counts=newHashMap<String,Integer>();publicvoidprepare(Mapconf,TopologyContextcontext){}publicvoidexecute(Tupletuple,BasicOutputCollectorcollector){Stringword=tuple.getString(0);intcount;if(_counts.containsKey(word)){count=_counts.get(word);}else{count=0;}count++;_counts.put(word,count);collector.emit(newValues(word,count));}publicvoidcleanup(){}publicvoiddeclareOutputFields(OutputFieldsDeclarerdeclarer){declarer.declare(newFields(“word”,“count”));}}SplitSentence對于句子里面的每個單詞發(fā)射一個新的tuple,WordCount在內(nèi)存里面維護一個單詞->次數(shù)的mapping，WordCount每收到一個單詞，它就更新內(nèi)存里面的統(tǒng)計狀態(tài)。storm的運行有兩種模式:本地模式和分布式模式.1)本地模式：storm用一個進程里面的線程來模擬所有的spout和bolt.本地模式對開發(fā)和測試來說比較有用。你運行storm-starter里面的topology的時候它們就是以本地模式運行的，你可以看到topology里面的每一個組件在發(fā)射什么消息。2)分布式模式：storm由一堆機器組成。當你提交topology給master的時候，你同時也把topology的代碼提交了。master負責分發(fā)你的代碼并且負責給你的topolgoy分配工作進程。如果一個工作進程掛掉了，master節(jié)點會把認為重新分配到其它節(jié)點。3)下面是以本地模式運行的代碼:Configconf=newConfig();conf.setDebug(true);conf.setNumWorkers(2);LocalClustercluster=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對象可以配置很多東西，下面兩個是最常見的：TOPOLOGY_WORKERS(setNumWorkers)定義你希望集群分配多少個工作進程給你來執(zhí)行這個topology.topology里面的每個組件會被需要線程來執(zhí)行。每個組件到底用多少個線程是通過setBolt和setSpout來指定的。這些線程都運行在工作進程里面.每一個工作進程包含一些節(jié)點的一些工作線程。比如，如果你指定300個線程，60個進程，那么每個工作進程里面要執(zhí)行6個線程，而這6個線程可能屬于不同的組件(Spout,Bolt)。你可以通過調(diào)整每個組件的并行度以及這些線程所在的進程數(shù)量來調(diào)整topology的性能。TOPOLOGY_DEBUG(setDebug),當它被設置成true的話，storm會記錄下每個組件所發(fā)射的每條消息。這在本地環(huán)境調(diào)試topology很有用，但是在線上這么做的話會影響性能的。結論：本章從storm的基本對象的定義，到廣泛的介紹了storm的開發(fā)環(huán)境，從一個簡單的例子講解了topology的構建和定義。希望大家可以從本章的內(nèi)容對storm有一個基本的理解和概念，并且已經(jīng)可以構建一個簡單的topology??！第三章Storm安裝部署步驟本文以TwitterStorm官方Wiki為基礎，詳細描述如何快速搭建一個Storm集群，其中，項目實踐中遇到的問題及經(jīng)驗總結，在相應章節(jié)以“注意事項”的形式給出。3.1Storm集群組件Storm集群中包含兩類節(jié)點：主控節(jié)點（MasterNode）和工作節(jié)點（WorkNode）。其分別對應的角色如下：1.主控節(jié)點（MasterNode）上運行一個被稱為Nimbus的后臺程序，它負責在Storm集群內(nèi)分發(fā)代碼，分配任務給工作機器，并且負責監(jiān)控集群運行狀態(tài)。Nimbus的作用類似于Hadoop中JobTracker的角色。2.每個工作節(jié)點（WorkNode）上運行一個被稱為Supervisor的后臺程序。Supervisor負責監(jiān)聽從Nimbus分配給它執(zhí)行的任務，據(jù)此啟動或停止執(zhí)行任務的工作進程。每一個工作進程執(zhí)行一個Topology的子集；一個運行中的Topology由分布在不同工作節(jié)點上的多個工作進程組成。

Storm集群組件Nimbus和Supervisor節(jié)點之間所有的協(xié)調(diào)工作是通過Zookeeper集群來實現(xiàn)的。此外，Nimbus和Supervisor進程都是快速失?。╢ail-fast)和無狀態(tài)（stateless）的；Storm集群所有的狀態(tài)要么在Zookeeper集群中，要么存儲在本地磁盤上。這意味著你可以用kill-9來殺死Nimbus和Supervisor進程，它們在重啟后可以繼續(xù)工作。這個設計使得Storm集群擁有不可思議的穩(wěn)定性。3.2安裝Storm集群這一章節(jié)將詳細描述如何搭建一個Storm集群。下面是接下來需要依次完成的安裝步驟：1.搭建Zookeeper集群；2.安裝Storm依賴庫；3.下載并解壓Storm發(fā)布版本；4.修改storm.yaml配置文件；5.啟動Storm各個后臺進程。3.2.1搭建Zookeeper集群Storm使用Zookeeper協(xié)調(diào)集群，由于Zookeeper并不用于消息傳遞，所以Storm給Zookeeper帶來的壓力相當?shù)?。大多?shù)情況下，單個節(jié)點的Zookeeper集群足夠勝任，不過為了確保故障恢復或者部署大規(guī)模Storm集群，可能需要更大規(guī)模節(jié)點的Zookeeper集群（對于Zookeeper集群的話，官方推薦的最小節(jié)點數(shù)為3個）。在Zookeeper集群的每臺機器上完成以下安裝部署步驟：1.下載安裝JavaJDK，官方下載鏈接為/javase/downloads/index.jsp，JDK版本為JDK6或以上。2.根據(jù)Zookeeper集群的負載情況，合理設置Java堆大小，盡可能避免發(fā)生swap，導致Zookeeper性能下降。保守起見，4GB內(nèi)存的機器可以為Zookeeper分配3GB最大堆空間。3.下載后解壓安裝Zookeeper包，官方下載鏈接為/zookeeper/releases.html。4.根據(jù)Zookeeper集群節(jié)點情況，在conf目錄下創(chuàng)建Zookeeper配置文件zoo.cfg：tickTime=2000

dataDir=/var/zookeeper/

clientPort=2181

initLimit=5

syncLimit=2

server.1=zoo1:2888:3888

server.2=zoo2:2888:3888

server.3=zoo3:2888:3888復制代碼其中，dataDir指定Zookeeper的數(shù)據(jù)文件目錄；其中server.id=host:port:port，id是為每個Zookeeper節(jié)點的編號，保存在dataDir目錄下的myid文件中，zoo1~zoo3表示各個Zookeeper節(jié)點的hostname，第一個port是用于連接leader的端口，第二個port是用于leader選舉的端口。5.在dataDir目錄下創(chuàng)建myid文件，文件中只包含一行，且內(nèi)容為該節(jié)點對應的server.id中的id編號。6.啟動Zookeeper服務：java-cpzookeeper.jar:lib/log4j-1.2.15.jar:conf\org.apache.zookeeper.server.quorum.QuorumPeerMainzoo.cfg復制代碼或者bin/zkServer.shstart復制代碼7.通過Zookeeper客戶端測試服務是否可用：java-cpzookeeper.jar:src/java/lib/log4j-1.2.15.jar:conf:src/java/lib/jline-0.9.94.jar\org.apache.zookeeper.ZooKeeperMain-server:2181復制代碼或者bin/zkCli.sh-server:2181復制代碼

注意事項：由于Zookeeper是快速失?。╢ail-fast)的，且遇到任何錯誤情況，進程均會退出，因此，最好能通過監(jiān)控程序將Zookeeper管理起來，保證Zookeeper退出后能被自動重啟。詳情參考這里。Zookeeper運行過程中會在dataDir目錄下生成很多日志和快照文件，而Zookeeper運行進程并不負責定期清理合并這些文件，導致占用大量磁盤空間，因此，需要通過cron等方式定期清除沒用的日志和快照文件。詳情參考這里。具體命令格式如下：java-cpzookeeper.jar:log4j.jar:conforg.apache.zookeeper.server.PurgeTxnLog<dataDir><snapDir>-n<count>3.2.2安裝Storm依賴庫接下來，需要在Nimbus和Supervisor機器上安裝Storm的依賴庫，具體如下：1.ZeroMQ2.1.7

–請勿使用2.1.10版本，因為該版本的一些嚴重bug會導致Storm集群運行時出現(xiàn)奇怪的問題。少數(shù)用戶在2.1.7版本會遇到”IllegalArgumentException”的異常，此時降為2.1.4版本可修復這一問題。2.

JZMQ3.Java64.Python2.6.65.unzip以上依賴庫的版本是經(jīng)過Storm測試的，Storm并不能保證在其他版本的Java或Python庫下可運行。安裝ZMQ2.1.7下載后編譯安裝ZMQ：wget/zeromq-2.1.7.tar.gztar-xzfzeromq-2.1.7.tar.gzcdzeromq-2.1.7./configuremakesudomakeinstall復制代碼

注意事項：如果安裝過程報錯uuid找不到，則通過如下的包安裝uuid庫：如果安裝過程報錯uuid找不到，則通過如下的包安裝uuid庫：sudoyuminstalle2fsprogslsudoyuminstalle2fsprogs-devel復制代碼安裝JZMQ下載后編譯安裝JZMQ：gitclone/nathanmarz/jzmq.gitcdjzmq./autogen.sh./configuremakesudomakeinstall復制代碼

為了保證JZMQ正常工作，可能需要完成以下配置：正確設置JAVA_HOME環(huán)境變量安裝Java開發(fā)包升級autoconf如果你是MacOSX，參考這里注意事項：如果運行./configure命令出現(xiàn)問題，參考這里。安裝Java61.下載并安裝JDK6，參考這里；2.配置JAVA_HOME環(huán)境變量；3.運行java、javac命令，測試java正常安裝。安裝Python2.6.61.下載Python2.6.6：wget[url]/ftp/python/2.6.6/Python-2.6.6.tar.bz2[/url]復制代碼2.編譯安裝Python2.6.6：tar–jxvfPython-2.6.6.tar.bz2cdPython-2.6.6./configuremakemakeinstall復制代碼

3.測試Python2.6.6：python-VPython2.6.6復制代碼

安裝unzip

1.如果使用RedHat系列Linux系統(tǒng)，執(zhí)行以下命令安裝unzip：yuminstallunzip復制代碼

2.如果使用Debian系列Linux系統(tǒng)，執(zhí)行以下命令安裝unzip：apt-getinstallunzip復制代碼3.2.3下載并解壓Storm發(fā)布版本下一步，需要在Nimbus和Supervisor機器上安裝Storm發(fā)行版本。1.下載Storm發(fā)行版本，推薦使用Storm0.8.1：wget/downloads/nathanmarz/storm/storm-0.8.1.zip復制代碼2.解壓到安裝目錄下：unzipstorm-0.8.1.zip復制代碼3.2.4修改storm.yaml配置文件

Storm發(fā)行版本解壓目錄下有一個conf/storm.yaml文件，用于配置Storm。默認配置在這里可以查看。conf/storm.yaml中的配置選項將覆蓋defaults.yaml中的默認配置。以下配置選項是必須在conf/storm.yaml中進行配置的：1)storm.zookeeper.servers:

Storm集群使用的Zookeeper集群地址，其格式如下：storm.zookeeper.servers:-“111.222.333.444″-“555.666.777.888″復制代碼如果Zookeeper集群使用的不是默認端口，那么還需要storm.zookeeper.port選項。2)

storm.local.dir:Nimbus和Supervisor進程用于存儲少量狀態(tài)，如jars、confs等的本地磁盤目錄，需要提前創(chuàng)建該目錄并給以足夠的訪問權限。然后在storm.yaml中配置該目錄，如：storm.local.dir:"/home/admin/storm/workdir"復制代碼

3)

java.library.path:Storm使用的本地庫（ZMQ和JZMQ）加載路徑，默認為”/usr/local/lib:/opt/local/lib:/usr/lib”，一般來說ZMQ和JZMQ默認安裝在/usr/local/lib下，因此不需要配置即可。4)

nimbus.host:Storm集群Nimbus機器地址，各個Supervisor工作節(jié)點需要知道哪個機器是Nimbus，以便下載Topologies的jars、confs等文件，如：nimbus.host:"111.222.333.444"復制代碼

5)

supervisor.slots.ports:對于每個Supervisor工作節(jié)點，需要配置該工作節(jié)點可以運行的worker數(shù)量。每個worker占用一個單獨的端口用于接收消息，該配置選項即用于定義哪些端口是可被worker使用的。默認情況下，每個節(jié)點上可運行4個workers，分別在6700、6701、6702和6703端口，如：supervisor.slots.ports:

-6700

-6701

-6702

-6703復制代碼3.2.5啟動Storm各個后臺進程最后一步，啟動Storm的所有后臺進程。和Zookeeper一樣，Storm也是快速失?。╢ail-fast)的系統(tǒng)，這樣Storm才能在任意時刻被停止，并且當進程重啟后被正確地恢復執(zhí)行。這也是為什么Storm不在進程內(nèi)保存狀態(tài)的原因，即使Nimbus或Supervisors被重啟，運行中的Topologies不會受到影響。以下是啟動Storm各個后臺進程的方式：Nimbus:在Storm主控節(jié)點上運行”bin/stormnimbus>/dev/null2>&1&”啟動Nimbus后臺程序，并放到后臺執(zhí)行；Supervisor:在Storm各個工作節(jié)點上運行”bin/stormsupervisor>/dev/null2>&1&”啟動Supervisor后臺程序，并放到后臺執(zhí)行；UI:在Storm主控節(jié)點上運行”bin/stormui>/dev/null2>&1&”啟動UI后臺程序，并放到后臺執(zhí)行，啟動后可以通過http://{nimbushost}:8080觀察集群的worker資源使用情況、Topologies的運行狀態(tài)等信息。注意事項：啟動Storm后臺進程時，需要對conf/storm.yaml配置文件中設置的storm.local.dir目錄具有寫權限。Storm后臺進程被啟動后，將在Storm安裝部署目錄下的logs/子目錄下生成各個進程的日志文件。經(jīng)測試，StormUI必須和StormNimbus部署在同一臺機器上，否則UI無法正常工作，因為UI進程會檢查本機是否存在Nimbus鏈接。為了方便使用，可以將bin/storm加入到系統(tǒng)環(huán)境變量中。至此，Storm集群已經(jīng)部署、配置完畢，可以向集群提交拓撲運行了。3.3向集群提交任務1.啟動StormTopology：stormjarallmycode.jarorg.me.MyTopologyarg1arg2arg3復制代碼其中，allmycode.jar是包含Topology實現(xiàn)代碼的jar包，org.me.MyTopology的main方法是Topology的入口，arg1、arg2和arg3為org.me.MyTopology執(zhí)行時需要傳入的參數(shù)。2.停止StormTopology：stormkill{toponame}其中，{toponame}為Topology提交到Storm集群時指定的Topology任務名稱。3.4參考資料1.

/nathanmarz/storm/wiki/Tutorial/nathanmarz/st...-up-a-Storm-cluster3./?p=1892第四章torm消息的可靠處理4.1簡介storm可以確保spout發(fā)送出來的每個消息都會被完整的處理。本章將會描述storm體系是如何達到這個目標的，并將會詳述開發(fā)者應該如何使用storm的這些機制來實現(xiàn)數(shù)據(jù)的可靠處理。4.2理解消息被完整處理一個消息(tuple)從spout發(fā)送出來，可能會導致成百上千的消息基于此消息被創(chuàng)建。我們來思考一下流式的“單詞統(tǒng)計”的例子：storm任務從數(shù)據(jù)源（Kestrelqueue）每次讀取一個完整的英文句子；將這個句子分解為獨立的單詞，最后，實時的輸出每個單詞以及它出現(xiàn)過的次數(shù)。本例中，每個從spout發(fā)送出來的消息（每個英文句子）都會觸發(fā)很多的消息被創(chuàng)建，那些從句子中分隔出來的單詞就是被創(chuàng)建出來的新消息。這些消息構成一個樹狀結構，我們稱之為“tupletree”，看起來如圖1所示：圖1示例tupletree在什么條件下，Storm才會認為一個從spout發(fā)送出來的消息被完整處理呢？答案就是下面的條件同時被滿足：tupletree不再生長樹中的任何消息被標識為“已處理”如果在指定的時間內(nèi)，一個消息衍生出來的tupletree未被完全處理成功，則認為此消息未被完整處理。這個超時值可以通過任務級參數(shù)Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS

進行配置，默認超時值為30秒。4.3消息的生命周期如果消息被完整處理或者未被完整處理，Storm會如何進行接下來的操作呢？為了弄清這個問題，我們來研究一下從spout發(fā)出來的消息的生命周期。這里列出了spout應該實現(xiàn)的接口：

首先，Storm使用spout實例的nextTuple()方法從spout請求一個消息（tuple）。收到請求以后，spout使用open方法中提供的SpoutOutputCollector向它的輸出流發(fā)送一個或多個消息。每發(fā)送一個消息，Spout會給這個消息提供一個messageID，它將會被用來標識這個消息。假設我們從kestrel隊列中讀取消息，Spout會將kestrel隊列為這個消息設置的ID作為此消息的messageID。向SpoutOutputCollector中發(fā)送消息格式如下：

接來下，這些消息會被發(fā)送到后續(xù)業(yè)務處理的bolts，并且Storm會跟蹤由此消息產(chǎn)生出來的新消息。當檢測到一個消息衍生出來的tupletree被完整處理后，Storm會調(diào)用Spout中的ack方法，并將此消息的messageID作為參數(shù)傳入。同理，如果某消息處理超時，則此消息對應的Spout的fail方法會被調(diào)用，調(diào)用時此消息的messageID會被作為參數(shù)傳入。

注意：一個消息只會由發(fā)送它的那個spout任務來調(diào)用ack或fail。如果系統(tǒng)中某個spout由多個任務運行，消息也只會由創(chuàng)建它的spout任務來應答（ack或fail），決不會由其他的spout任務來應答。

我們繼續(xù)使用從kestrel隊列中讀取消息的例子來闡述高可靠性下spout需要做些什么(假設這個spout的名字是KestrelSpout)。

我們先簡述一下kestrel消息隊列：當KestrelSpout從kestrel隊列中讀取一個消息，表示它“打開”了隊列中某個消息。這意味著，此消息并未從隊列中真正的刪除，而是將此消息設置為“pending”狀態(tài)，它等待來自客戶端的應答，被應答以后，此消息才會被真正的從隊列中刪除。處于“pending”狀態(tài)的消息不會被其他的客戶端看到。另外，如果一個客戶端意外的斷開連接，則由此客戶端“打開”的所有消息都會被重新加入到隊列中。當消息被“打開”的時候，kestrel隊列同時會為這個消息提供一個唯一的標識。KestrelSpout就是使用這個唯一的標識作為這個tuple的messageID的。稍后當ack或fail被調(diào)用的時候，KestrelSpout會把ack或者fail連同messageID一起發(fā)送給kestrel隊列，kestrel會將消息從隊列中真正刪除或者將它重新放回隊列中。4.4靠相關的API為了使用Storm提供的可靠處理特性，我們需要做兩件事情：

無論何時在tupletree中創(chuàng)建了一個新的節(jié)點，我們需要明確的通知Storm；當處理完一個單獨的消息時，我們需要告訴Storm這棵tupletree的變化狀態(tài)。通過上面的兩步，storm就可以檢測到一個tupletree何時被完全處理了，并且會調(diào)用相關的ack或fail方法。Storm提供了簡單明了的方法來完成上述兩步。為tupletree中指定的節(jié)點增加一個新的節(jié)點，我們稱之為錨定（anchoring）。錨定是在我們發(fā)送消息的同時進行的。為了更容易說明問題，我們使用下面代碼作為例子。本示例的bolt將包含整句話的消息分解為一系列的子消息，每個子消息包含一個單詞。

每個消息都通過這種方式被錨定：把輸入消息作為emit方法的第一個參數(shù)。因為word消息被錨定在了輸入消息上，這個輸入消息是spout發(fā)送過來的tupletree的根節(jié)點，如果任意一個word消息處理失敗，派生這個tupletree那個spout消息將會被重新發(fā)送。與此相反，我們來看看使用下面的方式emit消息時，Storm會如何處理：

如果以這種方式發(fā)送消息，將會導致這個消息不會被錨定。如果此tupletree中的消息處理失敗，派生此tupletree的根消息不會被重新發(fā)送。根據(jù)任務的容錯級別，有時候很適合發(fā)送一個非錨定的消息。一個輸出消息可以被錨定在一個或者多個輸入消息上，這在做join或聚合的時候是很有用的。一個被多重錨定的消息處理失敗，會導致與之關聯(lián)的多個spout消息被重新發(fā)送。多重錨定通過在emit方法中指定多個輸入消息來實現(xiàn)：多重錨定會將被錨定的消息加到多棵tupletree上。注意：多重綁定可能會破壞傳統(tǒng)的樹形結構，從而構成一個DAGs（有向無環(huán)圖），如圖2所示：

圖2多重錨定構成的鉆石型結構Storm的實現(xiàn)可以像處理樹那樣來處理DAGs。錨定表明了如何將一個消息加入到指定的tupletree中，高可靠處理API的接下來部分將向您描述當處理完tupletree中一個單獨的消息時我們該做些什么。這些是通過OutputCollector的ack和fail方法來實現(xiàn)的?；仡^看一下例子SplitSentence，可以發(fā)現(xiàn)當所有的word消息被發(fā)送完成后，輸入的表示句子的消息會被應答（acked）。每個被處理的消息必須表明成功或失?。╝cked或者failed）。Storm是使用內(nèi)存來跟蹤每個消息的處理情況的，如果被處理的消息沒有應答的話，遲早內(nèi)存會被耗盡！很多bolt遵循特定的處理流程：讀取一個消息、發(fā)送它派生出來的子消息、在execute結尾處應答此消息。一般的過濾器（filter）或者是簡單的處理功能都是這類的應用。Storm有一個BasicBolt接口封裝了上述的流程。示例SplitSentence可以使用BasicBolt來重寫：

使用這種方式，代碼比之前稍微簡單了一些，但是實現(xiàn)的功能是一樣的。發(fā)送到BasicOutputCollector的消息會被自動的錨定到輸入消息，并且，當execute執(zhí)行完畢的時候，會自動的應答輸入消息。很多情況下，一個消息需要延遲應答，例如聚合或者是join。只有根據(jù)一組輸入消息得到一個結果之后，才會應答之前所有的輸入消息。并且聚合和join大部分時候對輸出消息都是多重錨定。然而，這些特性不是IBasicBolt所能處理的。4.5高效的實現(xiàn)tupletreeStorm系統(tǒng)中有一組叫做“acker”的特殊的任務，它們負責跟蹤DAG（有向無環(huán)圖）中的每個消息。每當發(fā)現(xiàn)一個DAG被完全處理，它就向創(chuàng)建這個根消息的spout任務發(fā)送一個信號。拓撲中acker任務的并行度可以通過配置參數(shù)Config.TOPOLOGY_ACKERS來設置。默認的acker任務并行度為1，當系統(tǒng)中有大量的消息時，應該適當提高acker任務的并發(fā)度。為了理解Storm可靠性處理機制，我們從研究一個消息的生命周期和tupletree的管理入手。當一個消息被創(chuàng)建的時候（無論是在spout還是bolt中），系統(tǒng)都為該消息分配一個64bit的隨機值作為id。這些隨機的id是acker用來跟蹤由spout消息派生出來的tupletree的。每個消息都知道它所在的tupletree對應的根消息的id。每當bolt新生成一個消息，對應tupletree中的根消息的messageId就拷貝到這個消息中。當這個消息被應答的時候，它就把關于tupletree變化的信息發(fā)送給跟蹤這棵樹的acker。例如，他會告訴acker：本消息已經(jīng)處理完畢，但是我派生出了一些新的消息，幫忙跟蹤一下吧。舉個例子，假設消息D和E是由消息C派生出來的，這里演示了消息C被應答時，tupletree是如何變化的。

因為在C被從樹中移除的同時D和E會被加入到tupletree中，因此tupletree不會被過早的認為已完全處理。關于Storm如何跟蹤tupletree，我們再深入的探討一下。前面說過系統(tǒng)中可以有任意個數(shù)的acker，那么，每當一個消息被創(chuàng)建或應答的時候，它怎么知道應該通知哪個acker呢？系統(tǒng)使用一種哈希算法來根據(jù)spout消息的messageId確定由哪個acker跟蹤此消息派生出來的tupletree。因為每個消息都知道與之對應的根消息的messageId，因此它知道應該與哪個acker通信。當spout發(fā)送一個消息的時候，它就通知對應的acker一個新的根消息產(chǎn)生了，這時acker就會創(chuàng)建一個新的tupletree。當acker發(fā)現(xiàn)這棵樹被完全處理之后，他就會通知對應的spout任務。tuple是如何被跟蹤的呢？系統(tǒng)中有成千上萬的消息，如果為每個spout發(fā)送的消息都構建一棵樹的話，很快內(nèi)存就會耗盡。所以，必須采用不同的策略來跟蹤每個消息。由于使用了新的跟蹤算法，Storm只需要固定的內(nèi)存（大約20字節(jié)）就可以跟蹤一棵樹。這個算法是storm正確運行的核心，也是storm最大的突破。acker任務保存了spout消息id到一對值的映射。第一個值就是spout的任務id，通過這個id，acker就知道消息處理完成時該通知哪個spout任務。第二個值是一個64bit的數(shù)字，我們稱之為“ackval”，它是樹中所有消息的隨機id的異或結果。ackval表示了整棵樹的的狀態(tài)，無論這棵樹多大，只需要這個固定大小的數(shù)字就可以跟蹤整棵樹。當消息被創(chuàng)建和被應答的時候都會有相同的消息id發(fā)送過來做異或。每當acker發(fā)現(xiàn)一棵樹的ackval值為0的時候，它就知道這棵樹已經(jīng)被完全處理了。因為消息的隨機ID是一個64bit的值，因此ackval在樹處理完之前被置為0的概率非常小。假設你每秒鐘發(fā)送一萬個消息，從概率上說，至少需要50,000,000年才會有機會發(fā)生一次錯誤。即使如此，也只有在這個消息確實處理失敗的情況下才會有數(shù)據(jù)的丟失！4.6選擇合適的可靠性級別Acker任務是輕量級的，所以在拓撲中并不需要太多的acker存在。可以通過StormUI來觀察acker任務的吞吐量，如果看上去吞吐量不夠的話，說明需要添加額外的acker。如果你并不要求每個消息必須被處理（你允許在處理過程中丟失一些信息），那么可以關閉消息的可靠處理機制，從而可以獲取較好的性能。關閉消息的可靠處理機制意味著系統(tǒng)中的消息數(shù)會減半（每個消息不需要應答了）。另外，關閉消息的可靠處理可以減少消息的大?。ú恍枰總€tuple記錄它的根id了），從而節(jié)省帶寬。有三種方法可以關系消息的可靠處理機制：將參數(shù)Config.TOPOLOGY_ACKERS設置為0，通過此方法，當Spout發(fā)送一個消息的時候，它的ack方法將立刻被調(diào)用；第二個方法是Spout發(fā)送一個消息時，不指定此消息的messageID。當需要關閉特定消息可靠性的時候，可以使用此方法；最后，如果你不在意某個消息派生出來的子孫消息的可靠性，則此消息派生出來的子消息在發(fā)送時不要做錨定，即在emit方法中不指定輸入消息。因為這些子孫消息沒有被錨定在任何tupletree中，因此他們的失敗不會引起任何spout重新發(fā)送消息。4.7集群的各級容錯到現(xiàn)在為止，大家已經(jīng)理解了Storm的可靠性機制，并且知道了如何選擇不同的可靠性級別來滿足需求。接下來我們研究一下Storm如何保證在各種情況下確保數(shù)據(jù)不丟失。4.7.1任務級失敗因為bolt任務crash引起的消息未被應答。此時，acker中所有與此bolt任務關聯(lián)的消息都會因為超時而失敗，對應spout的fail方法將被調(diào)用。acker任務失敗。如果acker任務本身失敗了，它在失敗之前持有的所有消息都將會因為超時而失敗。Spout的fail方法將被調(diào)用。Spout任務失敗。這種情況下，Spout任務對接的外部設備（如MQ）負責消息的完整性。例如當客戶端異常的情況下，kestrel隊列會將處于pending狀態(tài)的所有的消息重新放回到隊列中。4.7.2

任務槽(slot)故障worker失敗。每個worker中包含數(shù)個bolt（或spout）任務。supervisor負責監(jiān)控這些任務，當worker失敗后，supervisor會嘗試在本機重啟它。supervisor失敗。supervisor是無狀態(tài)的，因此supervisor的失敗不會影響當前正在運行的任務，只要及時的將它重新啟動即可。supervisor不是自舉的，需要外部監(jiān)控來及時重啟。nimbus失敗。nimbus是無狀態(tài)的，因此nimbus的失敗不會影響當前正在運行的任務（nimbus失敗時，無法提交新的任務），只要及時的將它重新啟動即可。nimbus不是自舉的，需要外部監(jiān)控來及時重啟。4.7.3.

集群節(jié)點（機器）故障storm集群中的節(jié)點故障。此時nimbus會將此機器上所有正在運行的任務轉移到其他可用的機器上運行。zookeeper集群中的節(jié)點故障。zookeeper保證少于半數(shù)的機器宕機仍可正常運行，及時修復故障機器即可。4.8小結本章介紹了storm集群如何實現(xiàn)數(shù)據(jù)的可靠處理。借助于創(chuàng)新性的tupletree跟蹤技術，storm高效的通過數(shù)據(jù)的應答機制來保證數(shù)據(jù)不丟失。storm集群中除nimbus外，沒有單點存在，任何節(jié)點都可以出故障而保證數(shù)據(jù)不會丟失。nimbus被設計為無狀態(tài)的，只要可以及時重啟，就不會影響正在運行的任務。第五章一致性事務Storm是一個分布式的流處理系統(tǒng)，利用anchor和ack機制保證所有tuple都被成功處理。如果tuple出錯，則可以被重傳，但是如何保證出錯的tuple只被處理一次呢？Storm提供了一套事務性組件TransactionTopology，用來解決這個問題。TransactionalTopology目前已經(jīng)不再維護，由Trident來實現(xiàn)事務性topology，但是原理相同。5.1一致性事務的設計Storm如何實現(xiàn)即對tuple并行處理，又保證事務性。本節(jié)從簡單的事務性實現(xiàn)方法入手，逐步引出TransactionalTopology的原理。保證tuple只被處理一次，最簡單的方法就是將tuple流變成強順序的，并且每次只處理一個tuple。從1開始，給每個tuple都順序加上一個id。在處理tuple的時候，將處理成功的tupleid和計算結果存在數(shù)據(jù)庫中。下一個tuple到來的時候，將其id與數(shù)據(jù)庫中的id做比較。如果相同，則說明這個tuple已經(jīng)被成功處理過了，忽略它；如果不同，根據(jù)強順序性，說明這個tuple沒有被處理過，將它的id及計算結果更新到數(shù)據(jù)庫中。以統(tǒng)計消息總數(shù)為例。每來一個tuple，如果數(shù)據(jù)庫中存儲的id與當前tupleid不同，則數(shù)據(jù)庫中的消息總數(shù)加1，同時更新數(shù)據(jù)庫中的當前tupleid值。如圖：

但是這種機制使得系統(tǒng)一次只能處理一個tuple，無法實現(xiàn)分布式計算。為了實現(xiàn)分布式，我們可以每次處理一批tuple，稱為一個batch。一個batch中的tuple可以被并行處理。我們要保證一個batch只被處理一次，機制和上一節(jié)類似。只不過數(shù)據(jù)庫中存儲的是batchid。batch的中間計算結果先存在局部變量中，當一個batch中的所有tuple都被處理完之后，判斷batchid，如果跟數(shù)據(jù)庫中的id不同，則將中間計算結果更新到數(shù)據(jù)庫中。如何確保一個batch里面的所有tuple都被處理完了呢？可以利用Storm提供的CoordinateBolt。如圖：

但是強順序batch流也有局限，每次只能處理一個batch，batch之間無法并行。要想實現(xiàn)真正的分布式事務處理，可以使用storm提供的TransactionalTopology。在此之前，我們先詳細介紹一下CoordinateBolt的原理。CoordinateBolt具體原理如下：真正執(zhí)行計算的bolt外面封裝了一個CoordinateBolt。真正執(zhí)行任務的bolt我們稱為realbolt。每個CoordinateBolt記錄兩個值：有哪些task給我發(fā)送了tuple(根據(jù)topology的grouping信息)；我要給哪些tuple發(fā)送信息(同樣根據(jù)groping信息)Realbolt發(fā)出一個tuple后，其外層的CoordinateBolt會記錄下這個tuple發(fā)送給哪個task了。等所有的tuple都發(fā)送完了之后，CoordinateBolt通過另外一個特殊的stream以emitDirect的方式告訴所有它發(fā)送過tuple的task，它發(fā)送了多少tuple給這個task。下游task會將這個數(shù)字和自己已經(jīng)接收到的tuple數(shù)量做對比，如果相等，則說明處理完了所有的tuple。下游CoordinateBolt會重復上面的步驟，通知其下游。整個過程如圖所示：

CoordinateBolt主要用于兩個場景：DRPCTransactionalTopologyCoordinatedBolt對于業(yè)務是有侵入的，要使用CoordinatedBolt提供的功能，你必須要保證你的每個bolt發(fā)送的每個tuple的第一個field是request-id。所謂的“我已經(jīng)處理完我的上游”的意思是說當前這個bolt對于當前這個request-id所需要做的工作做完了。這個request-id在DRPC里面代表一個DRPC請求；在TransactionalTopology里面代表一個batch。Storm提供的TransactionalTopology將batch計算分為process和commit兩個階段。Process階段可以同時處理多個batch，不用保證順序性；commit階段保證batch的強順序性，并且一次只能處理一個batch，第1個batch成功提交之前，第2個batch不能被提交。還是以統(tǒng)計消息總數(shù)為例，以下代碼來自storm-starter里面的TransactionalGlobalCount。MemoryTransactionalSpoutspout=newMemoryTransactionalSpout(DATA,newFields(“word“),PARTITION_TAKE_PER_BATCH);TransactionalTopologyBuilderbuilder=newTransactionalTopologyBuilder(“global-count“,“spout“,spout,3);builder.setBolt(“partial-count“,newBatchCount(),5).noneGrouping(“spout“);builder.setBolt(“sum“,newUpdateGlobalCount()).globalGrouping(“partial-count“);TransactionalTopologyBuilder共接收四個參數(shù)。這個TransactionalTopology的id。Id用來在Zookeeper中保存當前topology的進度，如果這個topology重啟，可以繼續(xù)之前的進度執(zhí)行。Spout在這個topology中的id一個TransactionalSpout