Hadoop常用于離線的復(fù)雜的大數(shù)據(jù)處理，Spark常用于離線的快速(輕量級(jí))的大數(shù)據(jù)處理， Storm常用于在線的實(shí)時(shí)的大數(shù)據(jù)處理；這句話一定程度上反應(yīng)了三套大數(shù)據(jù)平臺(tái)的鮮明特征。Storm是一套實(shí)時(shí)、在線、分布式的大數(shù)據(jù)處理平臺(tái)。

1）Nimbus和Supervisor

Storm集群中包含兩類(lèi)節(jié)點(diǎn)：主控節(jié)點(diǎn)（Master Node）和工作節(jié)點(diǎn)（Work Node），角色定位如下：

主控節(jié)點(diǎn)（MasterNode）上運(yùn)行一個(gè)被稱(chēng)為Nimbus的后臺(tái)程序，它負(fù)責(zé)在Storm集群內(nèi)分發(fā)代碼，分配任務(wù)給工作機(jī)器，并且負(fù)責(zé)監(jiān)控集群運(yùn)行狀態(tài)。Nimbus的作用類(lèi)似于Hadoop中JobTracker的角色。

每個(gè)工作節(jié)點(diǎn)（WorkNode）上運(yùn)行一個(gè)被稱(chēng)為Supervisor的后臺(tái)程序。Supervisor負(fù)責(zé)監(jiān)聽(tīng)從Nimbus分配給它執(zhí)行的任務(wù)，據(jù)此啟動(dòng)或停止執(zhí)行任務(wù)的工作進(jìn)程。每一個(gè)工作進(jìn)程執(zhí)行一個(gè)Topology的子集；一個(gè)運(yùn)行中的Topology由分布在不同工作節(jié)點(diǎn)上的多個(gè)工作進(jìn)程組成。

Nimbus和Supervisor節(jié)點(diǎn)之間所有的協(xié)調(diào)工作是通過(guò)Zookeeper集群來(lái)實(shí)現(xiàn)的。此外，Nimbus和Supervisor進(jìn)程都是快速失敗（fail-fast)和無(wú)狀態(tài)（stateless）的；Storm集群所有的狀態(tài)要么在Zookeeper集群中，要么存儲(chǔ)在本地磁盤(pán)上。這意味著你可以用kill -9來(lái)殺死Nimbus和Supervisor進(jìn)程，它們?cè)谥貑⒑罂梢岳^續(xù)工作。這個(gè)設(shè)計(jì)使得Storm集群擁有不可思議的穩(wěn)定性。

2）Topology

一個(gè)topology是spouts和bolts組成的圖， 通過(guò)streamgroupings將圖中的spouts和bolts連接起來(lái)，如下圖：

一個(gè)topology會(huì)一直運(yùn)行直到你手動(dòng)kill掉，Storm自動(dòng)重新分配執(zhí)行失敗的任務(wù)， 并且Storm可以保證你不會(huì)有數(shù)據(jù)丟失（如果開(kāi)啟了高可靠性的話）。如果一些機(jī)器意外停機(jī)它上面的所有任務(wù)會(huì)被轉(zhuǎn)移到其他機(jī)器上。

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

storm jarall-my-code.jar backtype.storm.MyTopology arg1 arg2

這個(gè)命令會(huì)運(yùn)行主類(lèi): backtype.strom.MyTopology, 參數(shù)是arg1, arg2。這個(gè)類(lèi)的main函數(shù)定義這個(gè)topology并且把它提交給Nimbus。storm jar負(fù)責(zé)連接到Nimbus并且上傳jar包。Topology的定義是一個(gè)Thrift結(jié)構(gòu)，并且Nimbus就是一個(gè)Thrift服務(wù)，可以提交由任何語(yǔ)言創(chuàng)建的topology。

3）Stream

