在Hadoop生態圈中，針對大資料進行批量計算時，通常需要一個或者多個MapReduce作業來完成，但這種批量計算方式是滿足不了對實時性要求高的場景。

Storm是一個開源分散式實時計算系統，它可以實時可靠地處理流資料。

Storm特點

在Storm出現之前，進行實時處理是非常痛苦的事情，我們主要的時間都花在關注往哪裡發訊息，從哪裡接收訊息，訊息如何序列化，真正的業務邏輯只佔了原始碼的一小部分。一個應用程式的邏輯執行在很多worker上，但這些worker需要各自單獨部署，還需要部署訊息佇列。最大問題是系統很脆弱，而且不是容錯的：需要自己保證訊息佇列和worker程序工作正常。

Storm完整地解決了這些問題。它是為分散式場景而生的，抽象了訊息傳遞，會自動地在叢集機器上併發地處理流式計算，讓你專注於實時處理的業務邏輯。

Storm有如下特點：

• 程式設計簡單：開發人員只需要關注應用邏輯，而且跟Hadoop類似，Storm提供的程式設計原語也很簡單；

• 高效能，低延遲：可以應用於廣告搜尋引擎這種要求對廣告主的操作進行實時響應的場景；

• 分散式：可以輕鬆應對資料量大，單機搞不定的場景；

• 可擴充套件：隨著業務發展，資料量和計算量越來越大，系統可水平擴充套件；

• 容錯：單個節點掛了不影響應用；

• 訊息不丟失：保證訊息處理。

不過Storm不是一個完整的解決方案，使用Storm時你需要關注以下幾點：

• 如果使用的是自己的訊息佇列，需要加入訊息佇列做資料的來源和產出的程式碼；

• 需要考慮如何做故障處理：如何記錄訊息處理的進度，應對Storm重啟，掛掉的場景；

• 需要考慮如何做訊息的回退：如果某些訊息處理一直失敗怎麼辦？

Storm與Hadoop區別

定義及架構

Hadoop是Apache的一個專案，是一個能夠對大量資料進行分散式處理的軟體框架。

Storm是Apache基金會的孵化專案，是應用於流式資料實時處理領域的分散式計算系統。

應用方面

Hadoop是分散式批處理計算，強調批處理，常用於資料探勘和分析。

Storm是分散式實時計算，強調實時性，常用於實時性要求較高的地方。

計算處理方式

Hadoop是磁碟級計算，進行計算時，資料在磁碟上，需要讀寫磁碟；Hadoop應用MapReduce的思想，將資料切片計算來處理大量的離線資料。Hadoop處理的資料必須是已經存放在HDFS上或者類似HBase的資料庫中，所以Hadoop實現的時候是通過移動計算到這些存放資料的機器上來提高效率的。

Storm是記憶體級計算，資料直接通過網路匯入記憶體。Storm是一個流計算框架，處理的資料是實時訊息佇列中的，需要寫好一個Topology邏輯，然後將接收進來的資料進行處理，所以Storm是通過移動資料平均分配到機器資源來獲得高效率的。

資料處理方面

資料來源：Hadoop是HDFS上某個資料夾下的資料，資料量可能以TB來計；而Storm則是實時新增的某一筆資料。

處理過程：Hadoop是Map階段到Reduce階段的；Storm是由使用者定義處理流程，流程中可以包含多個步驟，每個步驟可以是資料來源(SPOUT)，也可以是處理邏輯(BOLT)。

是否結束：Hadoop最後必須要結束；而Storm沒有結束狀態，到最後一步時，就停在那，直到有新資料進入時再重新開始。

處理速度：Hadoop以處理HDFS上大量資料為目的，速度慢；Storm只要處理新增的某一筆資料即可，故此它的速度很快。

適用場景：Hadoop主要是處理一批資料，對時效性要求不高，需要處理就提交一個JOB；而Storm主要是處理某一新增資料的，故此時效性要求高。

