Storm入門教程第五章一致性事務

時間 2019-11-19

原文原文鏈接

Storm是一個分佈式的流處理系統，利用anchor和ack機制保證全部tuple都被成功處理。若是tuple出錯，則能夠被重傳，可是如何保證出錯的tuple只被處理一次呢？Storm提供了一套事務性組件Transaction Topology，用來解決這個問題。數據庫

Transactional Topology目前已經再也不維護，由Trident來實現事務性topology，可是原理相同。框架

5.1一致性事務的設計

Storm如何實現即對tuple並行處理，又保證事務性。本節從簡單的事務性實現方法入手，逐步引出Transactional Topology的原理。分佈式

5.1.1 簡單設計一：強順序流

保證tuple只被處理一次，最簡單的方法就是將tuple流變成強順序的，而且每次只處理一個tuple。從1開始，給每一個tuple都順序加上一個id。在處理tuple的時候，將處理成功的tuple id和計算結果存在數據庫中。下一個tuple到來的時候，將其id與數據庫中的id作比較。若是相同，則說明這個tuple已經被成功處理過了，忽略它；若是不一樣，根據強順序性，說明這個tuple沒有被處理過，將它的id及計算結果更新到數據庫中。ide

以統計消息總數爲例。每來一個tuple，若是數據庫中存儲的id 與當前tuple id不一樣，則數據庫中的消息總數加1，同時更新數據庫中的當前tuple id值。如圖：ui

可是這種機制使得系統一次只能處理一個tuple，沒法實現分佈式計算。spa

5.1.2 簡單設計二：強順序batch流

爲了實現分佈式，咱們能夠每次處理一批tuple，稱爲一個batch。一個batch中的tuple能夠被並行處理。設計

咱們要保證一個batch只被處理一次，機制和上一節相似。只不過數據庫中存儲的是batch id。batch的中間計算結果先存在局部變量中，當一個batch中的全部tuple都被處理完以後，判斷batch id，若是跟數據庫中的id不一樣，則將中間計算結果更新到數據庫中。orm

如何確保一個batch裏面的全部tuple都被處理完了呢？能夠利用Storm提供的CoordinateBolt。如圖：對象

可是強順序batch流也有侷限，每次只能處理一個batch，batch之間沒法並行。要想實現真正的分佈式事務處理，可使用storm提供的Transactional Topology。在此以前，咱們先詳細介紹一下CoordinateBolt的原理。blog

5.1.3 CoordinateBolt原理

CoordinateBolt具體原理以下：

真正執行計算的bolt外面封裝了一個CoordinateBolt。真正執行任務的bolt咱們稱爲real bolt。
每一個CoordinateBolt記錄兩個值：有哪些task給我發送了tuple（根據topology的grouping信息）；我要給哪些tuple發送信息（一樣根據groping信息）
Real bolt發出一個tuple後，其外層的CoordinateBolt會記錄下這個tuple發送給哪一個task了。
等全部的tuple都發送完了以後，CoordinateBolt經過另一個特殊的stream以emitDirect的方式告訴全部它發送過 tuple的task，它發送了多少tuple給這個task。下游task會將這個數字和本身已經接收到的tuple數量作對比，若是相等，則說明處理完了全部的tuple。
下游CoordinateBolt會重複上面的步驟，通知其下游。

整個過程如圖所示：

CoordinateBolt主要用於兩個場景：

DRPC
Transactional Topology

CoordinatedBolt對於業務是有侵入的，要使用CoordinatedBolt提供的功能，你必需要保證你的每一個bolt發送的每一個 tuple的第一個field是request-id。所謂的「我已經處理完個人上游」的意思是說當前這個bolt對於當前這個request-id所須要作的工做作完了。這個request-id在DRPC 裏面表明一個DRPC請求；在Transactional Topology裏面表明一個batch。

5.1.4 Trasactional Topology

Storm提供的Transactional Topology將batch計算分爲process和commit兩個階段。Process階段能夠同時處理多個batch，不用保證順序性；commit階段保證batch的強順序性，而且一次只能處理一個batch，第1個batch成功提交以前，第2個batch不能被提交。

仍是以統計消息總數爲例，如下代碼來自storm-starter裏面的TransactionalGlobalCount。

MemoryTransactionalSpout spout = new MemoryTransactionalSpout(DATA,new Fields(「word「), PARTITION_TAKE_PER_BATCH);

TransactionalTopologyBuilder builder = new TransactionalTopologyBuilder(「global-count「, 「spout「, spout, 3);

builder.setBolt(「partial-count「, new BatchCount(), 5).noneGrouping(「spout「);