消息流stream是storm里的關(guān)鍵抽象。一個(gè)消息流是一個(gè)沒(méi)有邊界的tuple序列， 而這些tuple序列會(huì)以一種分布式的方式并行地創(chuàng)建和處理。通過(guò)對(duì)stream中tuple序列中每個(gè)字段命名來(lái)定義stream。在默認(rèn)的情況下，tuple的字段類(lèi)型可以是：integer，long，short， byte，string，double，float，boolean和bytearray。你也可以自定義類(lèi)型（只要實(shí)現(xiàn)相應(yīng)的序列化器）。

每個(gè)消息流在定義的時(shí)候會(huì)被分配給一個(gè)id，因?yàn)閱蜗蛳⒘魇褂玫南喈?dāng)普遍， OutputFieldsDeclarer定義了一些方法讓你可以定義一個(gè)stream而不用指定這個(gè)id。在這種情況下這個(gè)stream會(huì)分配個(gè)值為‘default’默認(rèn)的id 。

Storm提供的最基本的處理stream的原語(yǔ)是spout和bolt。你可以實(shí)現(xiàn)spout和bolt提供的接口來(lái)處理你的業(yè)務(wù)邏輯。

4）Spouts

消息源spout是Storm里面一個(gè)topology里面的消息生產(chǎn)者。一般來(lái)說(shuō)消息源會(huì)從一個(gè)外部源讀取數(shù)據(jù)并且向topology里面發(fā)出消息：tuple。Spout可以是可靠的也可以是不可靠的。如果這個(gè)tuple沒(méi)有被storm成功處理，可靠的消息源spouts可以重新發(fā)射一個(gè)tuple， 但是不可靠的消息源spouts一旦發(fā)出一個(gè)tuple就不能重發(fā)了。

消息源可以發(fā)射多條消息流stream。使用OutputFieldsDeclarer.declareStream來(lái)定義多個(gè)stream，然后使用SpoutOutputCollector來(lái)發(fā)射指定的stream。

Spout類(lèi)里面最重要的方法是nextTuple。要么發(fā)射一個(gè)新的tuple到topology里面或者簡(jiǎn)單的返回如果已經(jīng)沒(méi)有新的tuple。要注意的是nextTuple方法不能阻塞，因?yàn)閟torm在同一個(gè)線程上面調(diào)用所有消息源spout的方法。

另外兩個(gè)比較重要的spout方法是ack和fail。storm在檢測(cè)到一個(gè)tuple被整個(gè)topology成功處理的時(shí)候調(diào)用ack，否則調(diào)用fail。storm只對(duì)可靠的spout調(diào)用ack和fail。

5）Bolts

所有的消息處理邏輯被封裝在bolts里面。Bolts可以做很多事情：過(guò)濾，聚合，查詢(xún)數(shù)據(jù)庫(kù)等等。

Bolts可以簡(jiǎn)單的做消息流的傳遞。復(fù)雜的消息流處理往往需要很多步驟，從而也就需要經(jīng)過(guò)很多bolts。比如算出一堆圖片里面被轉(zhuǎn)發(fā)最多的圖片就至少需要兩步：第一步算出每個(gè)圖片的轉(zhuǎn)發(fā)數(shù)量。第二步找出轉(zhuǎn)發(fā)最多的前10個(gè)圖片。（如果要把這個(gè)過(guò)程做得更具有擴(kuò)展性那么可能需要更多的步驟）。

Bolts可以發(fā)射多條消息流， 使用OutputFieldsDeclarer.declareStream定義stream，使用OutputCollector.emit來(lái)選擇要發(fā)射的stream。

Bolts的主要方法是execute,它以一個(gè)tuple作為輸入，bolts使用OutputCollector來(lái)發(fā)射tuple，bolts必須要為它處理的每一個(gè)tuple調(diào)用OutputCollector的ack方法，以通知Storm這個(gè)tuple被處理完成了，從而通知這個(gè)tuple的發(fā)射者spouts。 一般的流程是： bolts處理一個(gè)輸入tuple,? 發(fā)射0個(gè)或者多個(gè)tuple, 然后調(diào)用ack通知storm自己已經(jīng)處理過(guò)這個(gè)tuple了。storm提供了一個(gè)IBasicBolt會(huì)自動(dòng)調(diào)用ack。

