文档章节

Storm入门 第五章 一致性事务

坏坏一笑
 坏坏一笑
发布于 2014/12/03 11:14
字数 2179
阅读 80
收藏 7

第五章 一致性事务

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及计算结果更新到数据库中。

以统计消息总数为例。每来一个tuple,如果数据库中存储的id 与当前tuple id不同,则数据库中的消息总数加1,同时更新数据库中的当前tuple id值。如图:

强顺序流

         但是这种机制使得系统一次只能处理一个tuple,无法实现分布式计算。

5.1.2 简单设计二:强顺序batch流

为了实现分布式,我们可以每次处理一批tuple,称为一个batch。一个batch中的tuple可以被并行处理。

我们要保证一个batch只被处理一次,机制和上一节类似。只不过数据库中存储的是batch id。batch的中间计算结果先存在局部变量中,当一个batch中的所有tuple都被处理完之后,判断batch id,如果跟数据库中的id不同,则将中间计算结果更新到数据库中。

如何确保一个batch里面的所有tuple都被处理完了呢?可以利用Storm提供的CoordinateBolt。如图:

顺序batches

但是强顺序batch流也有局限,每次只能处理一个batch,batch之间无法并行。要想实现真正的分布式事务处理,可以使用storm提供的Transactional Topology。在此之前,我们先详细介绍一下CoordinateBolt的原理。

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

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的一些特性

  •  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。

5.2 Trident介绍

Trident是Storm之上的高级抽象,提供了joins,grouping,aggregations,fuctions和filters等接口。如果你使用过Pig或Cascading,对这些接口就不会陌生。

Trident将stream中的tuples分成batches进行处理,API封装了对这些batches的处理过程,保证tuple只被处理一次。处理batches中间结果存储在TridentState对象中。

Trident事务性原理这里不详细介绍,有兴趣的读者请自行查阅资料。

参考: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

本文转载自:http://blog.linezing.com/?p=1976

坏坏一笑
粉丝 10
博文 54
码字总数 29772
作品 0
昌平
程序员
私信 提问
Storm 是如何跟踪一条消息以及它衍生出来的消息都被成功处理的

2012·2汇总 我们做 Notify Server 时可以间接借鉴这个解决方案的思路。 Storm 是一个分布式的、容错的实时计算系统,由 Twitter 开源。 先不介绍术语和背景知识,直接来一些吸引眼球的内容:...

旁观者-郑昀
2013/02/08
292
0
探秘Hadoop生态13:初探Storm和入门实例

这位大侠,这是我的公众号:程序员江湖。 分享程序员面试与技术的那些事。 干货满满,关注就送。 Storm:最火的流式处理框架 伴随着信息科技日新月异的发展,信息呈现出爆发式的膨胀,人们获...

你的猫大哥
2017/03/08
0
0
Storm 并非完全适合所有实时应用

流数据的迅速崛起带来了一类全新的应用开发技术。为了应对不断增长的数据(如物联网和机器通信产生的大量数据),同时,利用实时个性化技术改进在线用户体验,越来越多的应用开发引入了流数据...

亚当李
2014/12/18
15K
9
Storm入门 第二章 构建Topology

2.1 Storm基本概念 在运行一个Storm任务之前,需要了解一些概念: Topologies Streams Spouts Bolts Stream groupings Reliability Tasks Workers Configuration Storm集群和Hadoop集群表面上...

坏坏一笑
2014/12/03
84
0
Twitter Storm Ubuntu 单机安装

第 121 章 Twitter Storm 目录 121.1. 单机版121.2. lein 安装 121.1. 单机版 操作系统环境:Ubuntu 13.04 KVM虚拟机 安装 storm 涉及到安装以下包:python、zookeeper、zeromq、jzmq、storm...

netkiller-
2013/08/02
2K
0

没有更多内容

加载失败,请刷新页面

加载更多

Docker 可视化管理 portainer

官网安装指南: https://portainer.readthedocs.io/en/latest/deployment.html docker-compose.yml 位置,下载地址:https://downloads.portainer.io/docker-compose.yml...

Moks角木
38分钟前
5
0
Spring Security 实战干货:必须掌握的一些内置 Filter

1. 前言 上一文我们使用 Spring Security 实现了各种登录聚合的场面。其中我们是通过在 UsernamePasswordAuthenticationFilter 之前一个自定义的过滤器实现的。我怎么知道自定义过滤器要加在...

码农小胖哥
41分钟前
7
0
常见分布式事务解决方案

1 微服务的发展 微服务倡导将复杂的单体应用拆分为若干个功能简单、松耦合的服务,这样可以降低开发难度、增强扩展性、便于敏捷开发。当前被越来越多的开发者推崇,很多互联网行业巨头、开源...

asdf08442a
42分钟前
3
0
influxdb continuous queries(cq)从入门到放弃

从前一篇influxdb的文章prometheus基于influxdb的监控数据持久化存储方案完成之后,就一直在折腾influxdb发布测试和生产环境的问题,经过接近2个月的验证,最终发现使用influxdb自带cq的方案...

狗陈
53分钟前
7
0
7.线程通信

在现实生活中,如果一个人团队正在共同完成任务,那么他们之间应该有通信,以便正确完成任务。 同样的比喻也适用于线程。 在编程中,要减少处理器的理想时间,我们创建了多个线程,并为每个线...

Eappo_Geng
59分钟前
6
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部