引入

通过前面的文章，我们从S4到Storm，再到Storm结合Kafka成为当时的实时处理最佳实践：

• S4论文详解
• S4论文总结
• Storm论文详解
• Storm论文总结
• Kafka论文详解
• Kafka论文总结

然而Kafka+Storm的第一代流式数据处理组合，还面临的三个核心挑战：

1. 数据的正确性，也就是需要能够保障“正好一次”的数据处理。

2. 系统的容错能力，也就是我们不能因为某一台服务器的硬件故障，就丢失掉一部分数据。

3. 对于时间窗口的正确处理，也就是能够准确地根据事件时间生成报表，而不是简单地使用进行处理的服务器的本地时间。并且，还需要能够考虑到分布式集群中，数据的传输可能会有延时的情况出现。

当然这些问题并不是解决不了，我们可以通过在应用层撰写大量的代码，来进行数据去重、状态持久化。但是一个合理的解决方案， 应该是在流式计算框架层面就解决这些问题，而不是把这些问题留给应用开发人员。

围绕着这三个核心挑战，在2013年，Google的一篇论文《MillWheel: Fault-Tolerant Stream Processing at Internet Scale》给我们带来了一套解决方案。这个解决方案，可以算得上是第二代流式数据处理系统。

那么今天，我们就来看看，在2013年这个时间点上，Google的工程师是怎么解决这个问题的。

MillWheel：互联网规模下的容错流处理

摘要

MillWheel是一个用于构建低延迟数据处理应用程序的框架，在谷歌被广泛使用。用户只需指定有向计算图以及各个节点的应用代码，系统便会在框架的容错保障范围内，管理持久状态和连续的记录流。

本文介绍了MillWheel的编程模型及其实现方式。通过谷歌正在使用的一个连续异常检测器的案例研究，阐述了MillWheel的诸多特性是如何发挥作用的。MillWheel的编程模型引入了逻辑时间的概念，使得编写基于时间的聚合操作变得简单。

MillWheel从设计之初就考虑到了容错性和可扩展性。在实际应用中，我们发现MillWheel将可扩展性、容错性以及通用的编程模型独特结合，能够解决谷歌内部各种各样的问题。

1. 引言

流处理系统对于向用户提供内容以及帮助组织做出更快更好的决策至关重要，尤其是因为它们能够提供低延迟的结果。用户希望获取关于周围世界的实时新闻。企业同样对诸如垃圾邮件过滤和入侵检测等实时情报源所带来的价值感兴趣。类似地，科学家必须从海量的原始数据流中筛选出有价值的结果。

谷歌的流处理系统需要具备容错性、持久状态和可扩展性。分布式系统运行在数千台共享机器上，其中任何一台都可能随时发生故障。基于模型的流处理系统，如异常检测器，依赖于根据数周数据生成的预测，并且随着新数据的到来，它们的模型必须实时更新。将这些系统的规模扩大几个数量级，不应导致构建和维护系统的运营成本相应增加。

像MapReduce [11]这样的分布式系统编程模型，将框架的实现细节隐藏在幕后，使用户能够创建表达简洁的大规模分布式系统。通过让用户只专注于应用逻辑，这种编程模型使用户无需成为分布式系统专家，就能对系统的语义进行推理。

具体而言，用户能够将框架层面的正确性和容错性保障视为公理，极大地限制了错误可能出现的范围。支持多种常见编程语言进一步推动了其应用，因为用户可以以熟悉的方式利用现有库的实用性和便利性，而不是局限于特定领域的语言。

MillWheel就是这样一种专门为流处理、低延迟系统量身定制的编程模型。用户将应用逻辑编写为有向计算图中的各个节点，并可以为其定义任意的、动态的拓扑结构。记录沿着图中的边持续传递。MillWheel在框架层面提供容错能力，拓扑结构中的任何节点或边随时可能发生故障，但不会影响结果的正确性。作为这种容错能力的一部分，系统中的每一条记录都保证会被传递给其消费者。此外，MillWheel为记录处理提供的API以幂等方式处理每条记录，从用户的角度来看，记录只会被传递一次。MillWheel以细粒度检查点记录其进展，无需在检查点之间的长时间内，在外部发送方缓冲待处理的数据。

其他流处理系统并未同时具备这种容错性、通用性和可扩展性。Spark Streaming [34]和Sonora [32]在高效检查点方面表现出色，但限制了用户代码可用的操作符范围。S4 [26]没有提供完全容错的持久状态，而Storm [23]用于记录传递的“恰好一次”机制Trident [22]，需要严格的事务顺序才能运行。试图将MapReduce和Hadoop [4]的批处理模型扩展以提供低延迟系统，会导致灵活性受损，例如Spark Streaming中操作符对弹性分布式数据集（Replicated Distributed Datasets）[33]的特定依赖。流SQL系统[1] [2] [5] [6] [21] [24]为许多流处理问题提供了简洁的解决方案，但直观的状态抽象和复杂的应用逻辑（例如矩阵乘法）使用命令式语言的操作流程来表达，比使用像SQL这样的声明式语言更为自然。

我们的贡献在于为流处理系统设计了一种编程模型，并实现了MillWheel框架。

• 我们设计了一种编程模型，无需分布式系统专业知识，就能创建复杂的流处理系统。

• 我们构建了MillWheel框架的高效实现，证明了它作为一个可扩展且容错的系统的可行性。

本文其余部分组织如下。第2节概述了MillWheel开发的一个动机示例，以及它所带来的相应需求。第3节对系统进行了高层次的概述。第4节定义了MillWheel模型的基本抽象，第5节讨论了MillWheel公开的API。第6节概述了MillWheel中容错的实现，第7节涵盖了一般实现。第8节提供实验结果以说明MillWheel的性能，第9节讨论相关工作。

2. 动机与需求

谷歌的Zeitgeist管道用于跟踪网络查询趋势。为了展示MillWheel功能集的实用性，我们将研究Zeitgeist系统的需求。这个管道持续接收搜索查询的输入，并进行异常检测，尽快输出搜索量激增或骤减的查询。该系统为每个查询构建一个历史模型，以便预期的流量变化（例如傍晚时分的“电视节目预告”查询）不会导致误报。尽快识别出搜索量激增或骤减的查询非常重要。例如，Zeitgeist为谷歌的热门趋势服务提供支持，该服务依赖最新信息。此管道的基本拓扑结构如图1所示。