6）Stream groupings

義一個(gè)topology的其中一步是定義每個(gè)bolt接收什么樣的流作為輸入。stream grouping就是用來(lái)定義一個(gè)stream應(yīng)該如何分配數(shù)據(jù)給bolts上面的多個(gè)tasks。

Storm里面有7種類(lèi)型的stream grouping

ü?? ShuffleGrouping: 隨機(jī)分組，隨機(jī)派發(fā)stream里面的tuple，保證每個(gè)bolt接收到的tuple數(shù)目大致相同。

ü?? FieldsGrouping：按字段分組，比如按userid來(lái)分組， 具有同樣userid的tuple會(huì)被分到相同的Bolts里的一個(gè)task， 而不同的userid則會(huì)被分配到不同的bolts里的task。

ü?? All Grouping：廣播發(fā)送，對(duì)于每一個(gè)tuple，所有的bolts都會(huì)收到。

ü?? GlobalGrouping：全局分組，這個(gè)tuple被分配到storm中的一個(gè)bolt的其中一個(gè)task。再具體一點(diǎn)就是分配給id值最低的那個(gè)task。

ü?? Non Grouping：不分組，這個(gè)分組的意思是說(shuō)stream不關(guān)心到底誰(shuí)會(huì)收到它的tuple。目前這種分組和Shuffle grouping是一樣的效果， 有一點(diǎn)不同的是storm會(huì)把這個(gè)bolt放到這個(gè)bolt的訂閱者同一個(gè)線程里面去執(zhí)行。

ü?? DirectGrouping： 直接分組，這是一種比較特別的分組方法，用這種分組意味著消息的發(fā)送者指定由消息接收者的哪個(gè)task處理這個(gè)消息。 只有被聲明為Direct Stream的消息流可以聲明這種分組方法。而且這種消息tuple必須使用emitDirect方法來(lái)發(fā)射。消息處理者可以通過(guò)TopologyContext來(lái)獲取處理它的消息的task的id （OutputCollector.emit方法也會(huì)返回task的id）。

ü?? Local orshuffle grouping：如果目標(biāo)bolt有一個(gè)或者多個(gè)task在同一個(gè)工作進(jìn)程中，tuple將會(huì)被隨機(jī)發(fā)生給這些tasks。否則，和普通的Shuffle Grouping行為一致。

7）Tasks和Workers

每一個(gè)spout和bolt會(huì)被當(dāng)作很多task在整個(gè)集群里執(zhí)行。每一個(gè)executor對(duì)應(yīng)到一個(gè)線程，在這個(gè)線程上運(yùn)行多個(gè)task，而stream grouping則是定義怎么從一堆task發(fā)射tuple到另外一堆task。你可以調(diào)用TopologyBuilder類(lèi)的setSpout和setBolt來(lái)設(shè)置并行度（也就是有多少個(gè)task）。

一個(gè)topology可能會(huì)在一個(gè)或者多個(gè)worker（工作進(jìn)程）里面執(zhí)行，每個(gè)worker是一個(gè)物理JVM并且執(zhí)行整個(gè)topology的一部分。比如，對(duì)于并行度是300的topology來(lái)說(shuō)，如果我們使用50個(gè)工作進(jìn)程來(lái)執(zhí)行，那么每個(gè)工作進(jìn)程會(huì)處理其中的6個(gè)tasks。Storm會(huì)盡量均勻的工作分配給所有的worker。

8）Configuration

Storm里面有一堆參數(shù)可以配置來(lái)調(diào)整Nimbus, Supervisor以及正在運(yùn)行的topology的行為，一些配置是系統(tǒng)級(jí)別的，一些配置是topology級(jí)別的。default.yaml里面有所有的默認(rèn)配置。你可以通過(guò)定義個(gè)storm.yaml在你的classpath里來(lái)覆蓋這些默認(rèn)配置。并且你也可以在代碼里面設(shè)置一些topology相關(guān)的配置信息（使用StormSubmitter）。

總結(jié)

以上是生活随笔為你收集整理的在线实时大数据平台Storm集群组件学习的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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