storm入門教程資料

storm入門教程第一章storm概述

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

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

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

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

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

Storm有如下特點(diǎn)：

編程模型簡單

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

可擴(kuò)展

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

因此，計(jì)算任務(wù)在多個線程、進(jìn)程和服務(wù)器之間并行進(jìn)行，支持靈活的水平擴(kuò)展。

高可靠性

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

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

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

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

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

高容錯性

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

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

支持多種編程語言

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

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

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

支持本地模式

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

高效

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

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

1)構(gòu)建maven環(huán)境：為了開發(fā)stormtopology,你需要把storm相關(guān)的jar包添加到classpath里面去：要么手動添加所有相關(guān)的jar包，要么使用maven來管理所有的依賴。storm的jar包發(fā)布在Clojars(一個maven庫),如果你使用maven的話，把下面的配置添加在你項(xiàng)目的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從句子隊(duì)列中讀取句子，在位于一個Kestrel的服務(wù)器端口22133。Spout用setSpout方法插入一個獨(dú)特的id到topology。Topology中的每個節(jié)點(diǎn)必須給予一個id，id是由其他bolts用于訂閱該節(jié)點(diǎn)的輸出流。KestrelSpout在topology中id為1。setBolt是用于在Topology中插入bolts。在topology中定義的第一個bolts是切割句子的bolts。這個bolts將句子流轉(zhuǎn)成成單詞流。讓我們看看SplitSentence實(shí)施：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”));}關(guān)鍵的方法是execute方法。正如你可以看到，它將句子拆分成單詞，并發(fā)出每個單詞作為一個新的元組。另一個重要的方法是declareOutputFields，其中宣布bolts輸出元組的架構(gòu)。在這里宣布，它發(fā)出一個域?yàn)閣ord的元組setBolt的最后一個參數(shù)是你想為bolts的并行量。SplitSentencebolts是10個并發(fā)，這將導(dǎo)致在storm集群中有十個線程并行執(zhí)行。你所要做的的是增加bolts的并行量在遇到topology的瓶頸時。setBolt方法返回一個對象，用來定義bolts的輸入。例如，SplitSentence螺栓訂閱組件“1”使用隨機(jī)分組的輸出流?！?”是指已經(jīng)定義KestrelSpout。我將解釋在某一時刻的隨機(jī)分組的一部分。到目前為止，最要緊的是，SplitSentencebolts會消耗KestrelSpout發(fā)出的每一個元組。下面在讓我們看看wordcount的實(shí)現(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)存里面維護(hù)一個單詞->次數(shù)的mapping，WordCount每收到一個單詞，它就更新內(nèi)存里面的統(tǒng)計(jì)狀態(tài)。storm的運(yùn)行有兩種模式:本地模式和分布式模式.1)本地模式：storm用一個進(jìn)程里面的線程來模擬所有的spout和bolt.本地模式對開發(fā)和測試來說比較有用。你運(yùn)行storm-starter里面的topology的時候它們就是以本地模式運(yùn)行的，你可以看到topology里面的每一個組件在發(fā)射什么消息。2)分布式模式：storm由一堆機(jī)器組成。當(dāng)你提交topology給master的時候，你同時也把topology的代碼提交了。master負(fù)責(zé)分發(fā)你的代碼并且負(fù)責(zé)給你的topolgoy分配工作進(jìn)程。如果一個工作進(jìn)程掛掉了，master節(jié)點(diǎn)會把認(rèn)為重新分配到其它節(jié)點(diǎn)。3)下面是以本地模式運(yùn)行的代碼: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對象來定義一個進(jìn)程內(nèi)的集群。提交topology給這個虛擬的集群和提交topology給分布式集群是一樣的。通過調(diào)用submitTopology方法來提交topology，它接受三個參數(shù)：要運(yùn)行的topology的名字，一個配置對象以及要運(yùn)行的topology本身。topology的名字是用來唯一區(qū)別一個topology的，這樣你然后可以用這個名字來殺死這個topology的。前面已經(jīng)說過了，你必須顯式的殺掉一個topology，否則它會一直運(yùn)行。Conf對象可以配置很多東西，下面兩個是最常見的：TOPOLOGY_WORKERS(setNumWorkers)定義你希望集群分配多少個工作進(jìn)程給你來執(zhí)行這個topology.topology里面的每個組件會被需要線程來執(zhí)行。每個組件到底用多少個線程是通過setBolt和setSpout來指定的。這些線程都運(yùn)行在工作進(jìn)程里面.每一個工作進(jìn)程包含一些節(jié)點(diǎn)的一些工作線程。比如，如果你指定300個線程，60個進(jìn)程，那么每個工作進(jìn)程里面要執(zhí)行6個線程，而這6個線程可能屬于不同的組件(Spout,Bolt)。你可以通過調(diào)整每個組件的并行度以及這些線程所在的進(jìn)程數(shù)量來調(diào)整topology的性能。TOPOLOGY_DEBUG(setDebug),當(dāng)它被設(shè)置成true的話，storm會記錄下每個組件所發(fā)射的每條消息。這在本地環(huán)境調(diào)試topology很有用，但是在線上這么做的話會影響性能的。結(jié)論：本章從storm的基本對象的定義，到廣泛的介紹了storm的開發(fā)環(huán)境，從一個簡單的例子講解了topology的構(gòu)建和定義。希望大家可以從本章的內(nèi)容對storm有一個基本的理解和概念，并且已經(jīng)可以構(gòu)建一個簡單的topology??！第三章Storm安裝部署步驟本文以TwitterStorm官方Wiki為基礎(chǔ)，詳細(xì)描述如何快速搭建一個Storm集群，其中，項(xiàng)目實(shí)踐中遇到的問題及經(jīng)驗(yàn)總結(jié)，在相應(yīng)章節(jié)以“注意事項(xiàng)”的形式給出。3.1Storm集群組件Storm集群中包含兩類節(jié)點(diǎn)：主控節(jié)點(diǎn)（MasterNode）和工作節(jié)點(diǎn)（WorkNode）。其分別對應(yīng)的角色如下：1.主控節(jié)點(diǎn)（MasterNode）上運(yùn)行一個被稱為Nimbus的后臺程序，它負(fù)責(zé)在Storm集群內(nèi)分發(fā)代碼，分配任務(wù)給工作機(jī)器，并且負(fù)責(zé)監(jiān)控集群運(yùn)行狀態(tài)。Nimbus的作用類似于Hadoop中JobTracker的角色。2.每個工作節(jié)點(diǎn)（WorkNode）上運(yùn)行一個被稱為Supervisor的后臺程序。Supervisor負(fù)責(zé)監(jiān)聽從Nimbus分配給它執(zhí)行的任務(wù)，據(jù)此啟動或停止執(zhí)行任務(wù)的工作進(jìn)程。每一個工作進(jìn)程執(zhí)行一個Topology的子集；一個運(yùn)行中的Topology由分布在不同工作節(jié)點(diǎn)上的多個工作進(jìn)程組成。