图 1：输入数据（搜索查询）经过一系列 MillWheel 计算，这些计算以分布式进程的形式呈现。该系统的输出由外部异常通知系统使用。

为了实现Zeitgeist系统，我们的方法是将记录按一秒的时间间隔进行分桶，然后将每个时间桶的实际流量与模型预测的预期流量进行比较。如果在相当数量的桶中，这些数量持续不同，那么我们就有很大把握认为某个查询的搜索量在激增或骤减。同时，我们用新接收到的数据更新模型，并存储起来供将来使用。

1. 持久存储：需要注意的是，此实现既需要短期存储，也需要长期存储。搜索量激增可能只持续几秒钟，因此依赖于一小段时间窗口内的状态，而模型数据可能对应数月的持续更新。

2. 低水位标记：一些Zeitgeist用户对检测流量骤减感兴趣，即某个查询的流量异常低（例如埃及政府关闭互联网的情况）。在一个接收来自世界各地输入的分布式系统中，数据到达时间并不严格对应其生成时间（在这种情况下是搜索时间），因此能够区分在t = 1296167641时预期的大量阿拉伯语查询是在传输中延迟了，还是实际上根本没有出现，这一点很重要。MillWheel通过为每个处理阶段（例如窗口计数器、模型计算器）的传入数据提供低水位标记来解决这个问题，该标记表明截至给定时间戳的所有数据都已收到。低水位标记跟踪分布式系统中的所有待处理事件。利用低水位标记，我们能够区分上述两种情况——如果低水位标记超过时间t，而查询仍未到达，那么我们就有很大把握认为这些查询没有被记录，而不仅仅是延迟了。这种语义也消除了对输入严格单调性的任何要求——乱序流是常态。

3. 防止重复：对于Zeitgeist来说，重复的记录传递可能会导致虚假的搜索量激增。此外，“恰好一次”处理是MillWheel众多收益处理客户的要求，他们都可以依赖框架实现的正确性，而无需重新发明自己的去重机制。用户无需编写代码手动回滚状态更新，或处理各种故障场景以保持正确性。

考虑到以上因素，我们提出谷歌对流处理框架的需求，这些需求在MillWheel中得到了体现：

• 数据一旦发布，就应该尽快提供给消费者（即摄取输入和提供输出数据不存在系统内在的障碍）。

• 用户代码应该能够使用持久状态抽象，并且这些抽象应该集成到系统的整体一致性模型中。

• 系统应该能够优雅地处理乱序数据。

• 系统应该计算数据时间戳的单调递增低水位标记。

• 随着系统扩展到更多机器，延迟应该保持不变。

• 系统应该提供记录的“恰好一次”传递。

3. 系统概述

从高层次来看，MillWheel是一个对输入数据进行用户定义转换以生成输出数据的图。我们将这些转换称为计算，下面会更详细地定义它们。这些转换中的每一个都可以在任意数量的机器上并行化，这样用户就无需在细粒度层面担心负载均衡问题。以图1所示的Zeitgeist系统为例，我们的输入是不断到达的一组搜索查询，输出则是搜索量激增或骤减的查询集合。

抽象地说，MillWheel中的输入和输出由（键、值、时间戳）三元组表示。键是系统中具有语义意义的元数据字段，而值可以是任意字节字符串，对应整个记录。用户代码运行的上下文限定在特定的键范围内，每个计算可以根据其逻辑需求为每个输入源定义键。例如，Zeitgeist中的某些计算可能会选择搜索词（如“猫视频”）作为键，以便按每个查询计算统计信息，而其他计算可能会选择地理来源作为键，以便按每个地区进行聚合。这些三元组中的时间戳可以由MillWheel用户指定任意值（但通常接近事件发生时的实际时钟时间），MillWheel将根据这些值计算低水位标记。

如果用户要按秒统计搜索词的计数（如图2所示的Zeitgeist系统中的情况），那么他们会希望为搜索执行的时间分配一个对应的时间戳值。

图2：将网页搜索聚合到以秒为单位的时间桶中，并使用每个键对应的持久状态来更新模型。每个计算都可以访问其自身每个键对应的状态，并根据输入记录对其进行更新。

3. 系统概述

总体而言，MillWheel是一个对输入数据进行用户定义转换以生成输出数据的图。我们将这些转换称为“计算”，后续会进一步详细定义。每个转换都能在任意数量的机器上并行处理，这样用户就无需操心细粒度的负载均衡问题。以图1所示的Zeitgeist系统为例，输入是源源不断的搜索查询，输出则是搜索量激增或骤减的查询集合。

抽象来讲，MillWheel中的输入和输出由（键、值、时间戳）三元组表示。键是系统中具有语义意义的元数据字段，值可以是任意字节字符串，对应整条记录。用户代码的运行上下文限定于特定的键，每个计算可依据自身逻辑需求，为每个输入源定义键。例如，在Zeitgeist系统中，某些计算可能会选择搜索词（如“猫视频”）作为键，以便按每个查询统计数据；而其他计算可能会选择地理来源作为键，以便按地区进行聚合。这些三元组中的时间戳可由MillWheel用户指定任意值（通常接近事件发生时的实际时钟时间），MillWheel会依据这些值计算低水位标记。

倘若用户要按秒统计搜索词的数量（如图2所示的Zeitgeist系统场景），那么就会为搜索执行的时间分配对应的时间戳值。

4. 核心概念

MillWheel呈现了流处理系统的关键要素，并提供了清晰的抽象概念。数据通过用户定义的有向计算图在系统中传输（图3），每个计算都能独立地处理和输出数据。

图3：MillWheel拓扑结构中单个节点的定义。输入流和输出流分别对应图中的有向边。

4.1 计算

应用逻辑包含在“计算”中，“计算”封装了任意的用户代码。当接收到输入数据时，会调用计算代码，此时会触发用户定义的操作，包括与外部系统交互、操作其他MillWheel基本元素或输出数据。如果要与外部系统交互，用户需确保其代码对这些系统的操作具有幂等性。计算代码在单个键的上下文中运行，并且无需知晓不同机器间键的分布情况。如图4所示，处理过程按每个键进行序列化，但不同键之间可以并行处理。

