Storm 事务性拓扑.pdfVIP

Storm 事务性拓扑 事务性拓扑 正如书中之前所提到的，使⽤ Storm 编程，可以通过调⽤ ack 和 fail ⽅法来确保⼀条 消息的处 成功或失败。不过当元组被重发时，会发⽣什么呢？你又该如何砍不会重 复计算？ Storm0 .7 .0 实现了⼀个新特性——事务性拓扑，这⼀特性使消息在语义上确保你可以 安全的⽅式重发消息，并保证它们只会被处 ⼀次。在不⽀持事务性拓扑的情况下， 你⽆法在准确性，可扩展性，以空错性上得到保证的前提下完成计算。 NOTE:事务性拓扑是⼀个构建于标准 Storm spout 和 bolt 之上的抽象概念。 设计 在事务性拓扑中，Storm 以并⾏和顺序处 混合的⽅式处 元组。spout 并⾏分批创建 供 bolt 处 的元组 （译者注：下⽂将这种分批创建、分批处 的元组称做批次）。其 中⼀些 bolt 作为提交者以严格有序的⽅式提交处 过的批次。这意味着如果你有每批 五个元组的两个批次，将有两个元组被 bolt 并⾏处 ，但是直到提交者成功提交了第 ⼀个元组之后，才会提交第⼆个元组。 NOTE ： 使⽤事务性拓扑时，数据源要能够重发批次，有时候甚⾄要重复多次。因此 确认你的数据源——你连接到的那个 spout ——具备这个能⼒。 这个过程可以被描述 为两个阶段： 处 阶段 纯并⾏阶段，许多批次同时处 。 提交阶段 严格有序阶段， 直到批次⼀成功提交之后，才会提交批次⼆。 这两个阶段合起来称为⼀个 Storm 事 务。 NOTE ： Storm 使⽤ zookeeper 储存事务元数据，默认情况下就是拓扑使⽤的那 个 zookeeper 。你可以修改以下两个配置参数键指定其它的 zookeeper —— transactional .zookeeper .servers 和transactional .zookeeper .port 。 事务实践 下⾯我们要创建⼀个 Twitter 分析⼯具来了解事务的⼯作⽅式。我们从⼀个 Redis 数据 库读取tweets ，通过⼏个 bolt 处 它们，最后把结果保存在另⼀个 Redis 数据库的列 表中。处 结果就是所有话题和它们的在 tweets 中出现的次数列表，所有⽤户和他们 在 tweets 中出现的次数列表，还有⼀个包含发起话题和频率的⽤户列表。 这个⼯具的 拓扑图。 图 拓扑概览 正如你看到的，TweetsTransactionalSpout 会连接你的 tweet 数据库并向拓扑分发批 次。UserSplitterBolt 和 HashTagSplitterBolt 两个 bolt ，从 spout 接收元组。 UserSplitterBolt 解析 tweets 并查找⽤户——以 开头的单词——然后把这些单词分发 到名为 users 的⾃定义数据流组。HashtagSplitterBolt 从 tweet 查找 # 开头的单词，并 把它们分发到名为 hashtags 的⾃定义数据流组。第三个 bolt ，UserHashtagJoinBolt ， 接收前⾯提到的两个数据流组，并计算具名⽤户的⼀条 tweet 内的话题数量。为了计 数并分发计算结果，这是个 BaseBatchBolt （稍后有更多介绍）。 最后⼀个 bolt ——RedisCommitterBolt —— 接收以上三个 bolt 的数据流组。它为每样 东西计数，并在对⼀个批次完成处 时，把所有结果保存到 redis 。这是⼀种特殊的 bolt ，叫做提交者，在本章后⾯做更多讲解。 ⽤ TransactionalTopologyBuilder 构建拓扑，代码如下： 01 TransactionalTopologyBuilder builder= 02 new TransactionalTopologyBuilder(test, spout, new TweetsT 03 0 builder.setBolt(users-splitter, new UserSplitterBolt(), ).shuf 05 buildeer.setBolt(hashtag-splitter, new HashtagSplitterBolt(), 06 07 builder.setBolt(users-hashtag-manager, new UserHashtagJoinBolt( 08 .fieldsGrouping(users