引入

我们通过以下两篇文章，深入探索了S4是如何抽象流式计算模型，如何设计架构和系统，存在那些局限：

• 论文详解
• 论文总结

Yahoo推出的S4 并没有在历史舞台上站稳脚跟，在S4的论文发表的同一年，我们今天的主角，也就是Storm走上了历史舞台。在接下来的几年里，Storm一度成为业界进行实时数据处理的标准解决方案。

而且就像最近风靡的DeepSeek类似，Storm并不是来自哪一个业界的大公司。而是来自一个只有3个人的创业公司BackType，它的主要作者南森·马茨（Nathan Marz）那个时候也才仅仅21岁。而没过多久之后，BackType就被当时如日中天的Twitter收购了。可以说 Storm证明了即使是到了今天，天才工程师仍然能够凭借一己之力，对整个行业产生重要的影响。

不过作为一个开源项目，Storm一开始并没有以论文的形式发表。直到2014年，Twitter才发表了《Storm @Twitter》这样一篇更加偏重于Storm如何在Twitter内部使用的论文。而且这篇论文中的作者中，也没有Storm最初的作者南森·马茨。虽然论文里没有南森的名字，但是Storm最大的功劳也仍然来自于他这个最初的作者。

今天我们还是先来看看Storm这篇论文：

Storm @Twitter

摘要

本文介绍了 Twitter 对 Storm 的使用情况。Storm 是一个实时、容错的分布式流数据处理系统。目前，Twitter 正大规模实时运用 Storm 来运行各种关键计算任务。本文阐述了 Storm 的架构，以及它实现分布式扩展和容错的方法。同时，本文还说明了查询（即拓扑结构）在 Storm 中是如何执行的，并分享了一些基于 Twitter 运行 Storm 的实际操作案例。我们还给出了实证评估结果，展示了 Storm 在应对机器故障时的弹性。Storm 在 Twitter 仍处于积极开发阶段，我们也提出了一些未来可能的研究方向。

1. 引言

许多现代数据处理环境需要对流数据进行实时的复杂计算处理。在 Twitter 尤其如此，每次与用户的交互都需要做出许多复杂决策，而这些决策往往基于刚刚生成的数据。

Storm 是 Twitter 的一个实时分布式流数据处理引擎，它支撑着对提供 Twitter 服务至关重要的实时流数据管理任务。Storm 旨在具备以下特性：

1. 可扩展性：运维团队需要能够轻松地在 Storm 集群中添加或移除节点，且不会干扰通过 Storm 拓扑结构（即常驻查询）的现有数据流。
2. 弹性：容错性对 Storm 至关重要，因为它通常部署在大型集群上，硬件组件可能会发生故障。Storm 集群必须在对性能影响最小的情况下，继续处理现有的拓扑结构。
3. 可扩展性：Storm 拓扑结构可能会调用任意外部函数（例如，从 MySQL 服务中查询社交图谱），因此需要一个支持可扩展性的框架。
4. 高效性：由于 Storm 用于实时应用程序，它必须具备良好的性能特征。Storm 采用了多种技术，包括将所有存储和计算数据结构都保留在内存中。
5. 易于管理：由于 Storm 是 Twitter 用户交互的核心，如果 Storm 出现（故障或性能）问题，终端用户会立即察觉。运维团队需要早期预警工具，并且必须能够在问题出现时迅速找出问题根源。因此，易于使用的管理工具并非 “锦上添花” 的功能，而是一项关键需求。

我们注意到，Storm 的起源可追溯到丰富的流数据处理相关工作（例如 [1, 2, 3, 4]），并且在很大程度上借鉴了该领域的思路。然而，其关键区别在于将上述所有方面整合到一个系统中。我们还注意到，虽然 Storm 是早期的流处理系统之一，但也有其他值得关注的系统，包括 S4 [5]，以及更新的系统，如 MillWheel [6]、Samza [7]、Spark Streaming [8] 和 Photon [19]。流数据处理技术也已作为传统数据库产品流水线的一部分被整合（例如 [9, 10, 11]）。

许多早期的流数据处理系统在引入各种概念（例如可扩展性、伸缩性、弹性）方面起到了引领作用，我们并不声称这些概念是在 Storm 中首创的，而是认识到流处理正迅速成为企业综合数据处理解决方案的关键组成部分，而 Storm 是目前使用的早期开源且流行的流处理系统之一。

Storm 最初由 BackType 的 Nathan Marz 创建，BackType 于 2011 年被 Twitter 收购。在 Twitter，Storm 在多个方面得到了改进，包括扩展到大量节点，以及减少 Storm 对 Zookeeper 的依赖。Twitter 在 2012 年将 Storm 开源，随后被其他各种组织采用。目前有超过 60 家公司正在使用或试验 Storm。一些当前使用 Storm 的组织包括：雅虎（Yahoo!）、高朋（Groupon）、天气频道（The Weather Channel）、阿里巴巴、百度和 Rocket Fuel。

我们注意到，如今正在使用的流处理系统（包括 Storm）仍在不断发展，并且将继续从流处理领域丰富的研究成果中汲取养分；例如，许多这些 “现代” 系统并不支持声明式查询语言，如 [12] 中提出的那种。因此，流处理领域是一个活跃且快速发展的研究和前沿开发空间。

我们还注意到，有许多关于 Storm 的在线教程 [20, 21]，它们仍然是 Storm 用户社区的宝贵资源。

向 YARN [23] 的转变也引发了将 Storm 与 Hadoop 生态系统集成的兴趣，现在也有许多与在 Hadoop 中使用 Storm 相关的资源（例如 [21, 22]）。