图4：基于每个键的处理随时间序列化，因此对于给定的键，一次只能处理一条记录。多个键的处理可以并行运行。

4.2 键

在MillWheel中，键是用于不同记录间聚合和比较的主要抽象概念。对于系统中的每条记录，使用者会指定一个键提取函数，为该记录分配一个键。计算代码在特定键的上下文中运行，并且只能访问该特定键对应的状态。例如，在Zeitgeist系统中，对于查询记录，选择查询文本本身作为键是个不错的选择，因为我们需要按每个查询来聚合计数并计算模型。又如，垃圾邮件检测器可能会选择cookie指纹作为键，以阻止恶意行为。图5展示了不同的使用者从同一输入流中提取不同的键。

图5：多个计算可以从同一流中提取不同的键。键提取器由流的消费者指定。

4.3 流

流是MillWheel中不同计算之间的数据传输机制。一个计算可以订阅零个或多个输入流，并发布一个或多个输出流，系统保证数据会沿这些通道传输。每个使用者针对每个流指定键提取函数，这样多个使用者就可以订阅同一流，并以不同方式聚合其数据。流通过名称唯一标识，没有其他限定条件——任何计算都可以订阅任何流，也可以向任何流生成记录（输出）。

4.4 持久状态

在MillWheel中，最基本形式的持久状态是一个不透明的字节字符串，按每个键进行管理。用户提供序列化和反序列化例程（例如将复杂数据结构转换为网络传输格式或从网络传输格式转换回复杂数据结构），对此有多种便捷机制（如Protocol Buffers [13]）可用。持久状态由一个复制的、高可用的数据存储（如Bigtable [7] 或Spanner [9]）支持，以一种对终端用户完全透明的方式确保数据完整性。状态的常见用途包括在记录窗口上聚合的计数器以及用于连接的缓冲数据。

4.5 低水位标记

一个计算的低水位标记为即将到达该计算的记录的时间戳设定了界限。

定义：我们基于管道的数据流递归定义低水位标记。给定一个计算A，设A的最早工作时间是与A中最早未完成（在途、存储或待交付）记录对应的时间戳。基于此，我们将A的低水位标记定义为： min(A的最早工作时间，C的低水位标记：C向A输出)

如果没有输入流，低水位标记和最早工作时间的值相等。

低水位标记值由注入器设定，注入器将外部系统的数据发送到MillWheel。对外部系统中待处理工作的测量通常是一种估计，因此在实际中，计算应该预期会从这类系统收到少量迟到的记录——即落后于低水位标记的记录。Zeitgeist系统通过丢弃此类数据来处理这个问题，同时记录丢弃的数据量（根据经验，大约为记录总数的0.001%）。其他管道如果收到迟到的记录，则会追溯修正其聚合结果。尽管上述定义中未体现，但系统保证即使面对迟到的数据，计算的低水位标记也是单调递增的。

通过等待一个计算的低水位标记超过某个值，用户可以确定截至该时间他们拥有完整的数据情况，正如Zeitgeist系统检测流量骤减的示例所示。在为新记录或聚合记录分配时间戳时，用户需选择不小于任何源记录时间戳的值。MillWheel框架报告的低水位标记衡量系统中已知的工作，如图6所示。

图6：随着记录在系统中流转，低水位标记不断推进。在每个时间点的展示中，待处理记录显示在时间戳轴上方，已完成记录显示在下方。新记录在后续时间点以待处理任务的形式出现，其时间戳值在水位标记之前。数据不一定按顺序处理，低水位标记反映了系统中所有的待处理任务。

4.6 定时器

定时器是基于每个键的编程钩子，在特定的实际时间或低水位标记值触发。定时器在计算的上下文中创建并运行，因此可以运行任意代码。使用实际时间还是低水位标记值取决于应用场景——一个希望按小时发送电子邮件（整点发送，无论数据是否延迟）的启发式监控系统可能会使用实际时间定时器，而执行窗口聚合的分析系统可能会使用低水位标记定时器。一旦设定，定时器保证按时间戳递增的顺序触发。它们被记录在持久状态中，并且可以在进程重启和机器故障后继续存在。当定时器触发时，它会运行指定的用户函数，并且与输入记录一样具有“恰好一次”的保证。Zeitgeist系统中检测流量骤减的一个简单实现可以为给定时间桶的结束时间设置一个低水位标记定时器，如果观察到的流量远低于模型预测，则报告流量骤减。

定时器的使用是可选的——不需要基于时间的屏障语义的应用程序可以跳过它们。例如，Zeitgeist系统可以在不使用定时器的情况下检测搜索量激增的查询，因为即使没有完整的数据，搜索量激增也可能很明显。如果观察到的流量已经超过模型预测，延迟的数据只会增加总数并增大激增的幅度。

5. API

在本节中，我们将概述与第4节中抽象概念相关的API。用户实现Computation类的自定义子类，如图7所示，该子类提供了访问所有MillWheel抽象（状态、定时器和输出）的方法。一旦用户提供了此代码，框架便会自动运行它。按键序列化在框架级别处理，用户无需构建任何按键锁定语义。

图7：MillWheel API包含一个父类Computation，通过它可以访问基于每个键的定时器、状态以及输出操作。用户通过重写ProcessRecord和ProcessTimer方法来实现应用逻辑。

5.1 计算API

用户代码的两个主要入口点由ProcessRecord和ProcessTimer钩子提供，如图8所示，它们分别在接收到记录和定时器到期时触发。这些共同构成了一个计算的应用逻辑。

图8：MillWheel系统会响应传入的远程过程调用（RPC）来调用用户定义的处理钩子。用户代码通过框架应用程序编程接口（API）访问状态、定时器和输出。框架负责执行所有实际的RPC操作和状态修改。

在这些钩子的执行过程中，MillWheel提供系统函数来获取和操作按键状态、生成额外记录以及设置定时器。图9展示了这些机制之间的交互。它以我们的Zeitgeist系统为例，展示了在检测查询流中的流量骤减时持久状态和定时器的使用。再次注意，这里没有故障恢复逻辑，因为这一切都由框架自动处理。

