88bf必发 2

Twitter Storm: Transactional Topolgoy简介

概述

Storm通过保证每个tuple至少被处理一次来提供可靠的数据处理。关于这一点最常被问到的问题就是“既然tuple可能会被重写发射(replay),
那么我们怎么在storm上面做统计个数之类的事情呢?storm有可能会重复计数吧?”

Storm 0.7.0引入了Transactional Topology,
它可以保证每个tuple”被且仅被处理一次”,
这样你就可以实现一种非常准确,非常可扩展,并且高度容错方式来实现计数类应用。

88bf必发,跟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内部是可以并行的。下面这个是设计图:

88bf必发 1

所以如果你每个batch处理1000个tuple的话,
那么你的应用将会少1000倍的数据库调用。同时它利用了storm的并行计算能力(每个batch内部可以并行)

虽然这个设计比第一个设计好多了,
它仍然不是一个完美的方案。topology里面的worker会花费大量的时间等待计算的其它部分完成。
比如看下面的这个计算。

88bf必发 2

在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我们会在后面介绍。

推荐阅读:

Twitter Storm安装配置(集群)笔记

安装Twitter Storm集群

Twitter Storm安装配置(单机版)笔记

Storm 实战及实例讲解一

  • 1
  • 2
  • 下一页

http://www.bkjia.com/Linux/766386.htmlwww.bkjia.comtruehttp://www.bkjia.com/Linux/766386.htmlTechArticle概述
Storm通过保证每个tuple至少被处理一次来提供可靠的数据处理。关于这一点最常被问到的问题就是既然tuple可能会被重写发射(replay),
那…