Storm: Transactional Topolgoy简介

作者: xumingming | 可以转载,但必须以超链接形式标明文章原始出处和作者信息及版权声明
网址: http://xumingming.sinaapp.com/736/twitter-storm-transactional-topolgoy/

 
本文翻译自: https://github.com/nathanmarz/storm/wiki/Transactional-topologies

概述

Storm通过保证每个tuple至少被处理一次来提供可靠的数据处理。关于这一点最常被问到的问题就是“既然tuple可能会被重写发射(replay),那么我们怎么在storm上面做统计个数之类的事情呢？storm有可能会重复计数吧？”
Storm 0.7.0引入了Transactional Topology,它可以保证每个tuple”被且仅被处理一次”,这样你就可以实现一种非常准确，非常可扩展，并且高度容错方式来实现计数类应用。
跟Distributed RPC类似， transactional topology其实不能算是storm的一个特性，它其实是用storm的底层原语spout,bolt,topology,stream等等抽象出来的一个特性。
这篇文章解释了事务性topology是怎样的一种抽象，怎样使用它的api，同时也讨论了有关它实现的一些细节。

概念

让我们一步步地建立transactional topology的抽象。我们先提出一种最简单的抽象方式， 然后一步步的完善改进，最后介绍storm代码里面所使用的抽象方式。

第一个设计： 最简单的抽象方法

事务性topology背后的核心概念是要在处理数据的提供一个强顺序性。这种强顺序性最简单的表现、同时也是我们第一个设计就是：我们每次只处理一个tuple， 除非这个tuple处理成功，否则我们不去处理下一个tuple。
每一个tuple都跟一个transaction id相关联。如果这个tuple处理失败了，然后需要重写发射，那么它会被重新发射 — 并且附着同样的transaction id。这里说的trasaction id其实就是一个数字， 来一个tuple，它就递增一个。所以第一个tuple的transaction id是1， 第二个tuple的transaction id是2，等等等等。
tuple的强顺序性使得我们即使在tuple重发的时候也能够实现“一次而且只有一次”的语义。 让我们看个例子:
比如你想统一个stream里面tuple的总数。那么为了保证统计数字的准确性，你在数据库里面不但要保存tuple的个数， 还要保存这个数字所对应的最新的transaction id。 当你的代码要到数据库里面去更新这个数字的时候，你要判断只有当新的transaction id跟数据库里面保存的transaction id不一样的时候才去更新。考虑两种情况:

数据库里面的transaction id跟当前的transaction id不一样： 由于我们transaction的强顺序性，我们知道当前的tuple肯定没有统计在数据库里面。所以我们可以安全地递增这个数字，并且更新这个transaction id.

数据库里面的transaction id一样： 那么我们知道当前tuple已经统计在数据库里面了，那么可以忽略这个更新。这个tuple肯定之前在更新了数据库之后，反馈给storm的时候失败了（ack超时之类的）。

这个逻辑以及事务的强顺序性保证数据库里面的个数(count)即使在tuple被重发的时候也是准确的。这个主意（保存count + transaction-id)是Kafka的开发者在这个设计文档里面提出来的。
更进一步来说，这个topology可以在一个事务里面更新很多不同的状态，并且可以到达”一次而且只有一次的逻辑”。如果有任何失败，那么已经成功的更新你再去更新它会忽略，失败的更新你去再次更新它则会接受。比如，如果你在处理一个url流，你可以更新每个url的转发次数， 同时更新每个domain下url的转发次数。
这个简单设计有一个很大的问题， 那就是你需要等待一个tuple完全处理成功之后才能去处理下一个tuple。这个性能是非常差的。这个需要大量的数据库调用（只要每个tuple一个数据库调用),而且这个设计也没有利用到storm的并行计算能力， 所以它的可扩展能力是非常差的。

第二个设计

与每次只处理一个tuple的简单方案相比， 一个更好的方案是每个transaction里面处理一批tuple。所以如果你在做一个计数应用， 那么你每次更新到总数里面的是这一整个batch的tuple数量。如果这个batch失败了，那么你重新replay这整个batch。相应地， 我们不是给每个tuple一个transaction id而是给整个batch一个transaction id，batch与batch之间的处理是强顺序性的， 而batch内部是可以并行的。下面这个是设计图:




所以如果你每个batch处理1000个tuple的话， 那么你的应用将会少1000倍的数据库调用。同时它利用了storm的并行计算能力（每个batch内部可以并行)
虽然这个设计比第一个设计好多了， 它仍然不是一个完美的方案。topology里面的worker会花费大量的时间等待计算的其它部分完成。 比如看下面的这个计算。



在bolt 1完成它的处理之后， 它需要等待剩下的bolt去处理当前batch， 直到发射下一个batch。

第三个设计(storm采用的设计)

一个我们需要意识到的比较重要的问题是，为了实现transactional的特性，在处理一批tuples的时候，不是所有的工作都需要强顺序性的。比如，当做一个全局计数应用的时候， 整个计算可以分为两个部分。