图9：基于低水位标记定时器，利用现有模型计算窗口计数和流量骤减的计算所对应的ProcessRecord和ProcessTimer的定义。  

5.2 注入器和低水位标记API

在系统层，每个计算为其所有待处理工作（进行中的和排队交付的）计算一个低水位标记值。持久状态也可以分配一个时间戳值（例如聚合窗口的后沿）。系统会自动汇总这些值，以便以透明的方式为定时器提供API语义——用户在计算代码中很少直接与低水位标记交互，而是通过为记录分配时间戳来间接操作它们。

注入器：注入器将外部数据引入MillWheel。由于注入器为管道的其余部分设定低水位标记值，它们能够发布一个注入器低水位标记，该标记会传播到其输出流中的任何订阅者，反映它们沿这些流的潜在交付情况。例如，如果一个注入器正在摄取日志文件，它可以发布一个与未完成文件中最小文件创建时间相对应的低水位标记值，如图10所示。

图10：一个简单的文件注入器报告的低水位标记值，与最旧的未完成文件相对应。

一个注入器可以分布在多个进程中，这些进程的聚合低水位标记将用作注入器低水位标记。用户可以指定预期的注入器进程集，使此指标能够抵御进程故障和网络中断。在实际中，谷歌针对常见输入类型（日志文件、发布/订阅服务源等）有相应的库实现，普通用户无需编写自己的注入器。

如果一个注入器违反低水位标记语义并发送了落后于低水位标记的迟到记录，用户的应用代码可以选择丢弃该记录或将其合并到现有聚合的更新中。

6. 容错机制

6.1 交付保证

MillWheel编程模型在概念上的简洁性，很大程度上依赖于它能够将非幂等的用户代码当作幂等代码来运行。通过从计算编写者那里移除这一要求，我们减轻了他们显著的实现负担。

6.1.1 恰好一次交付

当一个计算接收到输入记录时，MillWheel框架会执行以下步骤：

1. 将该记录与之前交付的去重数据进行比对，重复记录将被丢弃。

2. 针对输入记录运行用户代码，这可能会导致对定时器、状态和输出的待处理更改。

3. 将待处理更改提交到后端存储。

4. 向发送方发送确认（ACK）。

5. 发送待处理的下游输出。

作为一种优化，上述操作可能会合并为对多条记录的单个检查点。MillWheel中的交付会一直重试，直到收到确认，以满足我们“至少一次”的要求，这是实现“恰好一次”交付的先决条件。我们进行重试是考虑到接收方可能出现网络问题和机器故障。然而，这就带来了一种情况，即接收方可能在有机会确认输入记录之前崩溃，即使它已经持久化了与该记录成功处理相对应的状态。在这种情况下，当发送方重试交付时，我们必须防止重复处理。

系统在生成记录时为所有记录分配唯一ID。我们通过在与状态修改相同的原子写入中包含该记录的唯一ID来识别重复记录。如果之后重试相同的记录，我们可以将其与记录日志中的ID进行比较，并丢弃重复记录并发送确认（以免它无限期重试）。由于我们不一定能将所有重复数据存储在内存中，我们维护一个已知记录指纹的布隆过滤器，为我们肯定从未见过的记录提供快速路径。如果过滤器判断失误，我们必须读取后端存储以确定记录是否重复。在MillWheel能够保证所有内部发送方都完成重试之后，过去交付的记录ID将被垃圾回收。对于经常发送迟到数据的注入器，我们将此垃圾回收延迟相应的松弛值（通常为几个小时）。不过，“恰好一次”的数据通常可以在生成后的几分钟内清理完毕。

6.1.2 强输出

由于MillWheel处理的输入不一定是有序或确定的，我们在交付之前将生成的记录与状态修改在相同的原子写入中进行检查点操作。我们将这种在记录生成之前进行检查点操作的模式称为“强输出”。

以一个按实际时间聚合并向下游发送计数的计算为例。如果没有检查点，该计算有可能向下游生成一个窗口计数，但在保存其状态之前崩溃。一旦该计算重新启动，它可能在生成相同的聚合之前接收到另一条记录（并将其添加到计数中），从而创建一个与前一个记录在逐位上不同但对应相同逻辑窗口的记录！为了正确处理这种情况，下游消费者需要复杂的冲突解决逻辑。然而，使用MillWheel，简单的解决方案就能奏效，因为系统保证将用户的应用逻辑变成了幂等操作。

我们使用诸如Bigtable [7] 这样的存储系统，它有效地实现了盲目写入（与读取 - 修改 - 写入操作相反），使得检查点模仿日志的行为。当一个进程重新启动时，检查点被扫描到内存中并重新播放。一旦这些输出成功，检查点数据将被删除。

6.1.3 弱输出和幂等性

强输出和恰好一次交付相结合，使得许多计算在系统级重试方面具有幂等性。然而，一些计算本身可能已经具有幂等性，无论是否存在这些保证（这些保证伴随着资源和延迟成本）。根据应用程序的语义需求，用户可以自行决定禁用强输出和/或恰好一次交付。在系统层面，禁用恰好一次交付可以简单地通过跳过去重步骤来实现，但禁用强输出需要更多地关注性能。

对于弱输出，我们不是在交付之前对记录输出进行检查点操作，而是在持久化状态之前乐观地向下游广播交付。根据经验，这引入了一个新问题，即管道连续阶段的完成时间现在严格耦合，因为它们等待下游对记录的确认。再加上机器故障的可能性，随着管道深度的增加，这会极大地增加掉队输出的端到端延迟。例如，如果我们（相当悲观地）假设任何一台机器在给定分钟内有1%的故障概率，那么我们等待至少一次故障的概率会随着管道深度的增加而急剧增加——对于深度为5的管道，给定的输出每分钟可能有近5%的故障概率！我们通过对一小部分掉队的待处理输出进行检查点操作来缓解这个问题，允许这些阶段向它们的发送方发送确认。通过以这种方式选择性地进行检查点操作，我们既可以提高端到端延迟，又可以降低总体资源消耗。