本文的其余部分组织如下：第 2 节描述 Storm 的数据模型和架构。第 3 节介绍 Twitter 如何使用 Storm，包含一些实证结果，并讨论我们在 Twitter 运行 Storm 时遇到的一些操作方面的问题。最后，第 4 节给出我们的结论，并指出一些未来的工作方向。

2. 数据模型与执行架构

Storm 基本的数据处理架构由流经拓扑结构的元组流构成。拓扑结构是一种有向图，其中顶点代表计算过程，边则表示各计算组件之间的数据流。顶点又进一步分为两个不相交的集合 ——Spouts和Bolts。Spout是拓扑结构的元组来源。典型的Spout从诸如 Kafka [13] 或 Kestrel [14] 等队列中提取数据。另一方面，Bolt负责处理传入的元组，并将其传递给下游的下一组Bolt。需要注意的是，Storm 拓扑结构可以存在循环。从数据库系统的角度来看，可以将拓扑结构视为运算符的有向图。

图 1 展示了一个简单的拓扑结构，它用于统计推文中出现的单词，并每 5 分钟生成一次这些单词的计数结果。此拓扑结构有一个Spout（TweetSpout）和两个Bolt（ParseTweetBolt 和 WordCountBolt）。TweetSpout 可能从 Twitter 的 Firehose API 提取元组，并持续将新推文注入拓扑结构。ParseTweetBolt 将推文拆分为单词，并为每个单词发出二元元组（单词，计数）。WordCountBolt 接收这些二元元组，汇总每个单词的计数，并每 5 分钟输出一次计数结果。在输出单词计数后，它会清空内部计数器。

图 1：推文单词计数拓扑结构

2.1 Storm 概述

Storm 运行在分布式集群上，在 Twitter 通常还基于诸如 Mesos [15] 这样的另一层抽象来运行。客户端将拓扑结构提交给一个主节点，即 Nimbus。Nimbus 负责分发和协调拓扑结构的执行。实际工作由工作节点完成。每个工作节点运行一个或多个工作进程。在任何时间点，一台机器可能运行多个工作进程，但每个工作进程只对应一个拓扑结构。需注意，同一台机器上的多个工作进程可能执行同一拓扑结构的不同部分。Storm 的高层架构如图 2 所示。

图 2：Storm 的高层架构

每个工作进程运行一个 JVM，在其中运行一个或多个执行器。执行器由一个或多个任务组成。Bolt或Spout的实际工作在任务中完成。

因此，任务提供了Bolt内 / Spout内的并行性，而执行器提供了拓扑结构内的并行性。工作进程作为主机上运行 Storm 拓扑结构的容器。需要注意的是，每个Spout或Bolt都关联着一组在集群中多台机器的一组执行器中运行的任务。数据从生产者Spout / Bolt被打乱分发到消费者Bolt（生产者和消费者都可能有多个任务）。这种打乱分发类似于并行数据库中的交换操作符 [16]。

Storm 支持以下几种分区策略：

1. 随机分组（Shuffle grouping）：随机对元组进行分区。
2. 字段分组（Fields grouping）：根据元组属性 / 字段的一个子集进行哈希。
3. 全部分组（All grouping）：将整个流复制到所有消费者任务。
4. 全局分组（Global grouping）：将整个流发送到单个Bolt。
5. 本地分组（Local grouping）：将元组发送到同一执行器中的消费者Bolt。

分区策略是可扩展的，拓扑结构可以定义并使用自己的分区策略。

每个工作节点运行一个与 Nimbus 通信的 Supervisor。集群状态保存在 Zookeeper [17] 中，Nimbus 负责在工作节点上调度拓扑结构，并监控流经拓扑结构的元组的处理进度。关于 Nimbus 的更多细节将在 2.2.1 节中介绍。

大致来说，从数据库系统的角度，拓扑结构可被视为一个逻辑查询计划。作为拓扑结构的一部分，程序员需指定每个Spout和Bolt必须生成的实例数量。Storm 会创建这些实例，并建立数据流的互连。例如，推文单词计数拓扑结构的物理执行计划如图 3 所示。

图 3：推文单词计数拓扑结构的物理执行情况

我们注意到，目前程序员必须为每个Spout和Bolt指定实例数量。未来工作的一部分是根据某些更高级别的目标（如目标性能指标）自动选择并动态更改这个数量。

2.2 Storm 内部机制

在本节中，我们将描述 Storm 的关键组件（如图 2 所示），以及这些组件如何相互交互。

2.2.1 Nimbus 和 Zookeeper

Nimbus 在 Storm 中扮演着与 Hadoop 中 “JobTracker” 类似的角色，是用户与 Storm 系统的交互点。

Nimbus 是一个 Apache Thrift 服务，而 Storm 拓扑结构定义是 Thrift 对象。为了向 Storm 集群（即向 Nimbus）提交作业，用户将拓扑结构描述为一个 Thrift 对象，并将该对象发送给 Nimbus。通过这种设计，任何编程语言都可用于创建 Storm 拓扑结构。

在 Twitter，一种常用的生成 Storm 拓扑结构的方法是使用 Summingbird [18]。Summingbird 是一种通用的流处理抽象，它提供了一个独立的逻辑规划器，可以映射到各种流处理和批处理系统。Summingbird 为程序员提供了一种强大的、符合 Scala 习惯用法的方式来表达他们的计算和约束条件。由于 Summingbird 理解数据处理函数之间的类型和关系（如结合律），它可以执行多种优化。用 Summingbird 表达的查询可以自动转换为 Storm 拓扑结构。Summingbird 的一个有趣之处在于它还可以生成在 Hadoop 上运行的 MapReduce 作业。Twitter 的一个常见用例是使用 Storm 拓扑结构实时计算近似答案，随后再与 MapReduce 执行得到的准确结果进行核对。

