Twitter Storm: Transactional Topolgoy简介

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

transactional-batches

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

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/1dd5b802a91c5089205ed5d50f36d2ef.html