Storm筆記

時間 2019-11-19

標籤 storm 筆記欄目 Storm 简体版

原文原文鏈接

參考：http://www.open-open.com/lib/view/open1374979211233.html　　html

　　Apache Storm 是由Twitter開源的分佈式實時計算系統。Storm能夠很是容易而且可靠的處理無限的數據流。對比Hadoop的批處理，Storm是一個實時的、分佈式的、具有高容錯的計算系統。Storm應用可使用不一樣的編程語言來進行開發。java

　　Storm的集羣表面上看和hadoop的集羣很是像。可是在Hadoop上面你運行的是MapReduce的Job。而在Storm上面你運行的是Topology。它們是很是不同的—一個關鍵的區別是：一個MapReduce Job最終會結束，而一個Topology運永遠運行（除非你顯式的殺掉他）。在Storm的集羣裏面有兩種節點：控制節點(master node)和工做節點(worker node)。控制節點上面運行一個後臺程序：Nimbus，它的做用相似Hadoop裏面的JobTracker。Nimbus負責在集羣裏面分佈代碼，分配工做給機器，而且監控狀態。每個工做節點上面運行一個叫作Supervisor的節點（相似 TaskTracker）。Supervisor會監聽分配給它那臺機器的工做，根據須要啓動/關閉工做進程。每個工做進程執行一個Topology（相似 Job）的一個子集；一個運行的Topology由運行在不少機器上的不少工做進程 Worker（相似 Child）組成。node

Nimbus和Supervisor之間的全部協調工做都是經過一個Zookeeper集羣來完成。而且，nimbus進程和supervisor都是快速失敗（fail-fast)和無狀態的。全部的狀態要麼在Zookeeper裏面，要麼在本地磁盤上。這也就意味着你能夠用kill -9來殺死nimbus和supervisor進程，而後再重啓它們，它們能夠繼續工做，就好像什麼都沒有發生過似的。這個設計使得storm難以想象的穩定。數據庫

Storm 分佈式計算結構稱爲 topology（拓撲），由stream（數據流），spout（數據流的生成者），bolt（運算）組成。
Storm 的核心數據結構是tuple。tuple是包含了一個或者多個鍵值對的列表，Stream是由無限制的tuple組成的序列。
spout 表明了一個Storm topology的主要數據入口，充當採集器的角色，鏈接到數據源，將數據轉化爲一個個tuple，並將tuple做爲數據流進行發射。
bolt 能夠理解爲計算程序中的運算或者函數，將一個或者多個數據流做爲輸入，對數據實施運算後，選擇性地輸出一個或者多個數據流。 bolt能夠訂閱多個由spout或者其餘bolt發射的數據流，這樣就能夠創建複雜的數據流轉換網絡。編程

　　Topologyapi

爲了在storm上面作實時計算，你要去創建一些topologies。一個topology就是一個計算節點所組成的圖。Topology裏面的每一個處理節點都包含處理邏輯，而節點之間的鏈接則表示數據流動的方向。數組

運行一個Topology是很簡單的。首先，把你全部的代碼以及所依賴的jar打進一個jar包。而後運行相似下面的這個命令。cookie

strom jar all-your-code.jar backtype.storm.MyTopology arg1 arg2

這個命令會運行主類: backtype.strom.MyTopology, 參數是arg1, arg2。這個類的main函數定義這個topology而且把它提交給Nimbus。storm jar負責鏈接到nimbus而且上傳jar文件。由於topology的定義其實就是一個Thrift結構而且nimbus就是一個Thrift服務，有能夠用任何語言建立而且提交topology。上面的方面是用JVM-based語言提交的最簡單的方法。計算任務Topology是由不一樣的Spouts和Bolts,經過數據流(Stream)鏈接起來的圖｡下面是一個Topology的結構示意圖: 網絡

其中包含有：數據結構