作为提交拓扑结构的一部分，用户还会将用户代码以 JAR 文件的形式上传到 Nimbus。Nimbus 结合使用本地磁盘和 Zookeeper 来存储关于拓扑结构的状态。目前，用户代码存储在 Nimbus 机器的本地磁盘上，而拓扑结构的 Thrift 对象存储在 Zookeeper 中。

Supervisor 通过定期心跳协议与 Nimbus 联系，报告它们当前正在运行的拓扑结构，以及是否有可用空位来运行更多拓扑结构。Nimbus 跟踪需要分配的拓扑结构，并在待处理的拓扑结构和 Supervisor 之间进行匹配。

Nimbus 和 Supervisor 之间的所有协调工作都通过 Zookeeper 完成。此外，Nimbus 和 Supervisor 守护进程具有快速失败和无状态的特点，它们所有的状态都保存在 Zookeeper 或本地磁盘上。这种设计是 Storm 具备弹性的关键。如果 Nimbus 服务发生故障，工作节点仍能继续推进工作。另外，如果工作节点出现故障，Supervisor 会重新启动它们。

然而，如果 Nimbus 宕机，用户就无法提交新的拓扑结构。而且，如果正在运行的拓扑结构遇到机器故障，在 Nimbus 恢复之前，这些拓扑结构无法重新分配到不同的机器上。未来一个有趣的工作方向是解决这些限制，使 Storm 在面对故障时更具弹性和响应能力。

2.2.2 Supervisor

Supervisor 运行在每个 Storm 节点上。它从 Nimbus 接收任务分配，并根据分配启动工作进程。它还监控工作进程的健康状况，并在必要时重新启动它们。Supervisor 的高层架构如图 4 所示。

图 4：Supervisor 架构

如图所示，Supervisor 启动三个线程。主线程读取 Storm 配置，初始化 Supervisor 的全局映射，在文件系统中创建持久化的本地状态，并调度周期性的定时器事件。有三种类型的事件，分别是：

1. 心跳事件：计划每 15 秒运行一次，在主线程的上下文中运行。它向 Nimbus 报告 Supervisor 处于存活状态。
2. 同步 Supervisor 事件：每 10 秒在事件管理器线程中执行一次。该线程负责管理现有任务分配的变更。如果变更包括添加新的拓扑结构，它会下载必要的 JAR 文件和库，并立即调度一个同步进程事件。
3. 同步进程事件：在进程事件管理器线程的上下文中每 3 秒运行一次。该线程负责管理在与 Supervisor 同一节点上运行拓扑结构片段的工作进程。它从本地状态读取工作进程的心跳信息，并将这些工作进程分类为有效、超时、未启动或禁止运行。“超时” 的工作进程意味着该工作进程在指定时间内未提供心跳，现在被认为已死亡。“未启动” 的工作进程表示它尚未启动，因为它属于新提交的拓扑结构，或者现有拓扑结构的工作进程正在被转移到该 Supervisor。最后，“禁止运行” 的工作进程意味着该工作进程不应运行，原因可能是其所属的拓扑结构已被终止，或者该拓扑结构的工作进程已被转移到另一个节点。

2.2.3 工作进程和执行器

回想一下，每个工作进程在一个 JVM 内运行多个执行器。这些执行器是工作进程内的线程。每个执行器可以运行多个任务。一个任务是Spout或Bolt的一个实例。由于当前这种分配是静态的，所以一个任务严格绑定到一个执行器。未来一个有趣的工作方向是允许动态重新分配，以针对某些更高级的目标（如负载均衡或满足服务水平目标 SLO）进行优化。

为了路由传入和传出的元组，每个工作进程有两个专用线程 —— 一个工作进程接收线程和一个工作进程发送线程。工作进程接收线程监听一个 TCP/IP 端口，作为所有传入元组的解复用点。它检查元组的目标任务标识符，并相应地将传入的元组排队到与其执行器相关联的适当输入队列中。

每个执行器由两个线程组成，即用户逻辑线程和执行器发送线程。用户逻辑线程从输入队列中获取传入的元组，检查目标任务标识符，然后针对该元组运行实际任务（Spout或Bolt实例），并生成输出元组。这些传出的元组随后被放入与该执行器相关联的输出队列中。接下来，执行器发送线程从输出队列中取出这些元组，并将它们放入一个全局传输队列中。全局传输队列包含来自多个执行器的所有传出元组。

工作进程发送线程检查全局传输队列中的每个元组，并根据其任务目标标识符，将其发送到下游的下一个工作进程。对于发往同一工作进程上不同任务的传出元组，执行器发送线程会将该元组直接写入目标任务的输入队列中。工作进程内部的消息流如图 5 所示。

图 5. 工作进程内部的消息流

2.3 处理语义

Storm 的关键特性之一是它能够对所处理的数据提供保障。它提供两种语义保障 ——“至少一次” 和 “至多一次” 语义。

“至少一次” 语义保证输入到拓扑结构的每个元组至少会被处理一次。

采用 “至多一次” 语义时，每个元组要么被处理一次，要么在出现故障时被丢弃。

为了提供 “至少一次” 语义，拓扑结构会添加一个 “确认器（acker）” Bolt，它会为Spout发出的每个元组跟踪元组的有向无环图。例如，增强后的推文单词计数拓扑结构如图 6 所示。

图 6. 增强型单词计数拓扑结构