在图11中，我们展示了这种检查点机制的实际运行情况。计算A向计算B生成输出，计算B立即向计算C生成输出。然而，计算C确认速度较慢，因此计算B在1秒延迟后对输出进行检查点操作。这样，计算B可以向计算A确认交付，使A能够释放与该输出相关的任何资源。即使计算B随后重新启动，它也能够从检查点恢复记录并重新尝试向计算C交付，而不会丢失数据。

图11：弱输出检查点机制通过为计算B保存检查点，防止了掉队的输出在发送方（计算A）占用过多资源。

上述放宽条件适用于具有幂等计算的管道，因为重试不会影响正确性，并且下游输出也对重试不敏感。无状态过滤器就是一个幂等计算的实际例子，沿着输入流的重复交付不会改变结果。

6.2 状态操作

在实现MillWheel中用于操作用户状态的机制时，我们讨论了持久化到后端存储的“硬”状态以及包括任何内存缓存或聚合的“软”状态。我们必须满足以下用户可见的保证：

1. 系统不会丢失数据。

2. 状态更新必须遵循“恰好一次”语义。在任何给定时间，整个系统中的所有持久化数据都必须保持一致。

3. 低水位标记必须反映系统中的所有待处理状态。

4. 对于给定的键，定时器必须按顺序触发。

为避免持久化状态中的不一致（例如定时器、用户状态和输出检查点之间），我们将所有按键更新包装在单个原子操作中。这使得系统能够抵御进程故障和其他可能在任何给定时间中断进程的不可预测事件。如前所述，“恰好一次”数据也在同一操作中更新，将其添加到按键一致性范围内。

由于工作可能在机器之间转移（由于负载均衡、故障或其他原因），我们数据一致性的一个主要威胁是可能存在僵尸写入者和网络残留，它们会向后端存储发出过时的写入。为解决这一可能性，我们为每次写入附加一个序列器令牌，后端存储的中介在允许写入提交之前会检查其有效性。新的工作进程在开始工作之前会使任何现有的序列器无效，这样此后残留的写入就无法成功。序列器的作用类似于Centrifuge [3] 系统中的租约执行机制。因此，我们可以保证，对于给定的键，在特定时间点只有一个工作进程可以写入该键。

这种单写入者保证对于维护软状态也至关重要，并且无法通过依赖事务来保证。以挂起定时器的缓存为例：如果另一个进程的残留写入在缓存构建后可以改变持久化的定时器状态，那么缓存就会不一致。图12展示了这种情况，僵尸进程（B）响应来自A的输出发出一个在网络中延迟的事务。在事务开始之前，B的后继进程B - prime对挂起的定时器进行初始扫描。扫描完成后，应用事务并向A发送确认，导致B - prime的定时器状态不完整。丢失的定时器可能会无限期地孤立，任意延迟其任何输出操作。显然，这对于对延迟敏感的系统是不可接受的。

图12：事务无法防止软状态中的不一致情况。孤立的事务可能在只读扫描完成后提交，从而在MillWheel的定时器系统中导致状态不一致。

此外，对于已检查点的输出也可能出现同样的情况，系统可能会因未对后端存储进行初始扫描而不知道该输出。在发现该输出之前，低水位标记中不会考虑它，在此期间，我们可能会向消费者报告错误的低水位标记值。此外，由于我们的低水位标记是单调递增的，我们无法纠正错误增加的值。通过违反我们的低水位标记保证，可能会出现各种正确性违规情况，包括定时器过早触发和窗口输出不完整。

为了快速从意外的进程故障中恢复，MillWheel中的每个计算工作进程都可以以任意细粒度检查点其状态（在实际中，根据输入量，通常为亚秒级或每条记录的粒度）。我们使用始终一致的软状态，使得我们能够将必须扫描这些检查点的情况最小化到特定情况——机器故障或负载均衡事件。当我们确实执行扫描时，这些扫描通常可以是异步的，允许计算在扫描进行时继续处理输入记录。

7. 系统实现

7.1 架构

MillWheel部署在一组动态的主机服务器上作为分布式系统运行。管道中的每个计算在一台或多台机器上运行，流通过RPC（远程过程调用）进行交付。在每台机器上，MillWheel系统整理传入的工作并管理进程级元数据，必要时委托给适当的用户计算。

7. 系统实现

7.1 架构

MillWheel以分布式系统的形式在一组动态的主机服务器上运行。管道中的每个计算在一台或多台机器上执行，流数据通过远程过程调用（RPC）进行传输。在每台机器上，MillWheel系统会整理传入的任务，并管理进程级别的元数据，必要时将任务委托给相应的用户计算程序。

负载分配与均衡由一个复制的主节点负责，它将每个计算划分为一组按字典序排列的键区间（这些区间共同覆盖所有可能的键），并将这些区间分配给一组机器。当检测到CPU负载增加或内存压力增大（由标准的进程监视器报告）时，主节点可以移动、拆分或合并这些区间。每个区间都会被分配一个唯一的序列器，每当该区间被移动、拆分或合并时，序列器就会失效。第6.2节讨论了这个序列器的重要性。

对于持久状态，MillWheel使用类似Bigtable [7] 或Spanner [9] 这样的数据库，它们支持原子性的单行更新。给定键的定时器、待处理的输出以及持久状态都存储在数据存储的同一行中。

每当一个键区间被分配给新的所有者时，MillWheel通过从这个后端存储中扫描元数据，能够有效地从机器故障中恢复。这个初始扫描会填充内存中的数据结构，比如待处理定时器的堆和已检查点输出的队列，在区间分配的有效期内，这些内存结构会被认为与后端存储保持一致。为了支持这一假设，我们实施了第6.2节中详细描述的单写入者语义（每个计算工作进程）。

7.2 低水位标记

为确保数据一致性，低水位标记必须作为一个全局可用且准确的子系统来实现。

我们将其实现为一个中央机构（类似于OOP [19]），该机构跟踪系统中的所有低水位标记值，并将它们记录到持久状态中，以防止在进程故障时报告错误的值。

当向中央机构报告时，每个进程会聚合其所有已分配任务的时间戳信息。这包括任何已检查点或待处理的输出，以及任何待处理的定时器或持久化状态。每个进程能够高效地完成此操作，这得益于我们内存数据结构的一致性，从而无需对后端数据存储执行任何昂贵的查询。由于进程是根据键区间分配任务的，低水位标记更新也会按键区间进行分组，并发送到中央机构。