Storm集群組件Nimbus和Supervisor節(jié)點(diǎn)之間所有的協(xié)調(diào)工作是通過Zookeeper集群來實(shí)現(xiàn)的。此外，Nimbus和Supervisor進(jìn)程都是快速失?。╢ail-fast)和無狀態(tài)（stateless）的；Storm集群所有的狀態(tài)要么在Zookeeper集群中，要么存儲在本地磁盤上。這意味著你可以用kill-9來殺死Nimbus和Supervisor進(jìn)程，它們在重啟后可以繼續(xù)工作。這個設(shè)計(jì)使得Storm集群擁有不可思議的穩(wěn)定性。3.2安裝Storm集群這一章節(jié)將詳細(xì)描述如何搭建一個Storm集群。下面是接下來需要依次完成的安裝步驟：1.搭建Zookeeper集群；2.安裝Storm依賴庫；3.下載并解壓Storm發(fā)布版本；4.修改storm.yaml配置文件；5.啟動Storm各個后臺進(jìn)程。3.2.1搭建Zookeeper集群Storm使用Zookeeper協(xié)調(diào)集群，由于Zookeeper并不用于消息傳遞，所以Storm給Zookeeper帶來的壓力相當(dāng)?shù)汀４蠖鄶?shù)情況下，單個節(jié)點(diǎn)的Zookeeper集群足夠勝任，不過為了確保故障恢復(fù)或者部署大規(guī)模Storm集群，可能需要更大規(guī)模節(jié)點(diǎn)的Zookeeper集群（對于Zookeeper集群的話，官方推薦的最小節(jié)點(diǎn)數(shù)為3個）。在Zookeeper集群的每臺機(jī)器上完成以下安裝部署步驟：1.下載安裝JavaJDK，官方下載鏈接為/javase/downloads/index.jsp，JDK版本為JDK6或以上。2.根據(jù)Zookeeper集群的負(fù)載情況，合理設(shè)置Java堆大小，盡可能避免發(fā)生swap，導(dǎo)致Zookeeper性能下降。保守起見，4GB內(nèi)存的機(jī)器可以為Zookeeper分配3GB最大堆空間。3.下載后解壓安裝Zookeeper包，官方下載鏈接為/zookeeper/releases.html。4.根據(jù)Zookeeper集群節(jié)點(diǎn)情況，在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復(fù)制代碼其中，dataDir指定Zookeeper的數(shù)據(jù)文件目錄；其中server.id=host:port:port，id是為每個Zookeeper節(jié)點(diǎn)的編號，保存在dataDir目錄下的myid文件中，zoo1~zoo3表示各個Zookeeper節(jié)點(diǎn)的hostname，第一個port是用于連接leader的端口，第二個port是用于leader選舉的端口。5.在dataDir目錄下創(chuàng)建myid文件，文件中只包含一行，且內(nèi)容為該節(jié)點(diǎn)對應(yīng)的server.id中的id編號。6.啟動Zookeeper服務(wù)：java-cpzookeeper.jar:lib/log4j-1.2.15.jar:conf\org.apache.zookeeper.server.quorum.QuorumPeerMainzoo.cfg復(fù)制代碼或者bin/zkServer.shstart復(fù)制代碼7.通過Zookeeper客戶端測試服務(wù)是否可用：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復(fù)制代碼或者bin/zkCli.sh-server:2181復(fù)制代碼

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