Storm 会为流经系统的每个新元组附加一个随机生成的 64 位 “消息 ID”。这个 ID 在Spout从某个输入源首次提取元组时附加到该元组上。

在处理一个元组时可能会产生新的元组；例如，一个包含完整推文的元组被一个Bolt拆分成一组热门话题，为输入元组的每个话题生成一个元组。这些新元组会被分配一个新的随机 64 位 ID，并且元组 ID 列表也会保留在与输出元组相关联的溯源树中。当一个元组最终离开拓扑结构时，会使用一种回流机制来确认对该输出元组有贡献的任务。这种回流机制最终会到达最初启动元组处理的Spout，此时Spout可以将该元组标记为已处理完成。

这种机制的一种简单实现方式需要跟踪每个元组的来源。这意味着对于每个元组，其源元组 ID 必须保留到该元组处理结束。这样的实现可能会导致大量的内存使用（用于溯源跟踪），特别是对于复杂的拓扑结构。

为了避免这个问题，Storm 使用了一种基于按位异或（XOR）的新颖实现方式。如前所述，当一个元组进入Spout时，它会被赋予一个 64 位的消息 ID。Spout处理完这个元组后，可能会发出一个或多个元组。这些发出的元组会被分配新的消息 ID。这些消息 ID 会与原始元组消息 ID 以及一个超时参数进行异或运算，并发送给确认器Bolt。因此，确认器Bolt会跟踪所有的元组。当一个元组的处理完成或被确认时，其消息 ID 以及原始元组消息 ID 会被发送给确认器Bolt。确认器Bolt找到原始元组及其异或校验和。这个异或校验和会再次与已确认的元组 ID 进行异或运算。当异或校验和变为零时，确认器Bolt会向接收该元组的Spout发送最终确认消息。此时Spout就知道这个元组已被完全处理。

由于故障，有可能某些异或校验和永远不会变为零。为了处理这种情况，Spout最初会分配上述的超时参数。确认器Bolt会跟踪这个超时参数，如果在超时之前异或校验和没有变为零，该元组将被视为处理失败。

请注意，Storm 中的通信是通过 TCP/IP 进行的，它具有可靠的消息传递功能，所以没有元组会被传递多次。因此，即使异或运算不是幂等的，这种异或机制仍然有效。

对于 “至少一次” 语义，数据源必须 “持有” 一个元组。对于该元组，如果Spout收到肯定的确认消息，那么它可以告诉数据源移除该元组。如果在指定时间内没有收到确认或失败消息，那么数据源将取消对该元组的 “持有”，并在后续迭代中重新发送它。Kestrel 队列提供了这样的行为。另一方面，对于 Kafka 队列，每个Spout实例会将已处理的元组（或消息偏移量）在 Zookeeper 中进行检查点记录。当一个Spout实例发生故障并重新启动时，它会从 Zookeeper 中记录的最后一个 “检查点” 状态开始处理元组。

“至多一次” 语义意味着进入系统的元组要么至少被处理一次，要么根本不被处理。当拓扑结构的确认机制被禁用时，Storm 就实现了 “至多一次” 语义。当确认功能被禁用时，无法保证一个元组在拓扑结构的每个阶段都能成功处理或失败，处理会继续向前推进。

3. Twitter 中 Storm 的应用

在本节中，我们将介绍 Twitter 如何使用 Storm。我们还会给出三个应对运营和部署问题的实例。最后，我们会展示实证评估的结果。

3.1 运营概述

目前，Storm 在 Twitter 的数百台服务器（分布在多个数据中心）上运行。这些集群上运行着数百个拓扑结构，其中一些拓扑结构在数百个节点上运行。每天有数千 TB 的数据流经 Storm 集群，生成数十亿个输出元组。

Twitter 内部的多个团队都在使用 Storm 拓扑结构，包括营收、用户服务、搜索和内容发现团队。这些拓扑结构用于执行简单任务，如对 Twitter 各种流的内容进行过滤和聚合（例如计算数量），也用于更复杂的任务，如对流数据运行简单的机器学习算法（例如聚类）。

这些拓扑结构的复杂程度各不相同，大量拓扑结构的阶段数少于三个（即拓扑图的深度小于三），但有一个拓扑结构有八个阶段。目前，拓扑结构在各自独立的机器上运行，我们希望未来能够消除这一限制。

Storm 具有故障恢复能力，即使 Nimbus 宕机，它仍能继续工作（工作节点继续推进任务）。此外，如果我们需要关闭一台机器进行维护，也不会影响拓扑结构的运行。在过去 6 个月中，我们处理单个元组的 99% 延迟（即第 99 个百分位数的响应时间延迟）接近 1 毫秒，集群可用性达到 99.9%。

3.2 Storm 可视化操作

在实际使用 Storm 时，一个关键部分是对 Storm 操作进行可视化。通过内部开发的丰富可视化工具，Storm 的日志会持续显示，部分内容如图 7 所示。为了收集日志，每个拓扑结构都添加了一个指标Bolt。在每个Spout或Bolt处收集的所有指标都会发送到这个Bolt。这个Bolt再将指标写入 Scribe，Scribe 会将数据路由到一个持久化的键值存储中。对于每个拓扑结构，都会使用这些数据创建一个仪表盘，以可视化该拓扑结构的运行情况。

图 7：Storm 可视化界面

这种丰富的可视化对于帮助识别和解决引发警报的问题至关重要。

这些指标大致可分为系统指标和拓扑指标。系统指标展示平均 CPU 利用率、网络利用率、每分钟垃圾回收次数、每分钟垃圾回收耗时以及堆内存使用情况。拓扑指标针对每个Bolt和每个Spout进行报告。Spout指标包括每分钟发出的元组数量、元组确认数量、每分钟失败消息数量以及在拓扑结构中处理整个元组的延迟。Bolt指标包括执行的元组数量、每分钟确认数量、平均元组处理延迟以及确认特定元组的平均延迟。