为了准确计算系统的低水位标记，这个中央机构必须能够访问系统中所有待处理和已持久化任务的低水位标记信息。在聚合每个进程的更新时，它通过为每个计算构建低水位标记值的区间映射，来跟踪其关于每个计算的信息完整性。如果任何区间缺失，那么在该缺失区间报告新值之前，低水位标记将对应于该缺失区间的最后已知值。然后，中央机构会广播系统中所有计算的低水位标记值。

感兴趣的消费者计算会订阅其每个发送者计算的低水位标记值，并将其输入的低水位标记计算为这些值中的最小值。

之所以由工作进程计算这些最小值，而不是由中央机构计算，是出于一致性的考虑：中央机构的低水位标记值应始终至少与工作进程的一样保守。因此，通过让工作进程计算其各自输入的最小值，中央机构的低水位标记永远不会领先于工作进程的，从而保持了这一特性。

为了在中央机构维护一致性，我们为所有低水位标记更新附加序列器。与我们对键区间状态进行本地更新的单写入者方案类似，这些序列器确保只有给定键区间的最新所有者才能更新其低水位标记值。为了实现可扩展性，中央机构可以分布在多台机器上，每个工作节点上运行一个或多个计算。根据经验，这种方式可以扩展到500,000个键区间，且性能不会下降。

鉴于系统中工作的全局摘要，我们可以选择去除异常值，并为那些更关注速度而非准确性的管道提供启发式的低水位标记值。例如，我们可以计算一个99%低水位标记，它对应于系统中99%记录时间戳的进度。只对近似结果感兴趣的窗口消费者可以使用这些低水位标记值，以较低的延迟运行，因为无需等待掉队的数据。

总之，我们对低水位标记的实现并不要求系统中的流具有任何严格的时间顺序。低水位标记反映了在途和已持久化的状态。通过为低水位标记值建立一个全局真实源，我们防止了诸如低水位标记后退之类的逻辑不一致情况。

8. 评估

为了说明MillWheel的性能，我们提供针对流处理系统关键指标的实验结果。

8.1 输出延迟

延迟是衡量流处理系统性能的关键指标。MillWheel框架支持低延迟的处理结果，并且随着分布式系统扩展到更多机器，它能保持较低的延迟。为了展示MillWheel的性能，我们使用一个简单的单阶段MillWheel管道来测量记录交付延迟，该管道对数字进行分桶和排序。这类似于在具有不同键的连续计算之间发生的多对多洗牌操作，因此是MillWheel中记录交付的一种最坏情况。图13显示了在200个CPU上运行时记录的延迟分布。记录的中位延迟为3.6毫秒，95% 百分位数延迟为30毫秒，这轻松满足了谷歌许多流处理系统的要求（即使是95% 百分位数也在人类反应时间之内）。

图13：两个不同键阶段之间单阶段记录延迟的直方图。

此测试是在禁用强输出和恰好一次交付功能的情况下进行的。当启用这两个功能时，中位延迟跃升至33.7毫秒，95% 百分位数延迟升至93.8毫秒。这清楚地表明了幂等计算如何通过禁用这两个功能来降低延迟。为了验证MillWheel的延迟特性与系统资源占用的扩展性良好，我们使用从20个CPU到2000个CPU的不同规模设置运行单阶段延迟实验，并按比例缩放输入。图14显示，无论系统规模如何，中位延迟大致保持不变。99% 百分位数延迟确实显著变差（尽管仍在100毫秒量级）。然而，随着规模的扩大，尾部延迟预计会下降——更多的机器意味着出现问题的机会更多。

图14：随着系统资源规模的扩大，MillWheel的平均延迟并未显著增加。

8.2 水位标记滞后

虽然有些计算（如Zeitgeist中的峰值检测）不需要定时器，但许多计算（如骤降检测）使用定时器等待低水位标记推进后再输出聚合结果。对于这些计算，低水位标记落后于实时的程度限制了这些聚合结果的新鲜度。由于低水位标记从注入器通过计算图传播，我们预计一个计算的低水位标记滞后与它到注入器的最大管道距离成正比。我们在200个CPU上运行一个简单的三阶段MillWheel管道，并每秒轮询一次每个计算的低水位标记值。在图15中，我们可以看到第一阶段的水位标记滞后实时1.8秒，然而，对于后续阶段，每阶段的滞后增加小于200毫秒。减少水位标记滞后是一个活跃的开发领域。

图15：一个三阶段管道中的低水位标记滞后情况。具体数据：{阶段1：均值1795，标准差159。阶段2：均值1954，标准差127。阶段3：均值2081，标准差140}  

8.3 框架级缓存

由于MillWheel的检查点频率较高，它会给存储层带来大量流量。当使用诸如Bigtable这样的存储系统时，读取的成本高于写入，MillWheel通过框架级缓存来缓解这一问题。MillWheel的一个常见用例是在存储中缓冲数据，直到低水位标记超过窗口边界，然后获取数据进行聚合。这种使用模式对存储系统中常见的LRU缓存不友好，因为最近修改的行是最不可能很快被读取的。MillWheel知道这些数据可能的使用方式，并可以提供更好的缓存淘汰策略。在图16中，我们测量了MillWheel工作进程和存储层的综合CPU使用率与最大缓存大小的关系（出于公司保密原因，CPU使用率已标准化）。增加可用缓存会线性改善CPU使用率（在550MB之后，大多数数据被缓存，因此进一步增加没有帮助）。在这个实验中，MillWheel的缓存能够将CPU使用率降低一半。

图16：MillWheel与存储层的总CPU负载 对比 框架缓存大小。

8.4 实际应用部署

MillWheel为谷歌内部各种各样的系统提供支持。它为各种广告客户执行流连接操作，其中许多客户要求对客户可见的仪表板进行低延迟更新。计费管道依赖于MillWheel的恰好一次交付保证。除了Zeitgeist之外，MillWheel还为一个通用的异常检测服务提供支持，许多不同的团队将其作为一个交钥匙解决方案使用。其他应用包括网络交换机和集群健康监测。MillWheel还为面向用户的工具提供支持，如谷歌街景的图像全景生成和图像处理。