计算这个batch的局部数量。

把这个batch的局部数量更新到数据库里面去。

其中第二步在多个batch之前需要保证强的顺序性， 但是第一步并不许要， 所以我们可以把第一步并行化。所以当第一个batch在更新它的个数进入数据库的时候，第2到10个batch可以开始计算它们的局部数量了。
Storm通过把一个batch的计算分成两个阶段来实现上面所说的原理：

processing阶段： 这个阶段很多batch可以并行计算。

commit阶段： 这个阶段各个batch之间需要有强顺序性的保证。所以第二个batch必须要在第一个batch成功提交之后才能提交。

这两个阶段合起来称为一个transaction。许多batch可以在processing阶段的任何时刻并行计算，但是只有一个batch可以处在commit阶段。如果一个batch在processing或者commit阶段有任何错误， 那么整个transaction需要被replay。

设计细节

当使用Transactional Topologies的时候， storm为你做下面这些事情：
1) 管理状态: Storm把所有实现Transactional Topologies所必须的状态保存在zookeeper里面。 这包括当前transaction id以及定义每个batch的一些元数据。
2) 协调事务: Storm帮你管理所有事情， 以帮你决定在任何一个时间点是该proccessing还是该committing。
3) 错误检测: Storm利用acking框架来高效地检测什么时候一个batch被成功处理了，被成功提交了，或者失败了。Storm然后会相应地replay对应的batch。你不需要自己手动做任何acking或者anchoring — storm帮你搞定所有事情。
4) 内置的批处理API: Storm在普通bolt之上包装了一层API来提供对tuple的批处理支持。Storm管理所有的协调工作，包括决定什么时候一个bolt接收到一个特定transaction的所有tuple。Storm同时也会自动清理每个transaction所产生的中间数据。
5) 最后，需要注意的一点是Transactional Topologies需要一个可以完全重发(replay)一个特定batch的消息的队列系统(Message Queue)。Kestrel之类的技术做不到这一点。而Apache的Kafka对于这个需求来说是正合适的。storm-contrib里面的storm-kafka实现了这个。

一个基本的例子

你可以通过使用TransactionalTopologyBuilder来创建transactional topology. 下面就是一个transactional topology的定义， 它的作用是计算输入流里面的tuple的个数。这段代码来自storm-starter里面的TransactionalGlobalCount。

帮助

1 2 3 4 5 6 7 8

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)。第二个参数制定数据的fields， 第三个参数指定每个batch的最大tuple数量。关于如何自定义

TransactionalSpout

我们会在后面介绍。
现在说说 bolts。这个topology并行地计算tuple的总数量。第一个

bolt：BatchBolt

，随机地把输入tuple分给各个task，然后各个task各自统计局部数量。第二个

bolt：UpdateBlobalCount

,用全局grouping来从汇总这个batch的总的数量。然后再把总的数量更新到数据库里面去。
下面是

BatchCount

的定义:

帮助

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28

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

负责创建以及清理这个对象的实例。
这个对象的prepare方法接收如下参数:

包含storm config信息的map。

TopologyContext

OutputCollector

这个batch的id。而在Transactional Topologies里面， 这个id则是一个TransactionAttempt对象。

这个batch bolt的抽象在DRPC里面也可以用， 只是id的类型不一样而已。BatchBolt其实真的接收一个id类型的参数 — 它是一个java模板类，所以如果你只是想在transactioinal topology里面使用这个BatchBolt，你可以这样定义:

帮助

1 2 3

public abstract class BaseTransactionalBolt         extends BaseBatchBolt<TransactionAttempt> { }

在transaction topology里面发射的所有的tuple都必须以

TransactionAttempt

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

TransactionAttempt

包含两个值： 一个transaction id，一个attempt id。transaction id的作用就是我们上面介绍的对于每个batch是唯一的，而且不管这个batchreplay多少次都是一样的。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″,以此类推。
execute方法会为batch里面的每个tuple执行一次，你应该把这个batch里面的状态保持在一个本地变量里面。对于这个例子来说， 它在execute方法里面递增tuple的个数。
最后， 当这个bolt接收到某个batch的所有的tuple之后， finishBatch方法会被调用。这个例子里面的BatchCount类会在这个时候发射它的局部数量到它的输出流里面去。
下面是

UpdateGlobalCount

类的定义。

帮助

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

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的更深入的例子可以卡看storm-starter里面的TransactionalWords类， 这个类里面会在一个事务里面更新多个数据库。

Transactional Topology API

这一节介绍Transaction topology API

Bolts

在一个transactional topology里面最多有三种类型的bolt:

BasicBolt: 这个bolt不跟batch的tuple打交道，它只基于单个tuple的输入来发射新的tuple。

BatchBolt: 这个bolt处理batch在一起的tuples。对于每一个tuple调用execute方法。而在整个batch处理完成的时候调用finishBatch方法

