Transactionaltopology怎么使用

本篇内容介绍了“Transactional topology怎么使用”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

公司主营业务：做网站、网站设计、移动网站开发等业务。帮助企业客户真正实现互联网宣传，提高企业的竞争能力。成都创新互联是一支青春激扬、勤奋敬业、活力青春激扬、勤奋敬业、活力澎湃、和谐高效的团队。公司秉承以“开放、自由、严谨、自律”为核心的企业文化，感谢他们对我们的高要求，感谢他们从不同领域给我们带来的挑战，让我们激情的团队有机会用头脑与智慧不断的给客户带来惊喜。成都创新互联推出石家庄免费做网站回馈大家。

你可以通过使用TransactionalTopologyBuilder来创建transactional topology. 下面就是一个transactional topology的定义， 它的作用是计算输入流里面的tuple的个数。这段代码来自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)

        .shuffleGrouping("spout");

builder.setBolt("sum", new UpdateGlobalCount())

        .globalGrouping("partial-count");

TransactionalTopologyBuilder构造器中接受如下的参数：

一个transaction topology的id

spout在整个topology里面的id。

一个transactional spout。

一个可选的这个transactional spout的并行度。

topology的id是用来在zookeeper里面保存这个topology的当前进度状态的，所以如果你重启这个topology， 它可以接着前面的进度继续执行。

一个transaction topology里面有一个唯一的TransactionalSpout, 这个spout是通过TransactionalTopologyBuilder的构造函数来指定的。在这个例子里面，MemoryTransactionalSpout被用来从一个内存变量里面读取数据(DATA)。第二个参数指定spout发送的tuple的字段， 第三个参数指定每个batch的最大tuple数量。关于如何自定义TransactionalSpout我们会在后面介绍。

现在说说 bolts。这个topology并行地计算tuple的总数量。第一个bolt：BatchBolt，随机地把输入tuple分给各个task，然后各个task各自统计局部数量。第二个bolt：UpdateGlobalCount, 用全局grouping来汇总这个batch中tuple的数量，然后再更新到数据库里面的全局数量。

下面是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"));

    }

}

storm会为每个正在处理的batch创建一个BatchCount对象，这个BatchCount是运行在BatchBoltExecutor里面的。而BatchBoltExecutor负责创建以及清理这个对象的实例。

BatchCount对象的prepare方法接收如下参数:

Storm config

Topology context

Output collector

这个batch的id (txid)，在Transactional Topology中， 这个id则是一个TransactionAttempt对象。

这个batch bolt的抽象在DRPC里面也可以用， 只是txid的类型不一样而已。实际上，BatchBolt可以接收一个txid类型的参数，所以如果你只是想在transactioinal topology里面使用这个BatchBolt，你可以去继承BaseTransactionalBolt类，如下定义：

public abstract class BaseTransactionalBolt extends BaseBatchBolt {

}

在transaction topology里面发射的所有的tuple都必须以TransactionAttempt作为第一个field， 然后storm可以根据这个field来判断哪些tuple属于一个batch。所以你在发射tuple的时候需要满足这个条件。

TransactionAttempt包含两个值： 一个transaction id，一个attempt id。transaction id的作用就是我们上面介绍的对于每个batch是唯一的，而且不管这个batch 被replay多少次都是一样的。attempt id是对于每个batch唯一的一个id， 但是对于同一个batch，它replay之后的attempt id跟replay之前就不一样了， 我们可以把attempt id理解成replay-times, storm利用这个id来区别一个batch发射的tuple的不同版本。

transaction id对于每个batch加一， 所以第一个batch的transaction id是”1″, 第二个batch是”2″，依次类推。

每收到一个batch中的tuple，execute方法便被调用一次。每次当该方法被调用时，你应该把这个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是Transactional Topologies相关的类，所以它继承自BaseTransactionalBolt。在execute方法里面， UpdateGlobalCount累积这个batch的计数， 比较有趣的是finishBatch方法。

首先， 注意这个bolt实现了ICommitter接口，这告诉storm要在这个事务的commit阶段调用finishBatch方法，所以对于finishBatch的调用会保证强顺序性（顺序就是transaction id的升序)，另一方面execute方法在processing或者commit阶段都可以执行。另外一种把bolt标识为commiter的方法是调用TransactionalTopologyBuilder的setCommiterBolt来添加Bolt（而不是setBolt）。

UpdateGlobalCount里面finishBatch方法的逻辑是首先从数据库中获取当前的值，并且把数据库里面的transaction id与当前这个batch的transaction id进行比较。如果他们一样， 那么忽略这个batch。否则把这个batch的结果加到总结果里面去，并且更新数据库。

“Transactional topology怎么使用”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站，小编将为大家输出更多高质量的实用文章！