当前访客身份:游客 [ 登录  | 注册加入尚学堂]
直播

我来了!

拥有积分:3901
尚学堂雄起!!威武。。。

博客分类

笔记中心

课题中心

提问中心

答题中心

解答题中心

一致性事务(2)

我来了! 发表于 2年前 (2014-10-29 10:59:08)  |  评论(0)  |  阅读次数(447)| 0 人收藏此文章,   我要收藏   

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类会在这个时候发射它的局部数量到它的输出流里面去。

分享到:0
关注微信,跟着我们扩展技术视野。每天推送IT新技术文章,每周聚焦一门新技术。微信二维码如下:
微信公众账号:尚学堂(微信号:bjsxt-java)
声明:博客文章版权属于原创作者,受法律保护。如果侵犯了您的权利,请联系管理员,我们将及时删除!
(邮箱:webmaster#sxt.cn(#换为@))
北京总部地址:北京市海淀区西三旗桥东建材城西路85号神州科技园B座三层尚学堂 咨询电话:400-009-1906 010-56233821
Copyright 2007-2015 北京尚学堂科技有限公司 京ICP备13018289号-1 京公网安备11010802015183