Spout:Storm中的消息源,用於爲Topology生產消息(數據),通常是從外部數據源(如Message Queue､RDBMS､NoSQL､Realtime Log)不間斷地讀取數據併發送給Topology消息(tuple元組)｡Spout能夠是可靠的，也能夠是不可靠的。若是這個tuple沒有被Storm徹底處理，可靠的消息源能夠從新發射一個tuple，可是不可靠的消息源一旦發出一個tuple就不能重發了。（可靠性會在下面介紹）
Spout類裏面最重要的方法是nextTuple｡要麼發射一個新的tuple到topology裏面或者簡單的返回(若是已經沒有新的tuple)｡要注意的是nextTuple方法不能阻塞,由於storm在同一個線程上面調用全部消息源spout的方法｡
另外兩個比較重要的spout方法是ack和fail｡storm在檢測到一個tuple被整個topology成功處理的時候調用ack,不然調用fail｡storm只對可靠的spout調用ack和fail｡
Bolt:Storm中的消息處理者,用於爲Topology進行消息的處理,Bolt能夠執行過濾, 聚合, 查詢數據庫等操做,並且能夠一級一級的進行處理｡

View Code

下圖是Topology的提交流程圖：

下圖是Storm的數據交互圖｡能夠看出兩個模塊Nimbus和Supervisor之間沒有直接交互｡狀態都是保存在Zookeeper上｡Worker之間經過ZeroMQ傳送數據｡Storm全部的元數據信息保存在Zookeeper中｡

　　Stream

　　Stream是storm裏面的關鍵抽象。一個stream是一個沒有邊界的tuple序列。storm提供一些原語來分佈式地、可靠地把一個stream傳輸進一個新的stream。好比：你能夠把一個tweets流傳輸到熱門話題的流。storm提供的最基本的處理stream的原語是spout和bolt。你能夠實現Spout和Bolt對應的接口以處理你的應用的邏輯。spout的流的源頭。好比一個spout可能從Kestrel隊列裏面讀取消息而且把這些消息發射成一個流。又好比一個spout能夠調用twitter的一個api而且把返回的tweets發射成一個流。一般Spout會從外部數據源（隊列、數據庫等）讀取數據，而後封裝成Tuple形式，以後發送到Stream中。Spout是一個主動的角色，在接口內部有個nextTuple函數，Storm框架會不停的調用該函數。

bolt能夠接收任意多個輸入stream，做一些處理，有些bolt可能還會發射一些新的stream。一些複雜的流轉換，好比從一些tweet裏面計算出熱門話題，須要多個步驟，從而也就須要多個bolt。Bolt能夠作任何事情:運行函數，過濾tuple作一些聚合，作一些合併以及訪問數據庫等等。Bolt處理輸入的Stream，併產生新的輸出Stream。Bolt能夠執行過濾、函數操做、Join、操做數據庫等任何操做。Bolt是一個被動的角色，其接口中有一個execute(Tuple input)方法，在接收到消息以後會調用此函數，用戶能夠在此方法中執行本身的處理邏輯。

spout和bolt所組成一個網絡會被打包成topology， topology是storm裏面最高一級的抽象（相似 Job），你能夠把topology提交給storm的集羣來運行。

topology裏面的每個節點都是並行運行的。在你的topology裏面，你能夠指定每一個節點的並行度， storm則會在集羣裏面分配那麼多線程來同時計算。一個topology會一直運行直到你顯式中止它。storm自動從新分配一些運行失敗的任務，而且storm保證你不會有數據丟失，即便在一些機器意外停機而且消息被丟掉的狀況下。

　　數據模型

Storm使用tuple來做爲它的數據模型。每一個tuple是一堆值，每一個值有一個名字，而且每一個值能夠是任何類型，在個人理解裏面一個tuple能夠看做一個沒有方法的java對象。整體來看，storm支持全部的基本類型、字符串以及字節數組做爲tuple的值類型。你也可使用你本身定義的類型來做爲值類型，只要你實現對應的序列化器(serializer)。一個Tuple表明數據流中的一個基本的處理單元，例如一條cookie日誌，它能夠包含多個Field，每一個Field表示一個屬性。

Tuple原本應該是一個Key-Value的Map，因爲各個組件間傳遞的tuple的字段名稱已經事先定義好了，因此Tuple只須要按序填入各個Value，因此就是一個Value List。
一個沒有邊界的、源源不斷的、連續的Tuple序列就組成了Stream。

topology裏面的每一個節點必須定義它要發射的tuple的每一個字段。好比下面這個bolt定義它所發射的tuple包含兩個字段，類型分別是: double和triple。