builder.setBolt(「sum「, new UpdateGlobalCount()).globalGrouping(「partial-count「);

TransactionalTopologyBuilder共接收四個參數。

這個Transactional Topology的id。Id用來在Zookeeper中保存當前topology的進度，若是這個topology重啓，能夠繼續以前的進度執行。
Spout在這個topology中的id
一個TransactionalSpout。一個Trasactional Topology中只能有一個TrasactionalSpout.在本例中是一個MemoryTransactionalSpout，從一個內存變量（DATA）中讀取數據。
TransactionalSpout的並行度（可選）。

下面是BatchCount的定義:

public static class BatchCount extends BaseBatchBolt {

Object _id;

BatchOutputCollector _collector;

int _count = 0;

@Override

public void prepare(Map conf, TopologyContext context,

BatchOutputCollector collector, Object id) {

_collector = collector;

_id = id;

}

@Override

public void execute(Tuple tuple) {

_count++;

}

@Override

public void finishBatch() {

_collector.emit(new Values(_id, _count));

}

@Override

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new Fields(「id「, 「count「));

}

BatchCount的prepare方法的最後一個參數是batch id，在Transactional Tolpoloyg裏面這id是一個TransactionAttempt對象。

Transactional Topology裏發送的tuple都必須以TransactionAttempt做爲第一個field，storm根據這個field來判斷tuple屬於哪個batch。

TransactionAttempt包含兩個值：一個transaction id，一個attempt id。transaction id的做用就是咱們上面介紹的對於每一個batch中的tuple是惟一的，並且無論這個batch replay多少次都是同樣的。attempt id是對於每一個batch惟一的一個id，可是對於同一個batch，它replay以後的attempt id跟replay以前就不同了，咱們能夠把attempt id理解成replay-times， storm利用這個id來區別一個batch發射的tuple的不一樣版本。

execute方法會爲batch裏面的每一個tuple執行一次，你應該把這個batch裏面的計算狀態保持在一個本地變量裏面。對於這個例子來講，它在execute方法裏面遞增tuple的個數。

最後，當這個bolt接收到某個batch的全部的tuple以後， finishBatch方法會被調用。這個例子裏面的BatchCount類會在這個時候發射它的局部數量到它的輸出流裏面去。

下面是UpdateGlobalCount類的定義：

public static class UpdateGlobalCount extends BaseTransactionalBolt

implements ICommitter {

TransactionAttempt _attempt;

BatchOutputCollector _collector;

int _sum = 0;

@Override

public void prepare(Map conf, TopologyContext context,

BatchOutputCollector collector, TransactionAttempt attempt) {

_collector = collector;

_attempt = attempt;

}

@Override

public void execute(Tuple tuple) {

_sum+=tuple.getInteger(1);

}

@Override

public void finishBatch() {

Value val = DATABASE.get(GLOBAL_COUNT_KEY);

Value newval;

if(val == null || !val.txid.equals(_attempt.getTransactionId())) {

newval = new Value();

newval.txid = _attempt.getTransactionId();

if(val==null) {

newval.count = _sum;

} else {

newval.count = _sum + val.count;

}

DATABASE.put(GLOBAL_COUNT_KEY, newval);

} else {

newval = val;

}

_collector.emit(new Values(_attempt, newval.count));

}

@Override

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new Fields(「id「, 「sum「));

}

UpdateGlobalCount實現了ICommitter接口，因此storm只會在commit階段執行finishBatch方法。而execute方法能夠在任何階段完成。

在UpdateGlobalCount的finishBatch方法中，將當前的transaction id與數據庫中存儲的id作比較。若是相同，則忽略這個batch；若是不一樣，則把這個batch的計算結果加到總結果中，並更新數據庫。

Transactional Topolgy運行示意圖以下：

下面總結一下Transactional Topology的一些特性

Transactional Topology將事務性機制都封裝好了，其內部使用CoordinateBolt來保證一個batch中的tuple被處理完。
TransactionalSpout只能有一個，它將全部tuple分爲一個一個的batch，並且保證同一個batch的transaction id始終同樣。
BatchBolt處理batch在一塊兒的tuples。對於每個tuple調用execute方法，而在整個batch處理完成的時候調用finishBatch方法。
若是BatchBolt被標記成Committer，則只能在commit階段調用finishBolt方法。一個batch的commit階段由storm保證只在前一個batch成功提交以後纔會執行。而且它會重試直到topology裏面的全部bolt在commit完成提交。
Transactional Topology隱藏了anchor/ack框架，它提供一個不一樣的機制來fail一個batch，從而使得這個batch被replay。