總結，Hadoop和Storm並沒有真的優劣之分，它們只是在各自的領域上有著獨特的效能而已，若是真的把它們進行單純的比較，反而是有失公平了。事實上，只有在最合適的方面使用最合適的大資料平臺，才能夠真正體現出它們的價值，也才能夠真正為我們的工作提供最為便捷的助力！

Storm基本概念

1) Topology

一個Storm拓撲打包了一個實時處理程式的邏輯。一個Storm拓撲跟一個MapReduce的任務(job)是類似的。主要區別是MapReduce任務最終會結束，而拓撲會一直執行（當然直到你殺死它)。一個拓撲是一個通過流分組(Stream Grouping)把Spout和Bolt連線到一起的拓撲結構。圖的每條邊代表一個Bolt訂閱了其他Spout或者Bolt的輸出流。一個拓撲就是一個複雜的多階段的流計算。

2) Tuple

元組是Storm提供的一個輕量級的資料格式，可以用來包裝你需要實際處理的資料。元組是一次訊息傳遞的基本單元。一個元組是一個命名的值列表，其中的每個值都可以是任意型別的。元組是動態地進行型別轉化的—欄位的型別不需要事先宣告。在Storm中程式設計時，就是在操作和轉換由元組組成的流。通常，元組包含整數，位元組，字串，浮點數，布林值和位元組陣列等型別。要想在元組中使用自定義型別，就需要實現自己的序列化方式。

3) Stream

流是Storm中的核心抽象。一個流由無限的元組序列組成，這些元組會被分散式並行地建立和處理。通過流中元組包含的欄位名稱來定義這個流。

每個流宣告時都被賦予了一個ID。只有一個流的Spout和Bolt非常常見，所以OutputFieldsDeclarer提供了不需要指定ID來宣告一個流的函式（Spout和Bolt都需要宣告輸出的流）。這種情況下，流的ID是預設的“default”。

4) Spout

Spout（噴嘴，這個名字很形象）是Storm中流的來源。通常Spout從外部資料來源，如訊息佇列中讀取元組資料並吐到拓撲裡。Spout可以是可靠的（reliable）或者不可靠（unreliable）的。可靠的Spout能夠在一個元組被Storm處理失敗時重新進行處理，而非可靠的Spout只是吐資料到拓撲裡，不關心處理成功還是失敗了。

Spout可以一次給多個流吐資料。此時需要通過OutputFieldsDeclarer的declareStream函式來宣告多個流並在呼叫SpoutOutputCollector提供的emit方法時指定元組吐給哪個流。

Spout中最主要的函式是nextTuple，Storm框架會不斷呼叫它去做元組的輪詢。如果沒有新的元組過來，就直接返回，否則把新元組吐到拓撲裡。nextTuple必須是非阻塞的，因為Storm在同一個執行緒裡執行Spout的函式。

Spout中另外兩個主要的函式是Ack和fail。當Storm檢測到一個從Spout吐出的元組在拓撲中成功處理完時呼叫Ack，沒有成功處理完時呼叫Fail。只有可靠型的Spout會呼叫Ack和Fail函式。

5) Bolt

在拓撲中所有的計算邏輯都是在Bolt中實現的。一個Bolt可以處理任意數量的輸入流，產生任意數量新的輸出流。Bolt可以做函式處理，過濾，流的合併，聚合，儲存到資料庫等操作。Bolt就是流水線上的一個處理單元，把資料的計算處理過程合理的拆分到多個Bolt、合理設定Bolt的task數量，能夠提高Bolt的處理能力，提升流水線的併發度。

Bolt可以給多個流吐出元組資料。此時需要使用OutputFieldsDeclarer的declareStream方法來宣告多個流並在使用的emit方法時指定給哪個流吐資料。

當你聲明瞭一個Bolt的輸入流，也就訂閱了另外一個元件的某個特定的輸出流。如果希望訂閱另一個元件的所有流，需要單獨挨個訂閱。InputDeclarer有語法糖來訂閱ID為預設值的流。例如declarer.shuffleGrouping（"redBolt"）訂閱了redBolt元件上的預設流，跟declarer.shuffleGrouping（"redBolt", DEFAULT_STREAM_ID）是相同的。

