Storm的Transactional Topology怎么配置

這篇文章主要講解了“Storm的Transactional Topology怎么配置”，文中的講解內(nèi)容簡單清晰，易于學(xué)習(xí)與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“Storm的Transactional Topology怎么配置”吧！

1、什么是Transactional Topology？

    ○ 是一個每個tuple僅被處理一次的框架

    ○ 由Storm0.7引入，于Storm0.9被棄用，被triden取而代之

    ○ 底層依靠spout\bolt\topology\stream抽象的一個特性

2、Transactional Topology設(shè)計思路

    ○ 一次只處理一次tuple

        基于Storm處理tuple失敗時會重發(fā)(replay)，如何確保replay的記錄不被重復(fù)記錄，換句話說就是如何保證tuple僅被處理一次，這就依賴于一個稱作強(qiáng)順序性的思想。

        強(qiáng)順序性：每個tuple與一個transaction id相關(guān)聯(lián)，transaction id實際就是一個數(shù)字，每一個tuple都有一個按照順序的transaction id（例如：tuple1的transaction id 為 1,tuple2的transaction id 為 2,...以此類推），只有當(dāng)前的tuple處理并存儲完畢，下一個tuple（處于等待狀態(tài)）才能進(jìn)行存儲，tuple被存儲時連同transaction id一并存儲，此時考慮兩種情況：

                        tuple處理失敗時：重新發(fā)送一個和原來一模一樣的transaction id

                        tuple處理成功時：發(fā)送的transaction id會和存儲的transaction id對比，如果不存在transaction id，表示第一次記錄，直接存儲；如果發(fā)現(xiàn)存在，則忽略該tuple。

        這一思想是由Kafka開發(fā)者提出來的。

    ○ 一次處理一批tuple

        基于上面的一個優(yōu)化，將一批tuple直接打包成一個batch，然后分配一個transaction id ，讓batch與batch之間保證強(qiáng)順序性，且batch內(nèi)部的tuples可以并行。

    ○ Storm是如何采用的？

        兩個步驟：

            1、并行計算batch中的tuple數(shù)量

            2、batch強(qiáng)順序性存儲

            在batch強(qiáng)順序性存儲的同時讓其他等待存儲的batch內(nèi)部進(jìn)行并行運算，不必等到下一個batch存儲時才進(jìn)行內(nèi)部運算。

        在Storm上面的兩個步驟表現(xiàn)為processing階段和commit階段。

3、一些設(shè)計細(xì)節(jié)

使用Transactional Topology時,storm提供如下操作：

    ○ 管理狀態(tài)

        將需要處理的狀態(tài)如：transaction id 、batch meta等狀態(tài)信息放在zookeeper

    ○ 協(xié)調(diào)事務(wù)

        指定某個時間段執(zhí)行processing操作和commit操作

    ○ 錯誤檢測

        storm使用acking框架自動檢測batch被成功或失敗處理，然后相應(yīng)的重發(fā)（replay）

    ○ 內(nèi)置批處理API

        通過對普通的bolt進(jìn)行包裝，提供一套對batch處理的API、協(xié)調(diào)工作（即某個時刻處理某個processing或者commit），并且storm會自動清除中間結(jié)果

Transactional Topology是可以完全重發(fā)一個特定batch的消息隊列系統(tǒng)，在 Kakfa中正是有這樣的需求，為此Storm在storm-contrib里面的Storm-Kafka中為Kafka實現(xiàn)了一個事務(wù)性的spout。

4、來自Storm-Starter.jar的例子

    計算來自輸入流中tuple的個數(shù)

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)
        .shuffleGrouping("spout");
builder.setBolt("sum", new UpdateGlobalCount())
        .globalGrouping("partial-count");

    ○ 通過TransactionalTopologyBuilder類構(gòu)建Transactional

        參數(shù)：

        Transaction ID：transactional topology的ID，在zookeeper中用于保存進(jìn)度狀態(tài)，重啟topology時可以直接從執(zhí)行的進(jìn)度開始執(zhí)行而不用重頭到尾又執(zhí)行一遍

        Spout ID：位于整個Topology的Spout的ID

        Spout Object：Transactional中的Spout對象

        Spout：Trasactional中的Spout的并行數(shù)

    ○ MemoryTransactionalSpout用于從一個內(nèi)存變量中讀取數(shù)據(jù)

        DATA：數(shù)據(jù)

        tuple fields：字段

        tupleNum：在batch中最大的tuple數(shù)

    ○ Bolts

        第一個Bolt采用隨機(jī)分組的方式隨機(jī)分發(fā)到各個task

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"));
    }
}

        BatchBolt對象運行在BatchBoltExecutor中，BatchBoltExecutor負(fù)責(zé)BatchBolt對象的創(chuàng)建和清理 

        BatchBolt的ID在context對象中，該ID是一個TransactionAttempt對象.

        BatchBolt在DRPC中也可以使用，只是txid類型不一樣，如果在Transactional Topology中使用BatchBolt，可以繼承BaseTransactionalBolt.

        在Tranasctional Topology中所有的Tuple都必須以TransactionAttempt作為第一個field，然后storm才能根據(jù)該field判斷Tuple所屬的BatchBolt，所以在發(fā)射Tuple必須滿足此條件。

        TransactionAttempt對象中有兩個屬性：

            transaction id：強(qiáng)順序性，無論重發(fā)多少次都是一樣的數(shù)字

            attempt id：對每一個Batch標(biāo)識的ID，每次重發(fā)都其值不一致，通過該ID可以區(qū)分每次重發(fā)的Tuple的不同版本