也存在一些MillWheel不太适合的问题。本质上难以进行检查点操作的单体操作不太适合包含在计算代码中，因为系统的稳定性依赖于动态负载均衡。如果负载均衡器遇到与这种操作重合的热点，它必须选择要么中断操作并强制其重新启动，要么等待其完成。前者会浪费资源，而后者可能会使机器过载。作为一个分布式系统，MillWheel在那些不易在不同键之间并行化的问题上表现不佳。如果一个管道90% 的流量被分配到单个键，那么一台机器必须处理该流90% 的整体系统负载，这显然是不可取的。建议计算代码的编写者避免使用流量大到足以在单个机器上形成瓶颈的键（如客户语言或用户代理字符串），或者构建一个两阶段聚合器。

如果一个计算基于低水位标记定时器进行聚合，并且数据延迟长时间阻碍低水位标记推进，那么MillWheel的性能会下降。这可能导致系统中缓冲记录出现数小时的偏差。通常内存使用与偏差成正比，因为应用程序依赖低水位标记来刷新这些缓冲数据。为了防止内存使用无限制增长，一个有效的补救方法是通过等待低水位标记推进后再注入新记录，来限制系统中的总偏差。

9. 相关工作

我们构建流处理系统通用抽象的动机，在很大程度上受到了MapReduce [11] 在变革批处理领域所取得成功的影响，Apache Hadoop [4] 的广泛应用便是例证。将MillWheel与现有的流处理系统模型（如Yahoo! S4 [26]、Storm [23] 和Sonora [32]）进行比较，我们发现它们的模型对于我们期望解决的问题类别来说通用性不足。具体而言，S4和Sonora没有涉及恰好一次处理和容错持久状态，而Storm直到最近才通过Trident [22] 添加了此类支持，Trident为了正常运行，对事务ID施加了严格的排序要求。Logothetis等人也提出了类似的观点，强调了一流用户状态的必要性 [20]。Ciel [25] 针对一般的数据处理问题，同时动态生成数据流图。与MapReduce Online [8] 一样，我们认为为用户提供 “早期返回” 具有极大的实用价值。谷歌的Percolator [27] 也针对大型数据集的增量更新，但期望的延迟在分钟量级。

在评估我们的抽象时，我们注意到我们满足了Stonebraker等人 [30] 列举的流处理系统的要求。我们对乱序数据的灵活性与OOP方法 [19] 类似，OOP有力地论证了进行全局低水位标记计算（而非操作符级别的计算）的必要性，并令人信服地否定了使用静态松弛值来补偿乱序数据的可行性。

虽然我们赞赏Spark Streaming [34] 提出的针对特定操作符的流批系统统一方案，但我们认为MillWheel解决了更广泛的问题集，并且微批处理模型如果不将用户限制在预定义的操作符上是不可行的。具体来说，这个模型严重依赖RDDs [33]，这将用户限制在具有回滚能力的操作符上。

检查点和恢复是任何流处理系统的关键方面，我们的方法与之前的许多方法相呼应。我们对发送方缓冲的使用类似于 [14] 中的 “上游备份”，它也定义了恢复语义（精确、回滚和间隙），与我们自己灵活的数据交付选项类似。正如Spark Streaming [34] 中提到的，简单的上游备份解决方案可能会消耗过多资源，而我们对检查点和持久状态的使用消除了这些缺点。此外，我们的系统能够实现比Spark Streaming [34] 更细粒度的检查点，Spark Streaming建议每分钟进行一次备份，并依赖应用程序的幂等性和系统松弛来进行恢复。类似地，SGuard [17] 比MillWheel更少地使用检查点（同样是每分钟一次），尽管它的操作符分区方案类似于我们基于键的分片。

我们的低水位标记机制与其他流处理系统（如Gigascope [10]）中使用的标点 [31] 或心跳 [16] 类似。然而，在我们的系统中，我们不会将心跳与标准元组交错，而是选择像OOP系统 [19] 那样使用全局聚合器。我们的低水位标记概念也与OOP中定义的低水位标记类似。我们认同他们的分析，即在单个操作符上聚合心跳效率低下，最好留给全局机构处理。Srivastava等人在 [29] 中强调了这种低效率，其中讨论了在每个操作符上维护每个流的心跳数组。此外，它还在用户定义的时间戳（“应用时间”）和实际时钟时间（“系统时间”）的概念之间建立了类似的区别，我们发现这非常有用。

我们注意到Lamport [18] 以及其他人 [12] [15] 在为分布式系统开发引人注目的时间语义方面所做的工作。

流处理系统的许多灵感可以追溯到流数据库系统的开创性工作，如TelegraphCQ [6]、Aurora [2] 和STREAM [24]。我们观察到我们实现的组件与流SQL中的对应组件之间存在相似之处，例如在Flux [28] 中使用分区操作符进行负载均衡。虽然我们认为我们的低水位标记语义比 [2] 中的松弛语义更强大，但我们发现我们的百分位低水位标记概念与 [1] 中的QoS系统有一些相似之处。

10. 参考文献