在Bolt中最主要的函式是execute函式，它使用一個新的元組當作輸入。Bolt使用OutputCollector物件來吐出新的元組。Bolts必須為處理的每個元組呼叫OutputCollector的ack方法以便於Storm知道元組什麼時候被各個Bolt處理完了（最終就可以確認Spout吐出的某個元組處理完了）。通常處理一個輸入的元組時，會基於這個元組吐出零個或者多個元組，然後確認（ack）輸入的元組處理完了，Storm提供了IBasicBolt介面來自動完成確認。

必須注意OutputCollector不是執行緒安全的，所以所有的吐資料（emit）、確認（ack）、通知失敗（fail）必須發生在同一個執行緒裡。更多資訊可以參照問題定位。

6) Task

每個Spout和Bolt會以多個任務（Task）的形式在叢集上執行。每個任務對應一個執行執行緒，流分組定義瞭如何從一組任務（同一個Bolt）傳送元組到另外一組任務（另外一個Bolt）上。可以在呼叫TopologyBuilder的setSpout和setBolt函式時設定每個Spout和Bolt的併發數。

7) Component

元件（component）是對Bolt和Spout的統稱。

8) Stream Grouping

定義拓撲的時候，一部分工作是指定每個Bolt應該消費哪些流。流分組定義了一個流在一個消費它的Bolt內的多個任務（task）之間如何分組。流分組跟計算機網路中的路由功能是類似的，決定了每個元組在拓撲中的處理路線。

在Storm中有七個內建的流分組策略，你也可以通過實現CustomStreamGrouping介面來自定義一個流分組策略：

• 洗牌分組（Shuffle grouping）：隨機分配元組到Bolt的某個任務上，這樣保證同一個Bolt的每個任務都能夠得到相同數量的元組。

• 欄位分組（Fields grouping）：按照指定的分組欄位來進行流的分組。例如，流是用欄位“user-id”來分組的，那有著相同“user-id”的元組就會分到同一個任務裡，但是有不同“user-id”的元組就會分到不同的任務裡。這是一種非常重要的分組方式，通過這種流分組方式，我們就可以做到讓Storm產出的訊息在這個”user-id”級別是嚴格有序的，這對一些對時序敏感的應用（例如，計費系統）是非常重要的。

• Partial Key grouping：跟欄位分組一樣，流也是用指定的分組欄位進行分組的，但是在多個下游Bolt之間是有負載均衡的，這樣當輸入資料有傾斜時可以更好的利用資源。這篇論文很好的解釋了這是如何工作的，有哪些優勢。

• All grouping：流會複製給Bolt的所有任務。小心使用這種分組方式。

• Global grouping：整個流會分配給Bolt的一個任務。具體一點，會分配給有最小ID的任務。

• 不分組（None grouping）：說明不關心流是如何分組的。目前，None grouping等價於洗牌分組。

• Direct grouping：一種特殊的分組。對於這樣分組的流，元組的生產者決定消費者的哪個任務會接收處理這個元組。只能在宣告做直連的流（direct streams）上宣告Direct groupings分組方式。只能通過使用emitDirect系列函式來吐元組給直連流。一個Bolt可以通過提供的TopologyContext來獲得消費者的任務ID，也可以通過OutputCollector物件的emit函式（會返回元組被髮送到的任務的ID）來跟蹤消費者的任務ID。

• Local or shuffle grouping：如果目標Bolt在同一個worker程序裡有一個或多個任務，元組就會通過洗牌的方式分配到這些同一個程序內的任務裡。否則，就跟普通的洗牌分組一樣。

9) Reliability