第二個Bolt使用GlobalGrouping匯總batch中的tuple數(shù)

 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"));
    }
}

    ICommitter接口：實現(xiàn)該接口的Bolt會在commit階段調(diào)用finishBatch方法，該方法的調(diào)用會按照強(qiáng)順序性，此外還可以使用TransactionalTopologyBuilder的setCommiterBolt來添加Bolt實現(xiàn)和該接口一樣的功能。

    executor方法：在processing階段和commit階段都可以執(zhí)行。

    關(guān)于更多的transactional topology例子可以看看storm-starter中的TransactionalWords類，該例子會在一個事務(wù)中更新多個數(shù)據(jù)庫

5、Transaction Topology API

    Bolt類

    BaiscBolt：該Bolt不跟batch中的tuples交互，僅基于單個傳來的tuple和產(chǎn)生新的tuple

    BatchBolt：該Bolt處理batch中的tuples，對于每一個tuple調(diào)用executor方法，整個batch完成時調(diào)用finishBatch方法

    被Committer標(biāo)記的Bolt：在commit階段才調(diào)用finishBatch方法，commit具有強(qiáng)順序性，標(biāo)記Bolt為commit階段執(zhí)行finishBatch的方法有兩種：1、實現(xiàn)ICommiter接口。2、TransactionalTopologyBuilder的setCommiterBolt來添加Bolt。

    Processing階段和Commit階段

    紅色輪廓的Bolt被標(biāo)記過為commit

    Spout向Bolt A發(fā)送整個Batch

    Bolt A處理完整個Batch之后調(diào)用finishBatch方法分別向Bolt B 和 Bolt C發(fā)送Batch

    Bolt B接收到Bolt A傳遞過來的tuple進(jìn)行處理（此時還尚未處理完畢）不會調(diào)用finishBatch方法

    Bolt C接口Bolt A傳遞的tuple，盡管處理完Bolt A傳遞來的tuple，但是由于Bolt B還尚未commit，所以Bolt C處于等待Bolt B commit的狀態(tài)，不會調(diào)用finishBatch方法

    Bolt D接收來自Bolt C調(diào)用executor方法時發(fā)送的所有tuple

    此時一旦Bolt B進(jìn)行commit進(jìn)行finishBatch操作，那么Bolt C就會確認(rèn)接收到所有Bolt B的tuple，Bolt C也調(diào)用finishBatch方法，最終Bolt D也接收到所有來自Bolt C的batch。

    在這里盡管Bolt D是一個committer，它在接收到整個batch的tuple之后不需要等待第二個commit信號。因為它是在commit階段接收到的整個batch，它會調(diào)用finishBatch來完成整個事務(wù)。

    Acking

注意，當(dāng)使用transactional topology的時候你不需要顯式地去做任何的acking或者anchoring，storm在背后都做掉了。（storm對transactional topolgies里面的acking機(jī)制進(jìn)行了高度的優(yōu)化)

    Failing a transaction

在使用普通bolt的時候， 你可以通過調(diào)用OutputCollector的fail方法來fail這個tuple所在的tuple樹。Transactional Topology對用戶隱藏了acking框架， 它提供一個不同的機(jī)制來fail一個batch（從而使得這個batch被replay）：只要拋出一個FailedException就可以了。跟普通的異常不一樣， 這個異常只會導(dǎo)致當(dāng)前的batch被replay, 而不會使整個進(jìn)程崩潰掉。

    Transactional spout

TransactionalSpout接口跟普通的Spout接口完全不一樣。一個TransactionalSpout的實現(xiàn)會發(fā)送一批一批（batch）的tuple， 而且必須保證同一批次tuples的transaction id始終一樣。

在transactional topology運行的時候， transactional spout看起來是這樣的一個結(jié)構(gòu)：

coordinator是一個普通的storm的spout——它一直為事務(wù)的batch發(fā)射tuple。

Emitter則像一個普通的storm bolt，它負(fù)責(zé)為每個batch實際發(fā)射tuple，emitter以all grouping的方式訂閱coordinator的”batch emit”流。
關(guān)于如何實現(xiàn)一個TransactionalSpout的細(xì)節(jié)可以參見Javadoc

    Partitioned Transactional Spout

一種常見的TransactionalSpout是那種從多個queue broker讀取數(shù)據(jù)然后再發(fā)射的tuple。比如TransactionalKafkaSpout就是這樣工作的。IPartitionedTransactionalSpout把這些管理每個分區(qū)的狀態(tài)以保證可以replay的冪等性的工作都自動化掉了。更多可以參考Javadoc。

    配置

Transactional Topologies有兩個重要的配置：

Zookeeper：默認(rèn)情況下，transactional topology會把狀態(tài)信息保存在一個zookeeper里面（協(xié)調(diào)集群的那個)。你可以通過這兩個配置來指定其它的zookeeper：”transactional.zookeeper.servers” 和 “transactional.zookeeper.port“。

同時活躍的batch數(shù)量：你必須設(shè)置同時處理的batch數(shù)量，你可以通過”topology.max.spout.pending” 來指定， 如果你不指定，默認(rèn)是1。

6、實現(xiàn)

Transactional Topologies的實現(xiàn)是非常優(yōu)雅的。管理提交協(xié)議，檢測失敗并且串行提交看起來很復(fù)雜，但是使用storm的原語來進(jìn)行抽象是非常簡單的。

1、transactional spout是一個子topology, 它由一個coordinator spout和一個emitter bolt組成。

2、coordinator是一個普通的spout，并行度為1；emitter是一個bolt，并行度為P，使用all分組方式連接到coordinator的“batch”流上。

3、coordinator使用一個acking框架決定什么時候一個batch被成功執(zhí)行（process）完成，然后去決定一個batch什么時候被成功提交（commit）。

感謝各位的閱讀，以上就是“Storm的Transactional Topology怎么配置”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對Storm的Transactional Topology怎么配置這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關(guān)知識點的文章，歡迎關(guān)注！