Storm入門教程 第五章 一致性事務(wù)
Storm是一個(gè)分布式的流處理系統(tǒng)，利用anchor和ack機(jī)制保證所有tuple都被成功處理。如果tuple出錯(cuò)，則可以被重傳，但是如何 保證出錯(cuò)的tuple只被處理一次呢？Storm提供了一套事務(wù)性組件Transaction Topology，用來(lái)解決這個(gè)問(wèn)題。
Transactional Topology目前已經(jīng)不再維護(hù)，由Trident來(lái)實(shí)現(xiàn)事務(wù)性topology，但是原理相同。
5.1一致性事務(wù)的設(shè)計(jì)
Storm如何實(shí)現(xiàn)即對(duì)tuple并行處理，又保證事務(wù)性。本節(jié)從簡(jiǎn)單的事務(wù)性實(shí)現(xiàn)方法入手，逐步引出Transactional Topology的原理。
5.1.1 簡(jiǎn)單設(shè)計(jì)一：強(qiáng)順序流
保證tuple只被處理一次，最簡(jiǎn)單的方法就是將tuple流變成強(qiáng)順序的，并且每次只處理一個(gè)tuple。從1開(kāi)始，給每個(gè)tuple都順序加上 一個(gè)id。在處理tuple的時(shí)候，將處理成功的tuple id和計(jì)算結(jié)果存在數(shù)據(jù)庫(kù)中。下一個(gè)tuple到來(lái)的時(shí)候，將其id與數(shù)據(jù)庫(kù)中的id做比較。如果相同，則說(shuō)明這個(gè)tuple已經(jīng)被成功處理過(guò)了，忽略 它；如果不同，根據(jù)強(qiáng)順序性，說(shuō)明這個(gè)tuple沒(méi)有被處理過(guò)，將它的id及計(jì)算結(jié)果更新到數(shù)據(jù)庫(kù)中。
以統(tǒng)計(jì)消息總數(shù)為例。每來(lái)一個(gè)tuple，如果數(shù)據(jù)庫(kù)中存儲(chǔ)的id 與當(dāng)前tuple id不同，則數(shù)據(jù)庫(kù)中的消息總數(shù)加1，同時(shí)更新數(shù)據(jù)庫(kù)中的當(dāng)前tuple id值。如圖：
但是這種機(jī)制使得系統(tǒng)一次只能處理一個(gè)tuple，無(wú)法實(shí)現(xiàn)分布式計(jì)算。
5.1.2 簡(jiǎn)單設(shè)計(jì)二：強(qiáng)順序batch流
為了實(shí)現(xiàn)分布式，我們可以每次處理一批tuple，稱為一個(gè)batch。一個(gè)batch中的tuple可以被并行處理。
我們要保證一個(gè)batch只被處理一次，機(jī)制和上一節(jié)類似。只不過(guò)數(shù)據(jù)庫(kù)中存儲(chǔ)的是batch id。batch的中間計(jì)算結(jié)果先存在局部變量中，當(dāng)一個(gè)batch中的所有tuple都被處理完之后，判斷batch id，如果跟數(shù)據(jù)庫(kù)中的id不同，則將中間計(jì)算結(jié)果更新到數(shù)據(jù)庫(kù)中。
如何確保一個(gè)batch里面的所有tuple都被處理完了呢？可以利用Storm提供的CoordinateBolt。如圖：
但是強(qiáng)順序batch流也有局限，每次只能處理一個(gè)batch，batch之間無(wú)法并行。要想實(shí)現(xiàn)真正的分布式事務(wù)處理，可以使用storm提供的Transactional Topology。在此之前，我們先詳細(xì)介紹一下CoordinateBolt的原理。
5.1.3 CoordinateBolt原理
CoordinateBolt具體原理如下：
- 真正執(zhí)行計(jì)算的bolt外面封裝了一個(gè)CoordinateBolt。真正執(zhí)行任務(wù)的bolt我們稱為real bolt。
- 每個(gè)CoordinateBolt記錄兩個(gè)值：有哪些task給我發(fā)送了tuple（根據(jù)topology的grouping信息）；我要給哪些tuple發(fā)送信息（同樣根據(jù)groping信息）
- Real bolt發(fā)出一個(gè)tuple后，其外層的CoordinateBolt會(huì)記錄下這個(gè)tuple發(fā)送給哪個(gè)task了。
- 等所有的tuple都發(fā)送完了之后，CoordinateBolt通過(guò)另外一個(gè)特殊的stream以emitDirect的方式告訴所有它發(fā)送過(guò) tuple的task，它發(fā)送了多少tuple給這個(gè)task。下游task會(huì)將這個(gè)數(shù)字和自己已經(jīng)接收到的tuple數(shù)量做對(duì)比，如果相等，則說(shuō)明處理 完了所有的tuple。
- 下游CoordinateBolt會(huì)重復(fù)上面的步驟，通知其下游。
整個(gè)過(guò)程如圖所示：
CoordinateBolt主要用于兩個(gè)場(chǎng)景：
- DRPC
- Transactional Topology
CoordinatedBolt對(duì)于業(yè)務(wù)是有侵入的，要使用CoordinatedBolt提供的功能，你必須要保證你的每個(gè)bolt發(fā)送的每個(gè) tuple的***個(gè)field是request-id。 所謂的“我已經(jīng)處理完我的上游”的意思是說(shuō)當(dāng)前這個(gè)bolt對(duì)于當(dāng)前這個(gè)request-id所需要做的工作做完了。這個(gè)request-id在DRPC 里面代表一個(gè)DRPC請(qǐng)求；在Transactional Topology里面代表一個(gè)batch。
#p#
5.1.4 Trasactional Topology
Storm提供的Transactional Topology將batch計(jì)算分為process和commit兩個(gè)階段。Process階段可以同時(shí)處理多個(gè)batch，不用保證順序 性；commit階段保證batch的強(qiáng)順序性，并且一次只能處理一個(gè)batch，第1個(gè)batch成功提交之前，第2個(gè)batch不能被提交。
還是以統(tǒng)計(jì)消息總數(shù)為例，以下代碼來(lái)自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共接收四個(gè)參數(shù)。
- 這個(gè)Transactional Topology的id。Id用來(lái)在Zookeeper中保存當(dāng)前topology的進(jìn)度，如果這個(gè)topology重啟，可以繼續(xù)之前的進(jìn)度執(zhí)行。
- Spout在這個(gè)topology中的id
- 一個(gè)TransactionalSpout。一個(gè)Trasactional Topology中只能有一個(gè)TrasactionalSpout.在本例中是一個(gè)MemoryTransactionalSpout，從一個(gè)內(nèi)存變量（DATA）中讀取數(shù)據(jù)。
- 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方法的***一個(gè)參數(shù)是batch id，在Transactional Tolpoloyg里面這id是一個(gè)TransactionAttempt對(duì)象。
Transactional Topology里發(fā)送的tuple都必須以TransactionAttempt作為***個(gè)field，storm根據(jù)這個(gè)field來(lái)判斷tuple屬于哪一個(gè)batch。
TransactionAttempt包含兩個(gè)值：一個(gè)transaction id，一個(gè)attempt id。transaction id的作用就是我們上面介紹的對(duì)于每個(gè)batch中的tuple是唯一的，而且不管這個(gè)batch replay多少次都是一樣的。attempt id是對(duì)于每個(gè)batch唯一的一個(gè)id， 但是對(duì)于同一個(gè)batch，它replay之后的attempt id跟replay之前就不一樣了， 我們可以把a(bǔ)ttempt id理解成replay-times， storm利用這個(gè)id來(lái)區(qū)別一個(gè)batch發(fā)射的tuple的不同版本。
execute方法會(huì)為batch里面的每個(gè)tuple執(zhí)行一次，你應(yīng)該把這個(gè)batch里面的計(jì)算狀態(tài)保持在一個(gè)本地變量里面。對(duì)于這個(gè)例子來(lái)說(shuō)， 它在execute方法里面遞增tuple的個(gè)數(shù)。
***， 當(dāng)這個(gè)bolt接收到某個(gè)batch的所有的tuple之后， finishBatch方法會(huì)被調(diào)用。這個(gè)例子里面的BatchCount類會(huì)在這個(gè)時(shí)候發(fā)射它的局部數(shù)量到它的輸出流里面去。
#p#
下面是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實(shí)現(xiàn)了ICommitter接口，所以storm只會(huì)在commit階段執(zhí)行finishBatch方法。而execute方法可以在任何階段完成。
在UpdateGlobalCount的finishBatch方法中，將當(dāng)前的transaction id與數(shù)據(jù)庫(kù)中存儲(chǔ)的id做比較。如果相同，則忽略這個(gè)batch；如果不同，則把這個(gè)batch的計(jì)算結(jié)果加到總結(jié)果中，并更新數(shù)據(jù)庫(kù)。
Transactional Topolgy運(yùn)行示意圖如下：
下面總結(jié)一下Transactional Topology的一些特性
- Transactional Topology將事務(wù)性機(jī)制都封裝好了，其內(nèi)部使用CoordinateBolt來(lái)保證一個(gè)batch中的tuple被處理完。
- TransactionalSpout只能有一個(gè)，它將所有tuple分為一個(gè)一個(gè)的batch，而且保證同一個(gè)batch的transaction id始終一樣。
- BatchBolt處理batch在一起的tuples。對(duì)于每一個(gè)tuple調(diào)用execute方法，而在整個(gè)batch處理完成的時(shí)候調(diào)用finishBatch方法。
- 如果BatchBolt被標(biāo)記成Committer，則只能在commit階段調(diào)用finishBolt方法。一個(gè)batch的commit階 段由storm保證只在前一個(gè)batch成功提交之后才會(huì)執(zhí)行。并且它會(huì)重試直到topology里面的所有bolt在commit完成提交。
- Transactional Topology隱藏了anchor/ack框架，它提供一個(gè)不同的機(jī)制來(lái)fail一個(gè)batch，從而使得這個(gè)batch被replay。
5.2 Trident介紹
Trident是Storm之上的高級(jí)抽象，提供了joins，grouping，aggregations，fuctions和filters等接口。如果你使用過(guò)Pig或Cascading，對(duì)這些接口就不會(huì)陌生。
Trident將stream中的tuples分成batches進(jìn)行處理，API封裝了對(duì)這些batches的處理過(guò)程，保證tuple只被處理一次。處理batches中間結(jié)果存儲(chǔ)在TridentState對(duì)象中。
Trident事務(wù)性原理這里不詳細(xì)介紹，有興趣的讀者請(qǐng)自行查閱資料。
參考：http://xumingming.sinaapp.com/736/twitter-storm-transactional-topolgoy/
http://xumingming.sinaapp.com/811/twitter-storm-code-analysis-coordinated-bolt/
https://github.com/nathanmarz/storm/wiki/Trident-tutorial