被标记成Committer的BatchBolt: 和普通的BatchBolt的唯一的区别是finishBatch这个方法被调用的时机。作为committer的BatchBolt的finishBatch方法在commit阶段调用。一个batch的commit阶段由storm保证只在前一个batch成功提交之后才会执行。并且它会重试直到topology里面的所有bolt在commit完成提交。有两个方法可以让一个普通BatchBolt变成committer: 1) 实现ICommitter接口 2) 通过TransactionalTopologyBuilder的setCommitterBolt方法把BatchBolt添加到topology里面去。

Processing phase vs. commit phase in bolts

为了搞清除processing阶段与commit阶段的区别， 让我们看个例子:



在这个topology里面只有用红线标出来的是committers。
在processing阶段， bolt A会处理从spout发射出来的整个batch。并且发射tuple给bolt B和bolt C。Bolt B是一个committer， 所以它会处理所有的tuple， 但是不会调用

finishBatch

方法。Bolt C同样也不会调用

finishBatch

方法， 它的原因是：它不知道它有没有从Bolt B接收到所有的tuple。（因为Bolt B还在等着事务提交）最后Bolt D会接收到Bolt C在调用execute方法的时候发射的所有的tuple。
当batch提交的时候， Bolt B上的

finishBatch

被调用。Bolt C现在可以判断它接收到了所有的tuple， 所以可以调用

finishBatch

了。最后Bolt D接收到了它的所有的tuple所以就调用finishBatch了。
要注意的是，虽然Bolt D是一个committer,它在接收到整个batch的tuple之后不需要等待第二个commit信号。因为它是在commit阶段接收到的整个batch，它会调用finishBatch来完成整个事务。

Acking

注意， 你不需要显式地去做任何的acking或者anchoring。storm在背后都做掉了。（storm对transactional topolgies里面的acking机制进行了高度的优化)

Failing a transaction

在使用普通bolt的时候， 你可以通过调用OutputCollector的fail方法来fail这个tuple所在的tuple树。由于Transactional Topologies把acking框架从用户的视野里面隐藏掉了， 它提供一个不同的机制来fail一个batch（从而使得这个batch被replay）。只要抛出一个FailedException就可以了。跟普通的异常不一样， 这个异常只会导致当前的batch被replay,而不会使整个进程crash掉。

Transactional spout

TransactionalSpout接口跟普通的Spout接口完全不一样。一个TransactionalSpout的实现一个batch一个batch的tuple， 而且必须保证同一个batch的transaction id始终一样。
在transactional topology中运行的时候， transactional spout看起来是这样的一个结构：



在图的左边的coordinator是一个普通的storm的spout — 它一直为事务的batch发射tuple。Emitter则像一个普通的storm bolt，它负责为每个batch实际发射tuple。emitter以all grouping的方式订阅coordinator的”batch emit”流。
由于TransactionalSpout发射的tuple可能需要会被replay， 因此需要具有幂等性（否则多次replay同一个tuple会使得最后的结果不对）， 为了实现幂等性，需要保存Transactional Spout的少量的状态，这个状态是保存在ZooKeeper里面的。
关于如何实现一个

TransactionalSpout

的细节可以参见Javadoc。

Partitioned Transactional Spout

一种常见的TransactionalSpout是那种从多个queue broker夺取数据然后再发射的tuple。比如TransactionalKafkaSpout是这样工作的。

IPartitionedTransactionalSpout

把这些管理每个分区的状态以保证可以replay的幂等性的工作都自动化掉了。更多可以参考Javadoc

配置

Transactional Topologies有两个重要的配置：

Zookeeper: 默认情况下，transactional topology会把状态信息保存在主zookeeper里面（协调集群的那个)。你可以通过这两个配置来指定其它的zookeeper：”

transactional.zookeeper.servers

” 和 “

transactional.zookeeper.port

“。

同时活跃的batch数量：你必须设置同时处理的batch数量。你可以通过”

topology.max.spout.pending

” 来指定， 如果你不指定，默认是1。

实现

Transactional Topologies的实现是非常优雅的。管理提交协议，检测失败并且串行提交看起来很复杂，但是使用storm的原语来进行抽象是非常简单的。

transactional topology里面的spout是一个子topology,它由一个spout和一个bolt组成。

spout是协调者，它只包含一个task。

bolt是发射者

bolt以all grouping的方式订阅协调者的输出。

元数据的序列化用的是kryo。

协调者使用acking框架来决定什么时候一个batch被成功执行完成，然后去决定一个batch什么时候被成功提交。

状态信息被以

RotatingTransactionalState

的形式保存在zookeeper里面了。

commiting bolts以all grouping的方式订阅协调者的commit流。

CoordinatedBolt被用来检测一个bolt是否收到了一个特定batch的所有tuple。

这一点上面跟DRPC里面是一样的。

对于commiting bolt来说， 他会一直等待， 知道从coordinator的commit流里面接收到一个tuple之后，它才会调用

finishBatch

方法。

所以在没有从coordinator的commit流接收到一个tuple之前，committing bolt不可能调用

finishBolt

方法。