publicclassDoubleAndTripleBoltimplementsIRichBolt {
    privateOutputCollectorBase _collector;

    @Override
    publicvoidprepare(Map conf, TopologyContext context, OutputCollectorBase collector) {
        _collector = collector;
    }

    @Override
    publicvoidexecute(Tuple input) {
        intval = input.getInteger(0);
        _collector.emit(input,newValues(val*2, val*3));
        _collector.ack(input);
    }

    @Override
    publicvoidcleanup() {
    }

    @Override
    publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(newFields("double","triple"));
    }
}

declareOutputFields方法定義要輸出的字段： ["double", "triple"]。這個bolt的其它部分咱們接下來會解釋。

　　流分組策略(Stream grouping)

流分組策略告訴topology如何在兩個組件之間發送tuple。要記住，spouts和bolts以不少task的形式在topology裏面同步執行。若是從task的粒度來看一個運行的topology，它應該是這樣的:

從task角度來看topology

當Bolt A的一個task要發送一個tuple給Bolt B，它應該發送給Bolt B的哪一個task呢？
stream grouping專門回答這種問題的。在咱們深刻研究不一樣的stream grouping以前，讓咱們看一下storm-starter裏面的另一個topology。WordCountTopology讀取一些句子，輸出句子裏面每一個單詞出現的次數。

TopologyBuilder builder =newTopologyBuilder();
builder.setSpout(1,newRandomSentenceSpout(),5);
builder.setBolt(2,newSplitSentence(),8)
        .shuffleGrouping(1);
builder.setBolt(3,newWordCount(),12)
        .fieldsGrouping(2,newFields("word"));

SplitSentence對於句子裏面的每一個單詞發射一個新的tuple, WordCount在內存裏面維護一個單詞->次數的mapping， WordCount每收到一個單詞，它就更新內存裏面的統計狀態。

有好幾種不一樣的stream grouping:

最簡單的grouping是shuffle grouping, 它隨機發給任何一個task。上面例子裏面RandomSentenceSpout和SplitSentence之間用的就是shuffle grouping, shuffle grouping對各個task的tuple分配的比較均勻。
一種更有趣的grouping是fields grouping, SplitSentence和WordCount之間使用的就是fields grouping, 這種grouping機制保證相同field值的tuple會去同一個task，這對於WordCount來講很是關鍵，若是同一個單詞不去同一個task，那麼統計出來的單詞次數就不對了。

fields grouping是stream合併，stream聚合以及不少其它場景的基礎。在背後呢， fields grouping使用的一致性哈希來分配tuple的。

還有一些其它類型的stream grouping。下面是一些經常使用的「路由選擇」機制： Storm的Grouping即消息的Partition機制。當一個Tuple被髮送時，如何肯定將它發送個某個（些）Task來處理？？

ShuffleGrouping：隨機選擇一個Task來發送。
FiledGrouping：根據Tuple中Fields來作一致性hash，相同hash值的Tuple被髮送到相同的Task。
AllGrouping：廣播發送，將每個Tuple發送到全部的Task。
GlobalGrouping：全部的Tuple會被髮送到某個Bolt中的id最小的那個Task。
NoneGrouping：不關心Tuple發送給哪一個Task來處理，等價於ShuffleGrouping。
DirectGrouping：直接將Tuple發送到指定的Task來處理。

　　可靠的消息處理

　　Storm容許用戶在Spout中發射一個新的源Tuple時爲其指定一個MessageId，這個MessageId能夠是任意的Object對象。多個源Tuple能夠共用同一個MessageId，表示這多個源Tuple對用戶來講是同一個消息單元。Storm的可靠性是指Storm會告知用戶每個消息單元是否在一個指定的時間內被徹底處理。徹底處理的意思是該MessageId綁定的源Tuple以及由該源Tuple衍生的全部Tuple都通過了Topology中每個應該到達的Bolt的處理。