3.3 运营实例

在本节中，我们将介绍三个与 Storm 相关的运营场景 / 实例。

3.3.1 过载的 Zookeeper

如前所述，Storm 使用 Zookeeper 来跟踪状态信息。一个反复出现的问题是如何在 Storm 中设置和使用 Zookeeper，特别是当 Zookeeper 在 Twitter 还被用于其他系统时。我们对如何在 Storm 中使用和配置 Zookeeper 进行了各种考量。

我们尝试的第一种配置是使用 Twitter 现有的 Zookeeper 集群，该集群同时也被 Twitter 内部的许多其他系统使用。很快，我们就超出了这个 Zookeeper 集群所能支持的客户端数量，这反过来影响了共享该 Zookeeper 集群的其他系统的正常运行时间。

我们的 Storm 集群的第二种配置与第一种相同，只是为 Zookeeper 集群配备了专用硬件。虽然这显著提高了我们在 Storm 集群中可以运行的工作进程和拓扑结构的数量，但每个集群很快就达到了大约 300 个工作进程的限制。如果超过这个数量，我们就会看到工作进程被调度器终止并重新启动。这是因为每个工作进程都有一个对应的 Zookeeper 节点（zknode），必须每 15 秒写入一次，否则 Nimbus 会认为该工作进程已停止运行，并将其重新调度到一台新机器上。

在我们的 Storm 集群的第三种配置中，我们再次更改了 Zookeeper 的硬件和配置：我们使用数据库级硬件，配备 6 个 500GB SATA 磁盘，以 RAID1 + 0 模式存储 Zookeeper 事务日志，并使用 1 个 500GB 磁盘（无 RAID）存储快照。Zookeeper 文档强烈建议将事务日志和快照存储在不同的磁盘上，如果不遵循此建议，Zookeeper 集群可能会变得不稳定。这种第三种配置可扩展到大约 1200 个工作进程。如果超过这个数量，我们又会开始看到工作进程被终止并重新启动（与第二种配置情况相同）。

然后，我们通过解析其中一个 Zookeeper 节点的 tcpdump 日志来分析 Zookeeper 的写入流量。我们发现，每秒写入 Zookeeper 仲裁的流量中，有 67% 不是由 Storm 核心运行时执行的，而是由名为 KafkaSpout 的 Storm 库执行的。KafkaSpout 使用 Zookeeper 来存储关于从 Kafka 队列中已消费多少数据的少量状态信息。KafkaSpout 的默认配置是每个分区、每个 Storm 拓扑每 2 秒向 Zookeeper 写入一次。当时我们 Kafka 主题的分区数在 15 到 150 之间，集群中大约有 20 个拓扑结构。（Kafka 是一个通用的发布 - 订阅系统，有主题的概念。生产者可以写入某个主题，消费者可以消费感兴趣主题的数据。因此，就本讨论而言，一个主题类似于一个队列。）

在我们的 tcpdump 样本中，我们看到在 60 秒的时间窗口内，有 19956 次写入到由 KafkaSpout 代码所拥有的 zknode。此外，我们发现写入 Zookeeper 的流量中有 33% 是由 Storm 代码执行的。在这部分流量中，96% 来自 Storm 核心，默认情况下，工作进程每 3 秒向 Zookeeper 写入心跳信息。

由于我们认为在当前硬件条件下，Zookeeper 集群的写入性能已达到极限，因此我们决定大幅减少对 Zookeeper 的写入次数。因此，对于我们在 Twitter 的 Storm 集群的第四种也是当前的生产配置，我们修改了 KafkaSpout 代码，将其状态写入键值存储。我们还修改了 Storm 核心，将其心跳状态写入专门为存储 Storm 心跳数据而设计的自定义存储系统（称为 “心跳守护进程”）。心跳守护进程集群旨在牺牲读取一致性，以换取高可用性和高写入性能。它们可以水平扩展，以匹配 Storm 核心中运行的工作进程所施加的负载，现在这些工作进程将心跳信息写入心跳守护进程集群。

3.3.2 Storm 的开销

曾有一段时间，有人担心从 Kafka 队列消费数据（即在Spout中使用 Kafka）的 Storm 拓扑结构相对于直接使用 Kafka 客户端的手写 Java 代码性能较差。这种担忧源于一个从 Kafka 队列消费数据的 Storm 拓扑结构需要 10 台机器才能成功处理以 300K 条消息 / 秒的速率到达队列的输入数据。

如果使用的机器少于 10 台，那么该拓扑结构的消费速率将低于其消费的队列的生产速率。此时，该拓扑结构将不再是实时的。对于这个拓扑结构，实时的概念是指输入元组中最初表示的事件到实际对该元组进行计算的时间间隔应小于 5 秒。在这种情况下使用的机器配置为 2 个 Intel E5645@2.4Ghz CPU，12 个物理核心并支持超线程，24 个硬件线程，24GB 内存和一个 500GB SATA 磁盘。

在我们的第一个实验中，我们编写了一个不使用 Storm 或任何 Storm 流计算框架的 Java 程序。这个程序使用 Kafka Java 客户端从与 Storm 拓扑相同的 Kafka 集群和主题消费数据，仅使用 “for 循环” 尽可能快地读取消息，然后反序列化这些消息。反序列化后，如果该程序中不进行其他处理，该项将被垃圾回收。