–請勿使用2.1.10版本，因?yàn)樵摪姹镜囊恍﹪?yán)重bug會導(dǎo)致Storm集群運(yùn)行時出現(xiàn)奇怪的問題。少數(shù)用戶在2.1.7版本會遇到”IllegalArgumentException”的異常，此時降為2.1.4版本可修復(fù)這一問題。2.

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

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

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

3.測試Python2.6.6：python-VPython2.6.6復(fù)制代碼

安裝unzip

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

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

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

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

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

3)

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

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

5)

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

-6700

-6701

-6702

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

但是強(qiáng)順序batch流也有局限，每次只能處理一個batch，batch之間無法并行。要想實(shí)現(xiàn)真正的分布式事務(wù)處理，可以使用storm提供的TransactionalTopology。在此之前，我們先詳細(xì)介紹一下CoordinateBolt的原理。CoordinateBolt具體原理如下：真正執(zhí)行計(jì)算的bolt外面封裝了一個CoordinateBolt。真正執(zhí)行任務(wù)的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會重復(fù)上面的步驟，通知其下游。整個過程如圖所示：

CoordinateBolt主要用于兩個場景：DRPCTransactionalTopologyCoordinatedBolt對于業(yè)務(wù)是有侵入的，要使用CoordinatedBolt提供的功能，你必須要保證你的每個bolt發(fā)送的每個tuple的第一個field是request-id。所謂的“我已經(jīng)處理完我的上游”的意思是說當(dāng)前這個bolt對于當(dāng)前這個request-id所需要做的工作做完了。這個request-id在DRPC里面代表一個DRPC請求；在TransactionalTopology里面代表一個batch。Storm提供的TransactionalTopology將batch計(jì)算分為process和commit兩個階段。Process階段可以同時處理多個batch，不用保證順序性；commit階段保證batch的強(qiáng)順序性，并且一次只能處理一個batch，第1個batch成功提交之前，第2個batch不能被提交。還是以統(tǒng)計(jì)消息總數(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中保存當(dāng)前topology的進(jìn)度，如果這個topology重啟，可以繼續(xù)之前的進(jìn)度執(zhí)行。Spout在這個topology中的id一個TransactionalSpout