Storm簡介Storm簡介實時計算需要解決一些什么問題實現一個實時計算系統Storm基本概念Storm使用場景Storm分組機制Storm簡介實時計算需要解決一些什么問題

伴隨著信息科技日新月異的發展，信息呈現出爆發式的膨脹，人們獲取信息的途徑也更加多樣、更加便捷，同時對于信息的時效性要求也越來越高。舉個搜索場景中的例子，當一個賣家發布了一條寶貝信息時，他希望的當然是這個寶貝馬上就可以被賣家搜索出來、點擊、購買啦，相反，如果這個寶貝要等到第二天或者更久才可以被搜出來，估計這個大哥就要罵娘了。再舉一個推薦的例子，如果用戶昨天在淘寶上買了一雙襪子，今天想買一副泳鏡去游泳，但是卻發現系統在不遺余力地給他推薦襪子、鞋子，根本對他今天尋找泳鏡的行為視而不見，估計這哥們心里就會想推薦你妹呀。其實稍微了解點背景知識的碼農們都知道，這是因為后臺系統做的是每天一次的全量處理，而且大多是在夜深人靜之時做的，那么你今天白天做的事情當然要明天才能反映出來啦。Storm簡介實現一個實時計算系統

全量數據處理使用的大多是鼎鼎大名的hadoop或者hive，作為一個批處理系統，hadoop以其吞吐量大、自動容錯等優點，在海量數據處理上得到了廣泛的使用。但是，hadoop不擅長實時計算，因為它天然就是為批處理而生的，這也是業界一致的共識。否則最近這兩年也不會有s4,storm,puma這些實時計算系統如雨后春筍般冒出來啦。先拋開s4,storm,puma這些系統不談，我們首先來看一下，如果讓我們自己設計一個實時計算系統，我們要解決哪些問題。Storm簡介實現一個實時計算系統低延遲。都說了是實時計算系統了，延遲是一定要低的。高性能。性能不高就是浪費機器，浪費機器是要受批評的哦。分布式。系統都是為應用場景而生的，如果你的應用場景、你的數據和計算單機就能搞定，那么不用考慮這些復雜的問題了。我們所說的是單機搞不定的情況?？蓴U展。伴隨著業務的發展，我們的數據量、計算量可能會越來越大，所以希望這個系統是可擴展的。容錯。這是分布式系統中通用問題。一個節點掛了不能影響我的應用。Storm簡介實現一個實時計算系統

好，如果僅僅需要解決這5個問題，可能會有無數種方案，而且各有千秋，隨便舉一種方案，使用消息隊列+分布在各個機器上的工作進程就ok啦。我們再繼續往下看。容易在上面開發應用程序。親，你設計的系統需要應用程序開發人員考慮各個處理組件的分布、消息的傳遞嗎？如果是，那有點麻煩啊，開發人員可能會用不好，也不會想去用。消息不丟失。用戶發布的一個寶貝消息不能在實時處理的時候給丟了，對吧？更嚴格一點，如果是一個精確數據統計的應用，那么它處理的消息要不多不少才行。這個要求有點高哦。消息嚴格有序。有些消息之間是有強相關性的，比如同一個寶貝的更新和刪除操作消息，如果處理時搞亂順序完全是不一樣的效果了。Storm簡介Storm基本概念

對比Hadoop的批處理，Storm是個實時的、分布式以及具備高容錯的計算系統。同Hadoop一樣Storm也可以處理大批量的數據，然而Storm在保證高可靠性的前提下還可以讓處理進行的更加實時；也就是說，所有的信息都會被處理。Storm同樣還具備容錯和分布計算這些特性，這就讓Storm可以擴展到不同的機器上進行大批量的數據處理。他同樣還有以下的這些特性：Storm簡介Storm優勢1.簡單的編程模型。類似于MapReduce降低了并行批處理復雜性，Storm降低了進行實時處理的復雜性。2.服務化,一個服務框架,支持熱部署,即時上線或下線App.3.可以使用各種編程語言。你可以在Storm之上使用各種編程語言。默認支持Clojure、Java、Ruby和Python。要增加對其他語言的支持，只需實現一個簡單的Storm通信協議即可。4.容錯性。Storm會管理工作進程和節點的故障。5.水平擴展。計算是在多個線程、進程和服務器之間并行進行的。Storm簡介Storm基本概念6.可靠的消息處理。Storm保證每個消息至少能得到一次完整處理。任務失敗時，它會負責從消息源重試消息。7.快速。系統的設計保證了消息能得到快速的處理，使用ZeroMQ作為其底層消息隊列。8.本地模式。Storm有一個“本地模式”，可以在處理過程中完全模擬Storm集群。這讓你可以快速進行開發和單元測試。Storm簡介Storm存在的問題1.目前的開源版本中只是單節點Nimbus，掛掉只能自動重啟，可以考慮實現一個雙nimbus的布局。2.Clojure是一個在JVM平臺運行的動態函數式編程語言,優勢在于流程計算，Storm的部分核心內容由Clojure編寫，雖然性能上提高不少但同時也提升了維護成本。Storm簡介Storm的架構Storm集群由一個主節點和多個工作節點組成。主節點運行了一個名為“Nimbus”的守護進程，用于分配代碼、布置任務及故障檢測。每個工作節點都運行了一個名為“Supervisor”的守護進程，用于監聽工作，開始并終止工作進程。Nimbus和Supervisor都能快速失敗，而且是無狀態的，這樣一來它們就變得十分健壯，兩者的協調工作是由Zookeeper來完成的。ZooKeeper用于管理集群中的不同組件，ZeroMQ是內部消息系統，JZMQ是ZeroMQMQ的JavaBinding。有個名為storm-deploy的子項目，可以在AWS上一鍵部署Storm集群.