Storm保證了拓撲中Spout產生的每個元組都會被處理。Storm是通過跟蹤每個Spout所產生的所有元組構成的樹形結構並得知這棵樹何時被完整地處理來達到可靠性。每個拓撲對這些樹形結構都有一個關聯的“訊息超時”。如果在這個超時時間裡Storm檢測到Spout產生的一個元組沒有被成功處理完，那Spout的這個元組就處理失敗了，後續會重新處理一遍。

為了發揮Storm的可靠性，需要你在建立一個元組樹中的一條邊時告訴Storm，也需要在處理完每個元組之後告訴Storm。這些都是通過Bolt吐元組資料用的OutputCollector物件來完成的。標記是在emit函式裡完成，完成一個元組後需要使用Ack函式來告訴Storm。

10) Workers

拓撲以一個或多個Worker程序的方式執行。每個Worker程序是一個物理的Java虛擬機器，執行拓撲的一部分任務。例如，如果拓撲的併發設定成了300，分配了50個Worker，那麼每個Worker執行6個任務（作為Worker內部的執行緒）。Storm會盡量把所有的任務均分到所有的Worker上。

Storm系統架構

1) 主節點（Nimbus）

在分散式系統中，排程服務非常重要，它的設計，會直接關係到系統的執行效率，錯誤恢復（fail over）、故障檢測（error detection）和水平擴充套件（scale）的能力。

叢集上任務（task）的排程由一個Master節點來負責。這臺機器上執行的Nimbus程序負責任務的排程。另外一個程序是Storm UI，可以介面上檢視叢集和所有的拓撲的執行狀態。

2) 從節點（Supervisor）

Storm叢集上有多個從節點，他們從Nimbus上下載拓撲的程式碼，然後去真正執行。Slave上的Supervisor程序是用來監督和管理實際執行業務程式碼的程序。在Storm 0.9之後，又多了一個程序Logviewer,可以用Storm UI來檢視Slave節點上的log檔案。

3) 協調服務Zookeeper

ZooKeeper在Storm上不是用來做訊息傳輸用的，而是用來提供協調服務（coordination service），同時儲存拓撲的狀態和統計資料。

• Supervisor，Nimbus和worker都在ZooKeeper留下約定好的資訊。例如Supervisor啟動時，會在ZooKeeper上註冊，Nimbus就可以發現Supervisor；Supervisor在ZooKeeper上留下心跳資訊，Nimbus通過這些心跳資訊來對Supervisor進行健康檢測，檢測出壞節點。

• 由於Storm元件（component）的狀態資訊儲存在ZooKeeper上，所以Storm元件就可以無狀態，可以 kill -9來殺死。例如：Supervisors/Nimbus的重啟不影響正在執行中的拓撲，因為狀態都在ZooKeeper上，從ZooKeeper上重新載入一下就好了。

• 用來做心跳：Worker通過ZooKeeper把孩子executor的情況以心跳的形式彙報給Nimbus；Supervisor程序通過ZK把自己的狀態也以心跳的形式彙報給Nimbua。

• 儲存最近任務的錯誤情況（拓撲停止時會刪除）。

4) 程序Worker

執行具體處理元件邏輯的程序，一個Topology可能會在一個或者多個worker裡面執行，每個worker是一個物理JVM並且執行整個Topology的一部分。

例如：對於並行度是300的topology來說，如果我們使用50個工作程序來執行，那麼每個工作程序會處理其中的6個tasks，Storm會盡量均勻的工作分配給所有的worker。

5) Task

Worker中的每一個spout/bolt的執行緒稱為一個task，每一個spout和bolt會被當作很多task在整個叢集裡執行，每一個executor對應到一個執行緒，在這個執行緒上執行多個task，Stream Grouping則是定義怎麼從一堆task發射tuple到另外一堆task，可以呼叫TopologyBuilder類的setSpout和setBolt來設定並行度（也就是有多少個task）。

Storm容錯機制

Storm的容錯機制包括架構容錯和資料容錯。

1) 架構容錯

Nimbus和Supervisor程序被設計成快速失敗（fail fast）的（當遇到異常的情況，程序就會掛掉）並且是無狀態的（狀態都儲存在Zookeeper或者在磁碟上）。