由于这个程序不使用 Storm，它不支持可靠的消息处理、机器故障恢复，也不对流进行任何重新分区。这个程序能够以 300K 条消息 / 秒的速率消费输入数据，并且在单台机器上实时处理数据，根据 Unix 命令行工具 top 的报告，其 CPU 利用率平均约为 700%（有 12 个物理核心，CPU 利用率上限为 1200%）。

第二个实验是编写一个具有与 Java 程序类似逻辑 / 功能的 Storm 拓扑结构。我们构建了一个简单的 Storm 拓扑结构，与 Java 程序类似，它所做的只是反序列化输入数据。在这个实验中，我们还禁用了消息可靠性支持。使用 Storm 的隔离调度器，执行这个 Storm 拓扑结构的所有 JVM 进程都位于同一台机器上，模仿与 Java 程序相同的设置。这个拓扑结构有 10 个进程，每个进程有 38 个线程。这个拓扑结构也能够以 300K 条消息 / 秒的速率消费数据，并且在单台机器（即与上述配置相同）上实时处理数据，top 报告的平均 CPU 利用率约为 660%。这个 CPU 利用率略低于第一个不使用 Storm 的实验。

对于第三个实验，我们采用与第二个实验相同的拓扑结构，但现在启用了消息可靠性。这个拓扑结构至少需要 3 台机器才能以 300K 条消息 / 秒的速率消费输入数据。此外，它配置了 30 个 JVM 进程（每台机器 10 个），每个进程 5 个线程。top 报告的平均 CPU 利用率为 924%。这些实验大致表明，启用消息可靠性相对于反序列化消息所涉及的 CPU 成本，Storm 的 CPU 成本约为其 3 倍。

这些实验缓解了关于 Storm 与执行相同计算的普通 Java 代码相比会增加显著开销的担忧，因为当两个应用程序提供相同的消息可靠性保证时，它们的 CPU 利用率大致相同。

这些实验确实揭示了 Storm 中与消息可靠性机制相关的 CPU 成本不可忽视，与消息反序列化成本处于同一数量级。

我们无法在不使用 Storm 的 Java 程序中重现最初需要 10 台机器的 Storm 拓扑结构，因为这将涉及大量工作，因为这个拓扑结构有 3 层Bolt和Spout，并对流进行了两次重新分区。在不使用 Storm 的情况下重新实现所有这些功能需要太多时间。所需的额外机器可以用这个拓扑结构内的业务逻辑开销，和 / 或由于流需要重新分区而在网络上发送元组时产生的反序列化和序列化成本来解释。

3.3.3 最大Spout挂起参数调优

Storm 拓扑结构有一个 “最大Spout挂起（max spout pending）” 参数。拓扑结构的最大Spout挂起值可以通过拓扑配置 yaml 文件中的 “topology.max.spout.pending” 设置进行配置。这个值限制了在任何时间点，Storm 拓扑结构中有多少个元组处于未确认或未失败的 “飞行” 状态，即尚未得到确认或失败的元组数量。设置这个参数的原因是 Storm 使用 ZeroMQ [25] 将元组从一个任务分发到另一个任务。如果 ZeroMQ 的消费者端无法跟上元组的速率，那么 ZeroMQ 队列就会开始堆积。最终，元组在Spout处超时并被重新发送到拓扑结构中，从而给队列带来更大压力。为了避免这种严重的故障情况，Storm 允许用户对拓扑结构中处于 “飞行” 状态的元组数量设置限制。这个限制是基于每个Spout任务生效，而不是在拓扑级别生效。对于Spout不可靠的情况，即它们在元组中不发出消息 ID，这个值不起作用。

Storm 用户一直面临的问题之一是为这个最大Spout挂起参数确定合适的值。值过小很容易使拓扑结构 “饥饿”，而值过大则可能使拓扑结构因大量元组而过载，甚至导致故障和重发。用户必须通过多次部署拓扑结构并尝试不同的最大Spout挂起值，才能找到最适合他们的值。

为了缓解这个问题，在 Twitter 我们实现了一种针对最大Spout挂起值的自动调优算法，该算法会定期调整该值，以实现拓扑结构的最大吞吐量。在这种情况下，吞吐量的衡量标准是Spout能够推进的进度，而不一定是我们可以向拓扑结构中推送或处理的更多元组数量。该算法适用于 Kafka 和 Kestrel Spout，这些Spout已进行增强，以跟踪和报告它们随时间取得的进展。

该算法的工作原理如下：

a) Spout任务跟踪一个名为 “进度” 的指标。这个指标表示该Spout任务已成功处理的数据量。对于 Kafka Spout，这个指标通过查看 Kafka 日志中被视为 “已提交” 的偏移量来衡量，即在该偏移量之前的所有数据都已成功处理且不会再被重发。对于 Kestrel Spout，这个指标通过计算从 Storm 拓扑结构接收到的确认数量来衡量。请注意，我们不能将接收到的确认数量用作 Kafka Spout的进度指标，因为在其实现中，已确认但尚未提交的元组仍可能被重发。

b) 我们有一个可插拔的最大Spout参数 “调谐器” 类实现，用于自动调整最大Spout挂起值。默认实现支持的两个 API 是：

• void autoTune (long deltaProgress)，它使用上次调用 autoTune () 之间取得的进度来调整最大Spout挂起值。
• long get ()，它返回调整后的最大Spout挂起值。

c) 每隔 “t” 秒（在我们的案例中，t 的默认值为 120 秒），Spout会调用 autoTune，并向其提供Spout在过去 t 秒内取得的进度。