在Spout中由message 1綁定的tuple1和tuple2分別通過bolt1和bolt2的處理，而後生成了兩個新的Tuple，並最終流向了bolt3。當bolt3處理完以後，稱message 1被徹底處理了。
Storm中的每個Topology中都包含有一個Acker組件。Acker組件的任務就是跟蹤從Spout中流出的每個messageId所綁定的Tuple樹中的全部Tuple的處理狀況。若是在用戶設置的最大超時時間內這些Tuple沒有被徹底處理，那麼Acker會告訴Spout該消息處理失敗，相反則會告知Spout該消息處理成功。
那麼Acker是如何記錄Tuple的處理結果呢？？
A xor A = 0.
A xor B…xor B xor A = 0，其中每個操做數出現且僅出現兩次。
在Spout中，Storm系統會爲用戶指定的MessageId生成一個對應的64位的整數，做爲整個Tuple Tree的RootId。RootId會被傳遞給Acker以及後續的Bolt來做爲該消息單元的惟一標識。同時，不管Spout仍是Bolt每次新生成一個Tuple時，都會賦予該Tuple一個惟一的64位整數的Id。
當Spout發射完某個MessageId對應的源Tuple以後，它會告訴Acker本身發射的RootId以及生成的那些源Tuple的Id。而當Bolt處理完一個輸入Tuple併產生出新的Tuple時，也會告知Acker本身處理的輸入Tuple的Id以及新生成的那些Tuple的Id。Acker只須要對這些Id進行異或運算，就能判斷出該RootId對應的消息單元是否成功處理完成了。

實例：

咱們看一下storm-starter裏面的ExclamationTopology:

TopologyBuilder builder =newTopologyBuilder();
builder.setSpout(1,newTestWordSpout(),10);
builder.setBolt(2,newExclamationBolt(),3)
        .shuffleGrouping(1);
builder.setBolt(3,newExclamationBolt(),2)
        .shuffleGrouping(2);

這個Topology包含一個Spout和兩個Bolt。Spout發射單詞，每一個bolt在每一個單詞後面加個」!!!」。這三個節點被排成一條線: spout發射單詞給第一個bolt，第一個bolt而後把處理好的單詞發射給第二個bolt。若是spout發射的單詞是["bob"]和["john"], 那麼第二個bolt會發射["bolt!!!!!!"]和["john!!!!!!"]出來。咱們使用setSpout和setBolt來定義Topology裏面的節點。這些方法接收咱們指定的一個id，一個包含處理邏輯的對象(spout或者bolt), 以及你所須要的並行度。這個包含處理的對象若是是spout那麼要實現IRichSpout的接口，若是是bolt，那麼就要實現IRichBolt接口.最後一個指定並行度的參數是可選的。它表示集羣裏面須要多少個thread來一塊兒執行這個節點。若是你忽略它那麼storm會分配一個線程來執行這個節點。

setBolt方法返回一個InputDeclarer對象，這個對象是用來定義Bolt的輸入。這裏第一個Bolt聲明它要讀取spout所發射的全部的tuple — 使用shuffle grouping。而第二個bolt聲明它讀取第一個bolt所發射的tuple。shuffle grouping表示全部的tuple會被隨機的分發給bolt的全部task。給task分發tuple的策略有不少種。

若是你想第二個bolt讀取spout和第一個bolt所發射的全部的tuple，那麼你應該這樣定義第二個bolt:

builder.setBolt(3,newExclamationBolt(),5)
            .shuffleGrouping(1)
            .shuffleGrouping(2);

讓咱們深刻地看一下這個topology裏面的spout和bolt是怎麼實現的。Spout負責發射新的tuple到這個topology裏面來。TestWordSpout從["nathan", "mike", "jackson", "golda", "bertels"]裏面隨機選擇一個單詞發射出來。TestWordSpout裏面的nextTuple()方法是這樣定義的：

publicvoidnextTuple() {
    Utils.sleep(100);
    finalString[] words =newString[] {"nathan","mike",
                     "jackson","golda","bertels"};
    finalRandom rand =newRandom();
    finalString word = words[rand.nextInt(words.length)];
    _collector.emit(newValues(word));
}

ExclamationBolt把」!!!」拼接到輸入tuple後面。咱們來看下ExclamationBolt的完整實現。

publicstaticclassExclamationBoltimplementsIRichBolt {
    OutputCollector _collector;

    publicvoidprepare(Map conf, TopologyContext context,
                        OutputCollector collector) {
        _collector = collector;
    }

    publicvoidexecute(Tuple tuple) {
        _collector.emit(tuple,newValues(tuple.getString(0) +"!!!"));
        _collector.ack(tuple);
    }

    publicvoidcleanup() {
    }

    publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(newFields("word"));
    }
}

