Storm數據流模型的分析及討論
Storm基本概念
Storm是一個開源的實時計算系統，它提供了一系列的基本元素用于進行計算：Topology、Stream、Spout、Bolt等等。
在Storm中，一個實時應用的計算任務被打包作為Topology發布，這同Hadoop的MapReduce任務相似。但是有一點不同的是：在Hadoop中，MapReduce任務最終會執行完成后結束；而在Storm中，Topology任務一旦提交后永遠不會結束，除非你顯示去停止任務。
計算任務Topology是由不同的Spouts和Bolts，通過數據流（Stream）連接起來的圖。下面是一個Topology的結構示意圖：
其中包含有：
Spout：Storm中的消息源，用于為Topology生產消息（數據），一般是從外部數據源（如Message Queue、RDBMS、NoSQL、Realtime Log）不間斷地讀取數據并發送給Topology消息（tuple元組）。
Bolt：Storm中的消息處理者，用于為Topology進行消息的處理，Bolt可以執行過濾， 聚合， 查詢數據庫等操作，而且可以一級一級的進行處理。
最終，Topology會被提交到storm集群中運行；也可以通過命令停止Topology的運行，將Topology占用的計算資源歸還給Storm集群。
Storm數據流模型
數據流（Stream）是Storm中對數據進行的抽象，它是時間上無界的tuple元組序列。在Topology中，Spout是Stream的源頭，負責為Topology從特定數據源發射Stream；Bolt可以接收任意多個Stream作為輸入，然后進行數據的加工處理過程，如果需要，Bolt還可以發射出新的Stream給下級Bolt進行處理。
下面是一個Topology內部Spout和Bolt之間的數據流關系：
Topology中每一個計算組件（Spout和Bolt）都有一個并行執行度，在創建Topology時可以進行指定，Storm會在集群內分配對應并行度個數的線程來同時執行這一組件。
那么，有一個問題：既然對于一個Spout或Bolt，都會有多個task線程來運行，那么如何在兩個組件（Spout和Bolt）之間發送tuple元組呢？
Storm提供了若干種數據流分發（Stream Grouping）策略用來解決這一問題。在Topology定義時，需要為每個Bolt指定接收什么樣的Stream作為其輸入（注：Spout并不需要接收Stream，只會發射Stream）。
目前Storm中提供了以下7種Stream Grouping策略：Shuffle Grouping、Fields Grouping、All Grouping、Global Grouping、Non Grouping、Direct Grouping、Local or shuffle grouping，具體策略可以參考這里。
一種Storm不能支持的場景
以上介紹了一些Storm中的基本概念，可以看出，Storm中Stream的概念是Topology內唯一的，只能在Topology內按照“發布-訂閱”方式在不同的計算組件（Spout和Bolt）之間進行數據的流動，而Stream在Topology之間是無法流動的。
這一點限制了Storm在一些場景下的應用，下面通過一個簡單的實例來說明。
假設現在有一個Topology1的結構如下：通過Spout產生數據流后，依次需要經過Filter Bolt，Join Bolt，Business1 Bolt。其中，Filter Bolt用于對數據進行過濾，Join Bolt用于數據流的聚合，Business1 Bolt用于進行一個實際業務的計算邏輯。
目前這個Topology1已經被提交到Storm集群運行，而現在我們又有了新的需求，需要計算一個新的業務邏輯，而這個Topology的特點是和Topology1公用同樣的數據源，而且前期的預處理過程完全一樣（依次經歷Filter Bolt和Join Bolt），那么這時候Storm怎么來滿足這一需求？據個人了解，有以下幾種“曲折”的實現方式：
1） 第一種方式：首先kill掉已經在集群中運行的Topology1計算任務，然后實現Business2 Bolt的計算邏輯，并重新打包形成一個新的Topology計算任務jar包后，提交到Storm集群中重新運行，這時候Storm內的整體Topology結構如下：
這種方式的缺點在于：由于要重啟Topology，所以如果Spout或Bolt有狀態則會丟失掉；同時由于Topology結構發生了變化，因此重新運行Topology前需要對程序的穩定性、正確性進行驗證；另外Topology結構的變化也會帶來額外的運維開銷。
2） 第二種方式：完全開發部署一套新的Topology，其中前面的公共部分的Spout和Bolt可以直接復用，只需要重新開發新的計算邏輯Business2 Bolt來替換原有的Business1 Bolt即可。然后重新提交新的Topology運行。這時候Storm內的整體Topology結構如下：
這種方式的缺點在于：由于兩個Topology都會從External Data Source讀取同一份數據，無疑增加了External Data Source的負載壓力；而且會導致同樣的數據在Storm集群內被傳輸相同的兩份，被同樣的計算單元Bolt進行處理，浪費了Storm的計算資源和網絡傳輸帶寬。假設現在不止有兩個這樣的Topology計算任務，而是有N個，那么對Storm的計算Slot的浪費很嚴重。
注意：上述兩種方式還有一個公共的缺點——系統可擴展性不好，這意味著不管哪種方式，只要以后有這種新增業務邏輯的需求，都需要進行復雜的人工操作或線性的資源浪費現象。
3） 第三種方式：OK，看了以上兩種方式后，也許你會提出下面的解決方案：通過Kafka這樣的消息中間件，實現不同Topology的Spout共享數據源，而且這樣可以做到消息可靠傳輸、消息rewind回傳等，好處是對于Storm來說，已經有了storm-kafka插件的支持。這時候Storm內的整體Topology結構如下：
這種實現方式可以通過引入一層消息中間件減少對External Data Source的重復訪問的壓力，而且可以通過消息中間件層，屏蔽掉External Data Source的細節，如果需要擴展新的業務邏輯，只需要重新部署運行新的Topology，應該說是現有Storm版本下很好的實現方式了。不過消息中間件的引入，無疑將給系統帶來了一定的復雜性，這對于Storm上的應用開發來說提高了門檻。
值得注意的是，方案三中仍遺留有一點問題沒有解決：對于Storm集群來說，這種方式還是沒有能夠從根本上避免數據在Storm不同Topology內的重復發送與處理。這是由于Storm的數據流模型上的限制所導致的，如果Storm實現了不同Topology之間Stream的共享，那么這一問題也就迎刃而解了。
一個流處理系統的數據流模型
個人工作中有幸參與過一個流處理框架的開發與應用。下面我們來簡單看看其中所采用的數據流模型。
首先，先來看一下該流處理系統內的幾個基本概念：
1）數據流（data stream）：時間分布和數量上無限的一系列數據記錄的集合體；
2）數據記錄（data record）：數據流的最小組成單元，每條數據記錄包括 3 類數據：所屬數據流名稱（stream name）、用于路由的數據（keys）和具體數據處理邏輯所需的數據（value）；
3）數據處理任務定義（task definition）：定義一個數據處理任務的基本屬性，無法直接被執行，必須特化為具體的任務實例。其基本屬性包括：
（可選）輸入流（input stream）：描述該任務依賴哪些數據流作為輸入，是一個數據流名稱列表；數據流產生源不會依賴其他數據流，可忽略該配置；
數據處理邏輯（process logic）：描述該任務具體的處理邏輯，例如由獨立進程進行的外部處理邏輯；
（可選）輸出流（output stream）：描述該任務產生哪個數據流，是一個數據流名稱；數據流處理鏈末級任務不會產生新的數據流，可忽略該配置；
4）數據處理任務實例（task instance）：對一個數據處理任務定義進行具體約束后，可推送到某個處理結點上運行的邏輯實體。附加下列屬性：
數據處理任務定義：指向該任務實例對應的數據處理任務定義實體；
輸入流過濾條件（input filting condition）：一個 boolean 表達式列表，描述每個輸入流中符合什么條件的數據記錄可以作為有效數據交給處理邏輯；若某個輸入流中所有數據記錄都是有效數據，則可直接用 true 表示；
（可選）強制輸出周期（output interval）：描述以什么頻率強制該任務實例產生輸出流記錄，可以用輸入流記錄個數或間隔時間作為周期；忽略該配置時，輸出流記錄產生周期完全由處理邏輯自身決定，不受框架約束；
5）數據處理結點（node）：可容納多個數據處理任務實例運行的實體機器，每個數據處理結點的IPv4地址必須保證唯一。
該流處理系統，采用分布式策略，由多個數據處理結點進行數據的處理過程；將流式數據的處理過程劃分為不同的階段，每個階段伴隨數據流的流入、任務的處理及數據流的流出；各個階段會有若干個處理結點參與完成，其中，每個處理結點上會有若干個數據處理任務實例運行，每個數據處理任務實例則是對一個數據處理任務定義進行具體約束后，可推送到某個處理結點上運行的邏輯實體。在不同的處理結點之間，數據流根據配置信息進行傳輸；在處理結點內部，結點根據配置信息對流經該結點的數據進行處理。
下圖為系統對于流式數據的基本處理流程：
1）定義數據流：將流式數據的處理過程劃分成不同的階段，定義出不同的數據流名稱；
2）定義數據處理任務：為數據流的處理過程定義相應的數據處理任務，其中，各個處理任務定義了外部處理邏輯，且其輸入/輸出數據流須從1）中預定義的數據流列表中選取；
3）定義數據處理結點：定義各個數據處理結點的名稱及其IPv4地址信息；
4）定義數據處理任務實例：為3）中定義好的每個處理結點，分別定義運行在其上的數據處理任務實例，其中，每個任務實例所對應的數據處理任務實體須從2）中預定義的處理任務列表中選取；
5）加載數據流的相關配置信息及訂閱信息（具體格式見“附錄：配置信息格式”），然后開始從數據流產生源讀取數據；
6）運行數據流源結點上的任務實例：數據流處理鏈源結點上的處理任務實例直接對數據流產生源的數據進行處理，然后產生新的輸出數據流；
7）運行下一級結點上的任務實例：中間的處理結點上的處理任務實例依賴于上一級處理結點的輸出數據流作為輸入數據流，從中讀取數據，進行處理，產生輸出數據流，并傳遞到下一級處理結點；
8）判斷是否到達數據流末級結點：
-
如果不是，則繼續返回步驟7），按照數據流的流動關系，繼續運行下一級結點上的任務實例；
-
否則為數據流末級結點，則進行步驟9）。
9）輸出結果：數據流處理鏈末級結點上的處理任務不會產生新的數據流，完成最終的數據處理任務后將結果進行輸出。
以上是數據流在不同處理結點之間的處理流程。每個處理結點作為流處理的一個環節，其結點內部的處理流程如下圖所示：
1）每個處理結點啟動流處理過程后，開啟網絡服務，監聽從上一級處理結點發出的TCP連接請求，接收從上一級結點發來的數據；
2）處理結點不間斷地接收從上一級處理結點發來的數據，對于每條數據記錄，根據數據流名進行篩選，將其分發到該數據流所對應的處理進程中；
3）將從每個特定數據流發來的數據，廣播到所有以該數據流作為輸入數據流的數據處理任務實例中；
4）數據處理任務實例從其輸入數據流中接收數據，按照過濾條件進行篩選，然后將符合過濾條件的數據記錄發送給外部應用程序進行處理；
5）外部應用程序啟動外部處理進程，對數據進行實際處理過程，并將每條數據記錄的處理結果返回給相應的數據處理任務實例；
6）數據處理任務實例從外部應用程序收集處理后的結果數據，并依次將其轉發到對應的輸出數據流中；
7）輸出數據流進程接收發向該數據流的數據，然后按照數據流的訂閱關系，將數據發送到所有訂閱了該數據流的下一級處理結點；
8）根據下一級處理結點的IP地址和端口號，通過TCP請求與下一級處理結點建立網絡連接，然后將數據按序傳輸到下一級處理結點。
二者在數據流模型上的不同之處
至于兩個系統的實現細節，我們先不去做具體比較，下面僅列出二者在數據流模型上的一些不同之處（這里并不是為了全面對比二者的不同之處，只是列出其中的關鍵部分）：
1） 在Storm中，數據流Stream是在Topology內進行定義，并在Topology內進行傳輸的；而在上面提到的流處理系統中，數據流Stream是在整個系統內全局唯一的，可以在整個集群內被訂閱。
2） 在Storm中，數據流Stream的發布和訂閱都是靜態的，所謂靜態是指數據流的發布與訂閱關系在向Storm集群提交Topology計算任務時，被一次性生成的，這一關系在Topology的運行過程中是不能被改變的；而在上面提到的流處理系統中，數據流Stream的發布和訂閱都是動態的，即數據處理任務task可以動態的發布Stream，也可以動態的訂閱系統內已經生成的任意Stream，數據流的訂閱關于通過分布式應用程序協調服務ZooKeeper集群的動態節點來維護管理。
好了，有了以上的對比，我們不難發現，對于本文所舉的應用場景實例，Storm的數據流模式尚不能很方便的支持，而在這里提到的這個流處理系統的全局數據流模型下，這一應用場景的需求可以很方便的滿足。
總結的話
個人覺得，Storm有必要實現不同Topology之間Stream的共享，這個至少可以在不損失Storm現有功能的前提下，使得Storm在處理實際生產環境下的一些應用場景時更加從容應對。
至于如何在現有Storm的基礎上實現這一需求，可能的方式很多。一種簡單的方式是通過Zookeeper來集中存儲、動態感知Topology之間Stream的“發布-訂閱”關系，同時在Storm的消息分發過程中對這種情況加以處理。
以上觀點，如果不對之處，歡迎大家指出。
原文鏈接：http://www.cnblogs.com/panfeng412/archive/2012/07/29/storm-stream-model-analysis-and-discussion.html
【編輯推薦】