最重要的是，worker程序不會因為Nimbus或者Supervisor掛掉而受影響。這跟Hadoop是不一樣的，當JobTracker掛掉，所有的任務都會沒了。

當Nimbus掛掉會怎樣？

如果Nimbus是以推薦的方式處於程序監管（例如通過supervisord）之下，那它會被重啟，不會有任何影響。否則當Nimbus掛掉後：

• 已經存在的拓撲可以繼續正常執行，但是不能提交新拓撲；

• 正在執行的worker程序仍然可以繼續工作。而且當worker掛掉，supervisor會一直重啟worker；
  

• 失敗的任務不會被分配到其他機器（是Nimbus的職責）上了。
  

當一個Supervisor（slave節點）掛掉會怎樣？

如果Supervisor是以推薦的方式處於程序監管（例如通過supervisord）之下，那它會被重啟，不會有任何影響。否則當Supervisor掛掉：分配到這臺機器的所有任務（task）會超時，Nimbus會把這些任務（task）重新分配給其他機器。

當一個worker掛掉會怎麼樣？

當一個worker掛掉，supervisor會重啟它。如果啟動一直失敗那麼此時worker也就不能和Nimbus保持心跳了，Nimbus會重新分配worker到其他機器。

Nimbus算是一個單點故障嗎？

如果Nimbus節點掛掉，worker程序仍然可以繼續工作。而且當worker掛掉，supervisor會一直重啟worker。但是，沒有了Nimbus，當需要的時候（如果worker機器掛掉了）worker就不能被重新分配到其他機器了。

所以答案是，Nimbus在“某種程度”上屬於單點故障的。在實際中，這種情況沒什麼大不了的，因為當Nimbus程序掛掉，不會有災難性的事情發生。

2) 資料容錯

Storm中的每一個Topology中都包含有一個Acker元件。 Acker元件的任務就是跟蹤從某個task中的Spout流出的每一個messageId所繫結的Tuple樹中的所有Tuple的處理情況。如果在使用者設定的最大超時時間（timetout 可以通過 Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS來指定）內這些Tuple沒有被完全處理，那麼Acker會告訴Spout該訊息處理失敗，相反則會告知Spout該訊息處理成功,它會分別呼叫Spout中的fail和ack方法。

一個簡單的Storm實現

實現一個拓撲包括一個spout和兩個bolt。Spout傳送單詞。每個bolt在輸入資料的尾部追加字串“!!!”。三個節點排成一條線：spout發射給首個bolt，然後，這個bolt再發射給第二個bolt。如果spout發射元組“bob”和“john”，然後，第二個bolt將發射元組“bob!!!!!!”和“john!!!!!!”。

• 其中Topology程式碼如下，定義整個網路拓撲圖：

• Spout實現：

• Bolt實現：

Storm常用配置

• Config.TOPOLOGY_WORKERS

這個設定用多少個工作程序來執行這個topology。比如，如果你把它設定成25， 那麼叢集裡面一共會有25個java程序來執行這個topology的所有task。如果你的這個topology裡面所有元件加起來一共有150的並行度，那麼每個程序裡面會有6個執行緒(150 / 25 = 6)。

• Config.TOPOLOGY_ACKERS

這個配置設定acker任務的並行度。預設的acker任務並行度為1，當系統中有大量的訊息時，應該適當提高acker任務的併發度。設定為0，通過此方法，當Spout傳送一個訊息的時候，它的ack方法將立刻被呼叫。

• Config.TOPOLOGY_MAX_SPOUT_PENDING

這個設定一個spout task上面最多有多少個沒有處理的tuple（沒有ack/failed）回覆， 我們推薦你設定這個配置，以防止tuple佇列爆掉。

• Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS

這個配置storm的tuple的超時時間 – 超過這個時間的tuple被認為處理失敗了。這個設定的預設設定是30秒。

來源：來源中國，閱讀原文獲取更多精彩內容。