prepare方法提供給bolt一個Outputcollector用來發射tuple。Bolt能夠在任什麼時候候發射tuple — 在prepare, execute或者cleanup方法裏面, 或者甚至在另外一個線程裏面異步發射。這裏prepare方法只是簡單地把OutputCollector做爲一個類字段保存下來給後面execute方法使用。

execute方法從bolt的一個輸入接收tuple(一個bolt可能有多個輸入源). ExclamationBolt獲取tuple的第一個字段，加上」!!!」以後再發射出去。若是一個bolt有多個輸入源，你能夠經過調用Tuple#getSourceComponent方法來知道它是來自哪一個輸入源的。

execute方法裏面還有其它一些事情值得一提：輸入tuple被做爲emit方法的第一個參數，而且輸入tuple在最後一行被ack。這些呢都是Storm可靠性API的一部分，後面會解釋。

cleanup方法在bolt被關閉的時候調用，它應該清理全部被打開的資源。可是集羣不保證這個方法必定會被執行。好比執行task的機器down掉了，那麼根本就沒有辦法來調用那個方法。cleanup設計的時候是被用來在local mode的時候才被調用(也就是說在一個進程裏面模擬整個storm集羣), 而且你想在關閉一些topology的時候避免資源泄漏。最後，declareOutputFields定義一個叫作」word」的字段的tuple。

以local mode運行ExclamationTopology
讓咱們看看怎麼以local mode運行ExclamationToplogy。

storm的運行有兩種模式: 本地模式和分佈式模式。在本地模式中，Storm用一個進程裏面的線程來模擬全部的spout和bolt。本地模式對開發和測試來講比較有用。你運行storm-starter裏面的topology的時候它們就是以本地模式運行的，你能夠看到topology裏面的每個組件在發射什麼消息。

在分佈式模式下，Storm由一堆機器組成。當你提交topology給master的時候，你同時也把topology的代碼提交了。master負責分發你的代碼而且負責給你的topolgoy分配工做進程。若是一個工做進程掛掉了， master節點會把認爲從新分配到其它節點。關於如何在一個集羣上面運行topology，你能夠看看Running topologies on a production cluster文章。

下面是以本地模式運行ExclamationTopology的代碼:

Config conf =newConfig();
conf.setDebug(true);
conf.setNumWorkers(2);

LocalCluster cluster =newLocalCluster();
cluster.submitTopology("test", conf, builder.createTopology());
Utils.sleep(10000);
cluster.killTopology("test");
cluster.shutdown();

首先，這個代碼定義經過定義一個LocalCluster對象來定義一個進程內的集羣。提交topology給這個虛擬的集羣和提交topology給分佈式集羣是同樣的。經過調用submitTopology方法來提交topology，它接受三個參數：要運行的topology的名字，一個配置對象以及要運行的topology自己。

topology的名字是用來惟一區別一個topology的，這樣你而後能夠用這個名字來殺死這個topology的。前面已經說過了，你必須顯式的殺掉一個topology，不然它會一直運行。

Conf對象能夠配置不少東西，下面兩個是最多見的：

TOPOLOGY_WORKERS(setNumWorkers) 定義你但願集羣分配多少個工做進程給你來執行這個topology. topology裏面的每一個組件會被須要線程來執行。每一個組件到底用多少個線程是經過setBolt和setSpout來指定的。這些線程都運行在工做進程裏面. 每個工做進程包含一些節點的一些工做線程。好比，若是你指定300個線程，60個進程，那麼每一個工做進程裏面要執行6個線程，而這6個線程可能屬於不一樣的組件(Spout, Bolt)。你能夠經過調整每一個組件的並行度以及這些線程所在的進程數量來調整topology的性能。
TOPOLOGY_DEBUG(setDebug), 當它被設置成true的話， storm會記錄下每一個組件所發射的每條消息。這在本地環境調試topology頗有用，可是在線上這麼作的話會影響性能的。

運行中的Topology主要由如下三個組件組成的： Worker processes（進程） Executors (threads)（線程） Tasks