d) 调谐器类记录它采取的最后一个 “动作”，以及根据当前进度值接下来可以采取的动作。动作值会影响最大Spout挂起值，可能的值为：增加（Increase）、减少（Decrease）或不变（No Change）。标记为 “增加” 的动作会将最大Spout挂起值提高 25%。“减少” 动作会将最大Spout挂起值降低 Max (25%, (上次进度增量 - 当前进度增量)/ 上次进度增量 * 100)%。“不变” 动作表示最大Spout挂起参数应保持与当前值相同。

autoTune 函数有一个状态机来确定它应该进行的下一个转换。接下来描述这个状态机转换：

• 如果最后一个动作等于 “不变”，那么
  • (i) 如果这是第一次调用自动调优，那么将动作设置为 “增加”，并增加最大Spout挂起值。
  • (ii) 如果上次进度增量高于当前进度增量，那么将动作设置为 “减少”，并减少最大Spout挂起值。
  • (iii) 如果上次进度增量低于当前进度增量，那么将动作设置为 “增加”，并增加最大Spout挂起值。
  • (iv) 如果上次进度增量与当前进度增量相似，那么将动作设置为 “不变”，并将一个计数器加 1，该计数器表示我们在这个 “不变” 状态中连续停留的次数。如果该计数器等于 5，那么将动作设置为 “增加”，并增加最大Spout挂起值。
• 如果最后一个动作等于 “增加”，那么
  • (i) 如果上次进度增量高于当前进度增量，那么将动作设置为 “减少”，并减少最大Spout挂起值。
  • (ii) 如果上次进度增量低于当前进度增量，那么将动作设置为 “增加”，并增加最大Spout挂起值。
  • (iii) 如果上次进度增量与当前进度增量相似，那么将动作设置为 “不变”，并将最大Spout挂起值恢复到上次增加之前的值。
• 如果最后一个动作等于 “减少”，那么
  • (i) 如果上次进度增量低于当前进度增量，那么将动作设置为 “增加”，并增加最大Spout挂起值。
  • (ii) 对于任何其他情况，将动作设置为 “不变”。

 

3.4 实证评估

在本节中，我们将展示为本文进行的实证评估结果。此次实证评估的目的是检验 Storm 在面对机器故障时的弹性和效率。

在这个实验中，我们创建了如下所示的示例拓扑结构，并以 “至少一次” 语义运行它（见 2.3 节）。这个拓扑结构主要是为此次实证评估构建的，不应被视为 Twitter Storm 工作负载的代表性拓扑结构。

为简化起见，在图 8 中，我们没有展示确认器（acker）Bolt。

图 8：实验中使用的示例拓扑结构

如图 8 所示，此拓扑结构有一个Spout。这个Spout是针对 “client_event” 信息流的 Kafka Spout。来自Spout的元组通过随机分组（shuffle grouped）发送到分发器（Distributor）Bolt，分发器Bolt根据名为 “user_id” 的属性 / 字段对数据进行分区。用户计数（UserCount）Bolt计算各种事件（如 “关注”“取消关注”“查看推文” 以及来自移动和网页客户端的其他事件）的唯一用户数量。这些计数每秒计算一次（即频率为 1 赫兹）。这些计数根据时间戳属性 / 字段进行分区，然后发送到下一个（聚合器，Aggregator）Bolt。聚合器Bolt聚合它接收到的所有计数。

3.4.1 实验设置

在这个实验中，我们准备了 16 台物理机器。拓扑结构中每个组件的初始任务数量如下所列：

工作进程的总数设定为 50，并且在整个实验过程中保持不变。我们在 16 台机器上启动该拓扑结构。接着，等待约 15 分钟后，移除 / 关停 3 台机器，然后将此步骤再重复 3 次。以下是对该实验设置的总结：

我们持续监测吞吐量（拓扑结构每分钟处理的元组数量），以及在拓扑结构中处理单个元组的平均端到端延迟（每分钟）。吞吐量以每分钟确认的元组数量（在确认器Bolt中）来衡量。以下是这些结果的报告。

3.4.2 结果

我们首先在下方汇报稳定的平均吞吐量和延迟情况。

通过可视化工具（见 3.2 节）得到的本次实验的吞吐量和延迟图表，分别展示在图 9 和图 10 中。

如图 9 所示，每当我们移除一组机器时，吞吐量都会出现暂时的峰值，但系统很快就能恢复。另外，请注意吞吐量每 15 分钟会下降，这是预期之中的，因为相同的拓扑结构运行在更少的机器上。从图中可以看出，在每个 15 分钟的时间段内，吞吐量都能相当迅速地趋于稳定。

图 9：吞吐量测量结果

图 10 展示了本次实验的延迟图表，正如预期的那样，每次移除一组机器时，延迟都会增加。注意在前几个 15 分钟的时间段内，延迟图表中的峰值较小，但在最后两个时间段（此时资源更为紧张），峰值更高；不过，如图所示，在所有情况下系统都能相当迅速地稳定下来。

图 10：延迟测量结果

总体而言，从这个实验可以看出，Storm 对机器故障具有弹性，并且在机器故障事件发生后，能够高效地使性能恢复稳定。

4. 结论与未来工作

Storm 是 Twitter 的关键基础设施，支撑着 Twitter 众多基于实时数据驱动的决策。Storm 在 Twitter 的应用正在迅速扩展，这也为未来的工作提出了一些潜在的有趣方向。其中包括静态地自动优化拓扑结构（Bolt内并行性以及执行器中任务的封装），并在运行时动态地重新优化。我们还希望探索添加 “恰好一次” 语义（类似于 Trident [24]），同时不造成较大的性能影响。此外，我们希望改进可视化工具，提高某些部分的可靠性（例如，将 Nimbus 本地磁盘中存储的状态转移到像 HDFS 这样更具容错性的系统中），更好地将 Storm 与 Hadoop 集成，并且有可能利用 Storm 进行监控、响应和自我调整，以改进正在运行的拓扑结构的配置。