Storm簡介Storm基本概念

首先我們通過一個storm和hadoop的對比來了解storm中的基本概念。Storm簡介Storm基本概念Nimbus：負責資源分配和任務調度。Supervisor：負責接受nimbus分配的任務，啟動和停止屬于自己管理的worker進程。Worker：運行具體處理組件邏輯的進程。Task：worker中每一個spout/bolt的線程稱為一個task.在storm0.8之后，task不再與物理線程對應，同一個spout/bolt的task可能會共享一個物理線程，該線程稱為executor。Storm簡介Storm基本概念Storm簡介Storm基本概念Topology：storm中運行的一個實時應用程序，因為各個組件間的消息流動形成邏輯上的一個拓撲結構。Spout：在一個topology中產生源數據流的組件。通常情況下spout會從外部數據源中讀取數據，然后轉換為topology內部的源數據。Spout是一個主動的角色，其接口中有個nextTuple()函數，storm框架會不停地調用此函數，用戶只要在其中生成源數據即可。Bolt：在一個topology中接受數據然后執行處理的組件。Bolt可以執行過濾、函數操作、合并、寫數據庫等任何操作。Bolt是一個被動的角色，其接口中有個execute(Tupleinput)函數,在接受到消息后會調用此函數，用戶可以在其中執行自己想要的操作。Storm簡介Storm基本概念Tuple：一次消息傳遞的基本單元。本來應該是一個key-value的map，但是由于各個組件間傳遞的tuple的字段名稱已經事先定義好，所以tuple中只要按序填入各個value就行了，所以就是一個valuelist.Stream：源源不斷傳遞的tuple就組成了stream。Storm簡介Storm使用場景1.流聚合

流聚合把兩個或者多個數據流聚合成一個數據流—基于一些共同的tuple字段。builder.setBolt(5,newMyJoiner(),parallelism)

.fieldsGrouping(1,newFields("joinfield1","joinfield2"))

.fieldsGrouping(2,newFields("joinfield1","joinfield2"))

.fieldsGrouping(3,newFields("joinfield1","joinfield2"))Storm簡介Storm使用場景2.批處理

有時候為了性能或者一些別的原因，你可能想把一組tuple一起處理，而不是一個個單獨處理。3.BasicBolt

1.讀一個輸入tuple

2.根據這個輸入tuple發射一個或者多個tuple

3.在execute的方法的最后ack那個輸入tuple

遵循這類模式的bolt一般是函數或者是過濾器,這種模式太常見，storm為這類模式單獨封裝了一個接口:IbasicBolt

Storm簡介Storm使用場景4.內存內緩存+Fieldsgrouping組合

在bolt的內存里面緩存一些東西非常常見。緩存在和fieldsgrouping結合起來之后就更有用了。比如，你有一個bolt把短鏈接變成長鏈接(bit.ly,t.co之類的)。你可以把短鏈接到長鏈接的對應關系利用LRU算法緩存在內存里面以避免重復計算。比如組件一發射短鏈接，組件二把短鏈接轉化成長鏈接并緩存在內存里面。看一下下面兩段代碼有什么不一樣：builder.setBolt(2,newExpandUrl(),parallelism).shuffleGrouping(1);builder.setBolt(2,newExpandUrl(),parallelism)

.fieldsGrouping(1,newFields("url"));Storm簡介Storm使用場景5.計算topN

比如你有一個bolt發射這樣的tuple:"value","count"并且你想一個bolt基于這些信息算出topN的tuple。最簡單的辦法是有一個bolt可以做一個全局的grouping的動作并且在內存里面保持這topN的值。

這個方式對于大數據量的流顯然是沒有擴展性的，因為所有的數據會被發到同一臺機器。一個更好的方法是在多臺機器上面并行的計算這個流每一部分的topN,然后再有一個bolt合并這些機器上面所算出來的topN以算出最后的topN,代碼大概是這樣的:

Storm簡介Storm使用場景builder.setBolt(2,newRankObjects(),parallellism)

.fieldsGrouping(1,newFields("value"));builder.setBolt(3,newMergeObjects())

.globalGrouping(2);

這個模式之所以可以成功是因為第一個bolt的fieldsgrouping使得這種并行算法在語義上是正確的。

Storm簡介Storm使用場景6.用TimeCacheMap來高效地保存一個最近被更新的對象的緩存

有時候你想在內存里面保存一些最近活躍的對象，以及那些不再活躍的對象。TimeCacheMap是一個非常高效的數據結構，它提供了一些callback函數使得我們在對象不再活躍的時候我們可以做一些事情

Storm簡介Storm使用場景7.分布式RPC:CoordinatedBolt和KeyedFairBolt

用storm做分布式RPC應用的時候有兩種比較常見的模式:它們被封裝在CoordinatedBolt和KeyedFairBolt里面.CoordinatedBolt包裝你的bolt,并且確定什么時候你的bolt已經接收到所有的tuple,它主要使用DirectStream來做這個.

KeyedFairBolt同樣包裝你的bolt并且保證你的topology同時處理多個DRPC調用，而不是串行地一次只執行一個。

Storm簡介Storm分組機制 StreamGrouping定義了一個流在Bolt任務間該如何被切分。這里有Stor