[1] D. J. Abadi, Y. Ahmad, M. Balazinska, M. Cherniack,
J. hyon Hwang, W. Lindner, A. S. Maskey, E. Rasin,
E. Ryvkina, N. Tatbul, Y. Xing, and S. Zdonik. The design of
the borealis stream processing engine. In In CIDR, pages
277–289, 2005.
[2] D. J. Abadi, D. Carney, U. C¸ etintemel, M. Cherniack,
C. Convey, S. Lee, M. Stonebraker, N. Tatbul, and S. Zdonik.
Aurora: a new model and architecture for data stream
management. The VLDB Journal, 12(2):120–139, 2003.
[3] A. Adya, J. Dunagan, and A. Wolman. Centrifuge: Integrated
lease management and partitioning for cloud services. In
NSDI, pages 1–16. USENIX Association, 2010.
[4] Apache. Apache hadoop.
http://hadoop.apache.org, 2012.
[5] B. Babcock, S. Babu, M. Datar, R. Motwani, and J. Widom.
Models and issues in data stream systems. In Proceedings of
the twenty-first ACM SIGMOD-SIGACT-SIGART symposium
on Principles of database systems, pages 1–16. ACM, 2002.
[6] S. Chandrasekaran, O. Cooper, A. Deshpande, M. J.
Franklin, J. M. Hellerstein, W. Hong, S. Krishnamurthy,
S. R. Madden, F. Reiss, and M. A. Shah. Telegraphcq:
continuous dataflow processing. In Proceedings of the 2003
ACM SIGMOD international conference on Management of
data, pages 668–668. ACM, 2003.
[7] F. Chang, J. Dean, S. Ghemawat, W. C. Hsieh, D. A.
Wallach, M. Burrows, T. Chandra, A. Fikes, and R. E.
Gruber. Bigtable: A distributed storage system for structured
data. ACM Trans. Comput. Syst., 26:4:1–4:26, June 2008.
[8] T. Condie, N. Conway, P. Alvaro, J. M. Hellerstein,
K. Elmeleegy, and R. Sears. Mapreduce online. Technical
report, University of California, Berkeley, 2009.
[9] J. C. Corbett, J. Dean, M. Epstein, A. Fikes, C. Frost,
J. Furman, S. Ghemawat, A. Gubarev, C. Heiser,
P. Hochschild, et al. Spanner: Googles globally-distributed
database. To appear in Proceedings of OSDI, page 1, 2012.
[10] C. Cranor, Y. Gao, T. Johnson, V. Shkapenyuk, and
O. Spatscheck. Gigascope: High performance network
monitoring with an sql interface. In Proceedings of the 2002
ACM SIGMOD international conference on Management of
data, pages 623–623. ACM, 2002.
[11] J. Dean and S. Ghemawat. Mapreduce: simplified data
processing on large clusters. Commun. ACM, 51:107–113,
Jan. 2008.
[12] E. Deelman and B. K. Szymanski. Continuously monitored
global virtual time. Technical report, in Intern. Conf. Parallel
and Distributed Processing Techniques and Applications, Las
Vegas, NV, 1996.
[13] Google. Protocol buffers.
http://code.google.com/p/protobuf/, 2012.
[14] J.-H. Hwang, M. Balazinska, A. Rasin, U. Cetintemel,
M. Stonebraker, and S. Zdonik. High-availability algorithms
for distributed stream processing. In Data Engineering,
2005. ICDE 2005. Proceedings. 21st International
Conference on, pages 779–790. IEEE, 2005.
[15] D. R. Jefferson. Virtual time. ACM Transactions on
Programming Languages and Systems, 7:404–425, 1985.
[16] T. Johnson, S. Muthukrishnan, V. Shkapenyuk, and
O. Spatscheck. A heartbeat mechanism and its application in
gigascope. In Proceedings of the 31st international
conference on Very large data bases, pages 1079–1088.
VLDB Endowment, 2005.
[17] Y. Kwon, M. Balazinska, and A. Greenberg. Fault-tolerant
stream processing using a distributed, replicated file system.
Proceedings of the VLDB Endowment, 1(1):574–585, 2008.
[18] L. Lamport. Time, clocks, and the ordering of events in a
distributed system. Commun. ACM, 21(7):558–565, July
1978.
[19] J. Li, K. Tufte, V. Shkapenyuk, V. Papadimos, T. Johnson,
and D. Maier. Out-of-order processing: a new architecture
for high-performance stream systems. Proceedings of the
VLDB Endowment, 1(1):274–288, 2008.
[20] D. Logothetis, C. Olston, B. Reed, K. C. Webb, and
K. Yocum. Stateful bulk processing for incremental
analytics. In Proceedings of the 1st ACM symposium on
Cloud computing, pages 51–62. ACM, 2010.
[21] S. Madden and M. J. Franklin. Fjording the stream: An
architecture for queries over streaming sensor data. In Data
Engineering, 2002. Proceedings. 18th International
Conference on, pages 555–566. IEEE, 2002.
[22] N. Marz. Trident. https://github.com/
nathanmarz/storm/wiki/Trident-tutorial,
2012.
[23] N. Marz. Twitter storm.
https://github.com/nathanmarz/storm/wiki,
2012.
[24] R. Motwani, J. Widom, A. Arasu, B. Babcock, S. Babu,
M. Datar, G. Manku, C. Olston, J. Rosenstein, and R. Varma.
Query processing, resource management, and approximation
in a data stream management system. Technical Report
2002-41, Stanford InfoLab, 2002.
[25] D. G. Murray, M. Schwarzkopf, C. Smowton, S. Smith,
A. Madhavapeddy, and S. Hand. Ciel: a universal execution
engine for distributed data-flow computing. In Proceedings
of the 8th USENIX conference on Networked systems design
and implementation, page 9. USENIX Association, 2011.
[26] L. Neumeyer, B. Robbins, A. Nair, and A. Kesari. S4:
Distributed stream computing platform. In Data Mining
Workshops (ICDMW), 2010 IEEE International Conference
on, pages 170 –177, dec. 2010.
[27] D. Peng, F. Dabek, and G. Inc. Large-scale incremental
processing using distributed transactions and notifications. In
9th USENIX Symposium on Operating Systems Design and
Implementation, 2010.
[28] M. A. Shah, J. M. Hellerstein, S. Chandrasekaran, and M. J.
Franklin. Flux: An adaptive partitioning operator for
continuous query systems. In Data Engineering, 2003.
Proceedings. 19th International Conference on, pages
25–36. IEEE, 2003.
[29] U. Srivastava and J. Widom. Flexible time management in
data stream systems. In Proceedings of the twenty-third ACM
SIGMOD-SIGACT-SIGART symposium on Principles of
database systems, pages 263–274. ACM, 2004.
[30] M. Stonebraker, U. C¸ etintemel, and S. Zdonik. The 8
requirements of real-time stream processing. ACM SIGMOD
Record, 34(4):42–47, 2005.
[31] P. A. Tucker, D. Maier, T. Sheard, and L. Fegaras. Exploiting
punctuation semantics in continuous data streams.
Knowledge and Data Engineering, IEEE Transactions on,
15(3):555–568, 2003.
[32] F. Yang, Z. Qian, X. Chen, I. Beschastnikh, L. Zhuang,
L. Zhou, and J. Shen. Sonora: A platform for continuous
mobile-cloud computing. Technical report, Technical Report.
Microsoft Research Asia.