未来工作的另一个有趣方向是为 Storm 支持一种声明式查询范式，同时仍能保持易于扩展。

5. 致谢

Storm 项目由 Nathan Marz 在 BackType 发起，Twitter 数据基础设施团队的无数其他成员对其进行了贡献、维护、运行和调试。我们感谢所有这些贡献者，没有他们的帮助与合作，就不可能有这篇论文。

6. 参考文献

[1] Arvind Arasu, Brian Babcock, Shivnath Babu, Mayur Datar, Keith Ito, Rajeev Motwani, Itaru Nishizawa, Utkarsh Srivastava, Dilys Thomas, Rohit Varma, Jennifer Widom: STREAM: The Stanford Stream Data Manager. IEEE Data Eng. Bull. 26(1): 19-26 (2003)

[2] Hari Balakrishnan, Magdalena Balazinska, Donald Carney,Ugur Çetintemel, Mitch Cherniack, Christian Convey,Eduardo F. Galvez, Jon Salz, Michael Stonebraker, Nesime Tatbul, Richard Tibbetts, Stanley B. Zdonik: Retrospective on Aurora. VLDB J. 13(4): 370-383 (2004)

[3] Minos N. Garofalakis, Johannes Gehrke: Querying and Mining Data Streams: You Only Get One Look. VLDB 2002

[4] Daniel J. Abadi, Yanif Ahmad, Magdalena Balazinska, Ugur Çetintemel, Mitch Cherniack, Jeong-Hyon Hwang, Wolfgang Lindner, Anurag Maskey, Alex Rasin, Esther Ryvkina, Nesime Tatbul, Ying Xing, Stanley B. Zdonik: The Design of the Borealis Stream Processing Engine. CIDR 2005: 277-289

[5] S4 Distributed stream computing platform.http://incubator.apache.org/s4/

[6] Tyler Akidau, Alex Balikov, Kaya Bekiroglu, Slava Chernyak, Josh Haberman, Reuven Lax, Sam McVeety,Daniel Mills, Paul Nordstrom, Sam Whittle: MillWheel:Fault-Tolerant Stream Processing at Internet Scale. PVLDB6(11): 1033-1044 (2013)

[7] Apache Samza. http://samza.incubator.apache.org

[8] Spark Streaming.http://spark.incubator.apache.org/docs/latest/streamingprogramming-guide.html

[9] Mohamed H. Ali, Badrish Chandramouli, Jonathan Goldstein, Roman Schindlauer: The extensibility framework in Microsoft StreamInsight. ICDE 2011: 1242-1253

[10] Sankar Subramanian, Srikanth Bellamkonda, Hua-Gang Li,Vince Liang, Lei Sheng, Wayne Smith, James Terry, TsaeFeng Yu, Andrew Witkowski: Continuous Queries in Oracle.VLDB 2007: 1173-1184

[11] IBM Infosphere Streams. http://www03.ibm.com/software/products/en/infosphere-streams/

[12] Namit Jain, Shailendra Mishra, Anand Srinivasan, Johannes Gehrke, Jennifer Widom, Hari Balakrishnan, Ugur Çetintemel, Mitch Cherniack, Richard Tibbetts, Stanley B.Zdonik: Towards a streaming SQL standard. PVLDB 1(2):1379-1390 (2008)

[13] Jay Kreps, Neha Narkhede, and Jun Rao. Kafka: a distributed messaging system for log processing. SIGMOD Workshop on Networking Meets Databases, 2011.

[14] Kestrel: A simple, distributed message queue system.http://robey.github.com/kestrel

[15] Benjamin Hindman, Andy Konwinski, Matei Zaharia, Ali Ghodsi, Anthony D. Joseph, Randy Katz, Scott Shenker, and Ion Stoica. 2011. Mesos: a platform for fine-grained resource sharing in the data center. In NSDI, 2011.

[16] Goetz Graefe: Encapsulation of Parallelism in the Volcano Query Processing System. SIGMOD Conference 1990: 102-111

[17] Apache Zookeeper. http://zookeeper.apache.org/

[18] Summingbird. https://github.com/Twitter/summingbird

[19] Rajagopal Ananthanarayanan, Venkatesh Basker, Sumit Das,Ashish Gupta, Haifeng Jiang, Tianhao Qiu, Alexey Reznichenko, Deomid Ryabkov, Manpreet Singh,Shivakumar Venkataraman: Photon: fault-tolerant and scalable joining of continuous data streams. SIGMOD
Conference 2013: 577-588

[20] Nathan Marz: (Storm) Tutorial.https://github.com/nathanmarz/storm/wiki/Tutorial

[21] Storm, Stream Data Processing:http://hortonworks.com/labs/storm/

[22] Apache Storm: http://hortonworks.com/hadoop/storm/

[23] Vinod Kumar Vavilapalli, Arun C. Murthy, Chris Douglas,Sharad Agarwal, Mahadev Konar, Robert Evans, Thomas Graves, Jason Lowe, Hitesh Shah, Siddharth Seth, Bikas Saha, Carlo Curino, Owen O'Malley, Sanjay Radia,Benjamin Reed, Eric Baldeschwieler: Apache Hadoop YARN: yet another resource negotiator. SoCC 2013: 5

[24] Nathan Marz: Trident API Overview.https://github.com/nathanmarz/storm/wiki/Trident-APIOverview

[25] ZeroMQ: http://zeromq.org/