Question

到目前为止，本章中我们已经讨论了流的来源（用户活动事件，传感器和写入数据库），我们讨论了流如何传输（直接通过消息传送，通过消息代理，通过事件日志）。

剩下的就是讨论一下你可以用流做什么 —— 也就是说，你可以处理它。一般来说，有三种选项：

1. 你可以将事件中的数据写入数据库，缓存，搜索索引或类似的存储系统，然后能被其他客户端查询。如 图 11-5 所示，这是数据库与系统其他部分发生变更保持同步的好方法 —— 特别是当流消费者是写入数据库的唯一客户端时。如 批处理工作流的输出 中所讨论的，它是写入存储系统的流等价物。
2. 你能以某种方式将事件推送给用户，例如发送报警邮件或推送通知，或将事件流式传输到可实时显示的仪表板上。在这种情况下，人是流的最终消费者。
3. 你可以处理一个或多个输入流，并产生一个或多个输出流。流可能会经过由几个这样的处理阶段组成的流水线，最后再输出（选项 1 或 2）。

在本章的剩余部分中，我们将讨论选项 3：处理流以产生其他衍生流。处理这样的流的代码片段，被称为 算子（operator） 或 作业（job） 。它与我们在 第 10 章 中讨论过的 Unix 进程和 MapReduce 作业密切相关，数据流的模式是相似的：一个流处理器以只读的方式使用输入流，并将其输出以仅追加的方式写入一个不同的位置。

​ 流处理中的分区和并行化模式也非常类似于 第 10 章 中介绍的 MapReduce 和数据流引擎，因此我们不再重复这些主题。基本的 Map 操作（如转换和过滤记录）也是一样的。

​ 与批量作业相比的一个关键区别是，流不会结束。这种差异会带来很多隐含的结果。正如本章开始部分所讨论的，排序对无界数据集没有意义，因此无法使用 排序合并联接 （请参阅 Reduce 端连接与分组 ）。容错机制也必须改变：对于已经运行了几分钟的批处理作业，可以简单地从头开始重启失败任务，但是对于已经运行数年的流作业，重启后从头开始跑可能并不是一个可行的选项。

流处理的应用

长期以来，流处理一直用于监控目的，如果某个事件发生，单位希望能得到警报。例如：

• 欺诈检测系统需要确定信用卡的使用模式是否有意外地变化，如果信用卡可能已被盗刷，则锁卡。
• 交易系统需要检查金融市场的价格变化，并根据指定的规则进行交易。
• 制造系统需要监控工厂中机器的状态，如果出现故障，可以快速定位问题。
• 军事和情报系统需要跟踪潜在侵略者的活动，并在出现袭击征兆时发出警报。

这些类型的应用需要非常精密复杂的模式匹配与相关检测。然而随着时代的进步，流处理的其他用途也开始出现。在本节中，我们将简要比较一下这些应用。

复合事件处理

​ 复合事件处理（complex, event processing, CEP） 是 20 世纪 90 年代为分析事件流而开发出的一种方法，尤其适用于需要搜索某些事件模式的应用【65,66】。与正则表达式允许你在字符串中搜索特定字符模式的方式类似，CEP 允许你指定规则以在流中搜索某些事件模式。

​ CEP 系统通常使用高层次的声明式查询语言，比如 SQL，或者图形用户界面，来描述应该检测到的事件模式。这些查询被提交给处理引擎，该引擎消费输入流，并在内部维护一个执行所需匹配的状态机。当发现匹配时，引擎发出一个 复合事件（complex event） （因此得名），并附有检测到的事件模式详情【67】。

​ 在这些系统中，查询和数据之间的关系与普通数据库相比是颠倒的。通常情况下，数据库会持久存储数据，并将查询视为临时的：当查询进入时，数据库搜索与查询匹配的数据，然后在查询完成时丢掉查询。 CEP 引擎反转了角色：查询是长期存储的，来自输入流的事件不断流过它们，搜索匹配事件模式的查询【68】。

​ CEP 的实现包括 Esper 【69】，IBM InfoSphere Streams 【70】，Apama，TIBCO StreamBase 和 SQLstream。像 Samza 这样的分布式流处理组件，支持使用 SQL 在流上进行声明式查询【71】。

流分析

​ 使用流处理的另一个领域是对流进行分析。 CEP 与流分析之间的边界是模糊的，但一般来说，分析往往对找出特定事件序列并不关心，而更关注大量事件上的聚合与统计指标 —— 例如：

• 测量某种类型事件的速率（每个时间间隔内发生的频率）
• 滚动计算一段时间窗口内某个值的平均值
• 将当前的统计值与先前的时间区间的值对比（例如，检测趋势，当指标与上周同比异常偏高或偏低时报警）

这些统计值通常是在固定时间区间内进行计算的，例如，你可能想知道在过去 5 分钟内服务每秒查询次数的均值，以及此时间段内响应时间的第 99 百分位点。在几分钟内取平均，能抹平秒和秒之间的无关波动，且仍然能向你展示流量模式的时间图景。聚合的时间间隔称为 窗口（window） ，我们将在 理解时间 中更详细地讨论窗口。

​ 流分析系统有时会使用概率算法，例如 Bloom filter（我们在 性能优化 中遇到过）来管理成员资格，HyperLogLog 【72】用于基数估计以及各种百分比估计算法（请参阅 实践中的百分位点 ）。概率算法产出近似的结果，但比起精确算法的优点是内存使用要少得多。使用近似算法有时让人们觉得流处理系统总是有损的和不精确的，但这是错误看法：流处理并没有任何内在的近似性，而概率算法只是一种优化【73】。

​ 许多开源分布式流处理框架的设计都是针对分析设计的：例如 Apache Storm，Spark Streaming，Flink，Concord，Samza 和 Kafka Streams 【74】。托管服务包括 Google Cloud Dataflow 和 Azure Stream Analytics。

维护物化视图

​ 我们在 数据库和数据流 中看到，数据库的变更流可以用于维护衍生数据系统（如缓存，搜索索引和数据仓库），使其与源数据库保持最新。我们可以将这些示例视作维护 物化视图（materialized view） 的一种具体场景（参阅 聚合：数据立方体和物化视图 ）：在某个数据集上衍生出一个替代视图以便高效查询，并在底层数据变更时更新视图【50】。

​ 同样，在事件溯源中，应用程序的状态是通过 应用（apply） 事件日志来维护的；这里的应用状态也是一种物化视图。与流分析场景不同的是，仅考虑某个时间窗口内的事件通常是不够的：构建物化视图可能需要任意时间段内的 所有 事件，除了那些可能由日志压缩丢弃的过时事件（请参阅 日志压缩 ）。实际上，你需要一个可以一直延伸到时间开端的窗口。

​ 原则上讲，任何流处理组件都可以用于维护物化视图，尽管 永远运行 与一些面向分析的框架假设的 主要在有限时间段窗口上运行 背道而驰， Samza 和 Kafka Streams 支持这种用法，建立在 Kafka 对日志压缩 comp 的支持上【75】。

在流上搜索

​ 除了允许搜索由多个事件构成模式的 CEP 外，有时也存在基于复杂标准（例如全文搜索查询）来搜索单个事件的需求。

​ 例如，媒体监测服务可以订阅新闻文章 Feed 与来自媒体的播客，搜索任何关于公司，产品或感兴趣的话题的新闻。这是通过预先构建一个搜索查询来完成的，然后不断地将新闻项的流与该查询进行匹配。在一些网站上也有类似的功能：例如，当市场上出现符合其搜索条件的新房产时，房地产网站的用户可以要求网站通知他们。 Elasticsearch 的这种过滤器功能，是实现这种流搜索的一种选择【76】。

​ 传统的搜索引擎首先索引文件，然后在索引上跑查询。相比之下，搜索一个数据流则反了过来：查询被存储下来，文档从查询中流过，就像在 CEP 中一样。在简单的情况就是，你可以为每个文档测试每个查询。但是如果你有大量查询，这可能会变慢。为了优化这个过程，可以像对文档一样，为查询建立索引。因而收窄可能匹配的查询集合【77】。

消息传递和 RPC

​ 在 消息传递数据流 中我们讨论过，消息传递系统可以作为 RPC 的替代方案，即作为一种服务间通信的机制，比如在 Actor 模型中所使用的那样。尽管这些系统也是基于消息和事件，但我们通常不会将其视作流处理组件：

• Actor 框架主要是管理模块通信的并发和分布式执行的一种机制，而流处理主要是一种数据管理技术。

• Actor 之间的交流往往是短暂的，一对一的；而事件日志则是持久的，多订阅者的。
• Actor 可以以任意方式进行通信（允许包括循环的请求/响应），但流处理通常配置在无环流水线中，其中每个流都是一个特定作业的输出，由良好定义的输入流中派生而来。

也就是说，RPC 类系统与流处理之间有一些交叉领域。例如，Apache Storm 有一个称为 分布式 RPC 的功能，它允许将用户查询分散到一系列也处理事件流的节点上；然后这些查询与来自输入流的事件交织，而结果可以被汇总并发回给用户【78】（另参阅 多分区数据处理 ）。

​ 也可以使用 Actor 框架来处理流。但是，很多这样的框架在崩溃时不能保证消息的传递，除非你实现了额外的重试逻辑，否则这种处理不是容错的。

时间推理

​ 流处理通常需要与时间打交道，尤其是用于分析目的时候，会频繁使用时间窗口，例如 过去五分钟的平均值 。 最后五分钟 的含义看上去似乎是清晰而无歧义的，但不幸的是，这个概念非常棘手。

​ 在批处理中过程中，大量的历史事件迅速收缩。如果需要按时间来分析，批处理器需要检查每个事件中嵌入的时间戳。读取运行批处理机器的系统时钟没有任何意义，因为处理运行的时间与事件实际发生的时间无关。

​ 批处理可以在几分钟内读取一年的历史事件；在大多数情况下，感兴趣的时间线是历史中的一年，而不是处理中的几分钟。而且使用事件中的时间戳，使得处理是 确定性 的：在相同的输入上再次运行相同的处理过程会得到相同的结果（参阅 故障容错 ）。

​ 另一方面，许多流处理框架使用处理机器上的本地系统时钟（ 处理时间（processing time） ）来确定 窗口 【79】。这种方法的优点是简单，事件创建与事件处理之间的延迟可以忽略不计。然而，如果存在任何显著的处理延迟 —— 即，事件处理显著地晚于事件实际发生的时间，处理就失效了。

事件时间与处理时间

​ 很多原因都可能导致处理延迟：排队，网络故障（参阅 不可靠的网络 ），性能问题导致消息代理/消息处理器出现争用，流消费者重启，重新处理过去的事件（参阅 重放旧消息 ），或者在修复代码 BUG 之后从故障中恢复。

​ 而且，消息延迟还可能导致无法预测消息顺序。例如，假设用户首先发出一个 Web 请求（由 Web 服务器 A 处理），然后发出第二个请求（由服务器 B 处理）。 A 和 B 发出描述它们所处理请求的事件，但是 B 的事件在 A 的事件发生之前到达消息代理。现在，流处理器将首先看到 B 事件，然后看到 A 事件，即使它们实际上是以相反的顺序发生的。

​ 有一个类比也许能帮助理解， 星球大战 电影：第四集于 1977 年发行，第五集于 1980 年，第六集于 1983 年，紧随其后的是 1999 年的第一集，2002 年的第二集，和 2005 年的三集，以及 2015 年的第七集【80】ⁱⁱ 。如果你按照按照它们上映的顺序观看电影，你处理电影的顺序与它们叙事的顺序就是不一致的。 （集数编号就像事件时间戳，而你观看电影的日期就是处理时间）作为人类，我们能够应对这种不连续性，但是流处理算法需要专门写就，以适应这种时机与顺序的问题。

ⁱⁱ. 感谢 Flink 社区的 Kostas Kloudas 提出这个比喻。 ↩

​ 将事件时间和处理时间搞混会导致错误的数据。例如，假设你有一个流处理器用于测量请求速率（计算每秒请求数）。如果你重新部署流处理器，它可能会停止一分钟，并在恢复之后处理积压的事件。如果你按处理时间来衡量速率，那么在处理积压日志时，请求速率看上去就像有一个异常的突发尖峰，而实际上请求速率是稳定的（ 图 11-7 ）。

图 11-7 按处理时间分窗，会因为处理速率的变动引入人为因素

知道什么时候准备好了

​ 用事件时间来定义窗口的一个棘手的问题是，你永远也无法确定是不是已经收到了特定窗口的所有事件，还是说还有一些事件正在来的路上。

​ 例如，假设你将事件分组为一分钟的窗口，以便统计每分钟的请求数。你已经计数了一些带有本小时内第 37 分钟时间戳的事件，时间流逝，现在进入的主要都是本小时内第 38 和第 39 分钟的事件。什么时候才能宣布你已经完成了第 37 分钟的窗口计数，并输出其计数器值？

​ 在一段时间没有看到任何新的事件之后，你可以超时并宣布一个窗口已经就绪，但仍然可能发生这种情况：某些事件被缓冲在另一台机器上，由于网络中断而延迟。你需要能够处理这种在窗口宣告完成之后到达的 滞留（straggler） 事件。大体上，你有两种选择【1】：

1. 忽略这些滞留事件，因为在正常情况下它们可能只是事件中的一小部分。你可以将丢弃事件的数量作为一个监控指标，并在出现大量丢消息的情况时报警。
2. 发布一个 更正（correction） ，一个包括滞留事件的更新窗口值。更新的窗口与包含散兵队员的价值。你可能还需要收回以前的输出。

在某些情况下，可以使用特殊的消息来指示 从现在开始，不会有比 t 更早时间戳的消息了 ，消费者可以使用它来触发窗口【81】。但是，如果不同机器上的多个生产者都在生成事件，每个生产者都有自己的最小时间戳阈值，则消费者需要分别跟踪每个生产者。在这种情况下，添加和删除生产者都是比较棘手的。

你用的是谁的时钟？

​ 当事件可能在系统内多个地方进行缓冲时，为事件分配时间戳更加困难了。例如，考虑一个移动应用向服务器上报关于用量的事件。该应用可能会在设备处于脱机状态时被使用，在这种情况下，它将在设备本地缓冲事件，并在下一次互联网连接可用时向服务器上报这些事件（可能是几小时甚至几天）。对于这个流的任意消费者而言，它们就如延迟极大的滞留事件一样。

​ 在这种情况下，事件上的事件戳实际上应当是用户交互发生的时间，取决于移动设备的本地时钟。然而用户控制的设备上的时钟通常是不可信的，因为它可能会被无意或故意设置成错误的时间（参见 时钟同步与准确性 ）。服务器收到事件的时间（取决于服务器的时钟）可能是更准确的，因为服务器在你的控制之下，但在描述用户交互方面意义不大。

要校正不正确的设备时钟，一种方法是记录三个时间戳【82】：

• 事件发生的时间，取决于设备时钟
• 事件发送往服务器的时间，取决于设备时钟
• 事件被服务器接收的时间，取决于服务器时钟

通过从第三个时间戳中减去第二个时间戳，可以估算设备时钟和服务器时钟之间的偏移（假设网络延迟与所需的时间戳精度相比可忽略不计）。然后可以将该偏移应用于事件时间戳，从而估计事件实际发生的真实时间（假设设备时钟偏移在事件发生时与送往服务器之间没有变化）。

​ 这并不是流处理独有的问题，批处理有着完全一样的时间推理问题。只是在流处理的上下文中，我们更容易意识到时间的流逝。

窗口的类型

​ 当你知道如何确定一个事件的时间戳后，下一步就是如何定义时间段的窗口。然后窗口就可以用于聚合，例如事件计数，或计算窗口内值的平均值。有几种窗口很常用【79,83】：

滚动窗口（Tumbling Window）

​ 滚动窗口有着固定的长度，每个事件都仅能属于一个窗口。例如，假设你有一个 1 分钟的滚动窗口，则所有时间戳在 10:03:00 和 10:03:59 之间的事件会被分组到一个窗口中， 10:04:00 和 10:04:59 之间的事件被分组到下一个窗口，依此类推。通过将每个事件时间戳四舍五入至最近的分钟来确定它所属的窗口，可以实现 1 分钟的滚动窗口。

跳动窗口（Hopping Window）

​ 跳动窗口也有着固定的长度，但允许窗口重叠以提供一些平滑。例如，一个带有 1 分钟跳跃步长的 5 分钟窗口将包含 10:03:00 至 10:07:59 之间的事件，而下一个窗口将覆盖 10:04:00 至 10:08 之间的事件： 59，等等。通过首先计算 1 分钟的滚动窗口，然后在几个相邻窗口上进行聚合，可以实现这种跳动窗口。

滑动窗口（Sliding Window）

​ 滑动窗口包含了彼此间距在特定时长内的所有事件。例如，一个 5 分钟的滑动窗口应当覆盖 10:03:39 和 10:08:12 的事件，因为它们相距不超过 5 分钟（注意滚动窗口与步长 5 分钟的跳动窗口可能不会把这两个事件分组到同一个窗口中，因为它们使用固定的边界）。通过维护一个按时间排序的事件缓冲区，并不断从窗口中移除过期的旧事件，可以实现滑动窗口。

会话窗口（Session window）

​ 与其他窗口类型不同，会话窗口没有固定的持续时间，而定义为：将同一用户出现时间相近的所有事件分组在一起，而当用户一段时间没有活动时（例如，如果 30 分钟内没有事件）窗口结束。会话切分是网站分析的常见需求（参阅 GROUP BY ）。

流式连接

​ 在 第 10 章 中，我们讨论了批处理作业如何通过键来连接数据集，以及这种连接是如何成为数据管道的重要组成部分的。由于流处理将数据管道泛化为对无限数据集进行增量处理，因此对流进行连接的需求也是完全相同的。

​ 然而，新事件随时可能出现在一个流中，这使得流连接要比批处理连接更具挑战性。为了更好地理解情况，让我们先来区分三种不同类型的连接： 流-流 连接， 流-表 连接，与 表-表 连接【84】。我们将在下面的章节中通过例子来说明。

流流连接（窗口连接）

​ 假设你的网站上有搜索功能，而你想要找出搜索 URL 的近期趋势。每当有人键入搜索查询时，都会记录下一个包含查询与其返回结果的事件。每当有人点击其中一个搜索结果时，就会记录另一个记录点击事件。为了计算搜索结果中每个 URL 的点击率，你需要将搜索动作与点击动作的事件连在一起，这些事件通过相同的会话 ID 进行连接。广告系统中需要类似的分析【85】。

​ 如果用户丢弃了搜索结果，点击可能永远不会发生，即使它出现了，搜索与点击之间的时间可能是高度可变的：在很多情况下，它可能是几秒钟，但也可能长达几天或几周（如果用户执行搜索，忘掉了这个浏览器页面，过了一段时间后重新回到这个浏览器页面上，并点击了一个结果）。由于可变的网络延迟，点击事件甚至可能先于搜索事件到达。你可以选择合适的连接窗口 —— 例如，如果点击与搜索之间的时间间隔在一小时内，你可能会选择连接两者。

​ 请注意，在点击事件中嵌入搜索详情与事件连接并不一样：这样做的话，只有当用户点击了一个搜索结果时你才能知道，而那些没有点击的搜索就无能为力了。为了衡量搜索质量，你需要准确的点击率，为此搜索事件和点击事件两者都是必要的。

​ 为了实现这种类型的连接，流处理器需要维护 状态 ：例如，按会话 ID 索引最近一小时内发生的所有事件。无论何时发生搜索事件或点击事件，都会被添加到合适的索引中，而流处理器也会检查另一个索引是否有具有相同会话 ID 的事件到达。如果有匹配事件就会发出一个表示搜索结果被点击的事件；如果搜索事件直到过期都没看见有匹配的点击事件，就会发出一个表示搜索结果未被点击的事件。

流表连接（流扩展）

​ 在 示例：用户活动事件分析 （ 图 10-2 ）中，我们看到了连接两个数据集的批处理作业示例：一组用户活动事件和一个用户档案数据库。将用户活动事件视为流，并在流处理器中连续执行相同的连接是很自然的想法：输入是包含用户 ID 的活动事件流，而输出还是活动事件流，但其中用户 ID 已经被扩展为用户的档案信息。这个过程有时被称为 使用数据库的信息来 扩充（enriching） 活动事件。

​ 要执行此联接，流处理器需要一次处理一个活动事件，在数据库中查找事件的用户 ID，并将档案信息添加到活动事件中。数据库查询可以通过查询远程数据库来实现。但正如在 示例：分析用户活动事件 一节中讨论的，此类远程查询可能会很慢，并且有可能导致数据库过载【75】。

​ 另一种方法是将数据库副本加载到流处理器中，以便在本地进行查询而无需网络往返。这种技术与我们在 Map 端连接 中讨论的散列连接非常相似：如果数据库的本地副本足够小，则可以是内存中的散列表，比较大的话也可以是本地磁盘上的索引。

​ 与批处理作业的区别在于，批处理作业使用数据库的时间点快照作为输入，而流处理器是长时间运行的，且数据库的内容可能随时间而改变，所以流处理器数据库的本地副本需要保持更新。这个问题可以通过变更数据捕获来解决：流处理器可以订阅用户档案数据库的更新日志，如同活跃事件流一样。当增添或修改档案时，流处理器会更新其本地副本。因此，我们有了两个流之间的连接：活动事件和档案更新。

​ 流表连接实际上非常类似于流流连接；最大的区别在于对于表的变更日志流，连接使用了一个可以回溯到 时间起点 的窗口（概念上是无限的窗口），新版本的记录会覆盖更早的版本。对于输入的流，连接可能压根儿就没有维护窗口。

表表连接（维护物化视图）

​ 我们在 描述负载 中讨论的推特时间线例子时说过，当用户想要查看他们的主页时间线时，迭代用户所关注人群的推文并合并它们是一个开销巨大的操作。

​ 相反，我们需要一个时间线缓存：一种每个用户的 收件箱 ，在发送推文的时候写入这些信息，因而读取时间线时只需要简单地查询即可。物化与维护这个缓存需要处理以下事件：

• 当用户 u 发送新的推文时，它将被添加到每个关注用户 u 的时间线上。
• 用户删除推文时，推文将从所有用户的时间表中删除。
• 当用户$u_1$开始关注用户$u_2$时，$u_2$最近的推文将被添加到$u_1$的时间线上。
• 当用户$u_1$取消关注用户$u_2$时，$u_2$的推文将从$u_1$的时间线中移除。

要在流处理器中实现这种缓存维护，你需要推文事件流（发送与删除）和关注关系事件流（关注与取消关注）。流处理需要为维护一个数据库，包含每个用户的粉丝集合。以便知道当一条新推文到达时，需要更新哪些时间线【86】。

观察这个流处理过程的另一种视角是：它维护了一个连接了两个表（推文与关注）的物化视图，如下所示：

SELECT follows.follower_id AS timeline_id, 
    array_agg(tweets.* ORDER BY tweets.timestamp DESC)
FROM tweets
JOIN follows ON follows.followee_id = tweets.sender_id 
GROUP BY follows.follower_id

​ 流连接直接对应于这个查询中的表连接。时间线实际上是这个查询结果的缓存，每当基础表发生变化时都会更新ⁱⁱⁱ 。

ⁱⁱⁱ. 如果你将流视作表的衍生物，如 图 11-6 所示，而把一个连接看作是两个表的乘法 u·v，那么会发生一些有趣的事情：物化连接的变化流遵循乘积法则：(u·v)'= u'v + uv'(u·v)'= u'v + uv'。 换句话说，任何推文的变化量都与当前的关注联系在一起，任何关注的变化量都与当前的推文相连接【49,50】。 ↩

连接的时间依赖性

​ 这里描述的三种连接（流流，流表，表表）有很多共通之处：它们都需要流处理器维护连接一侧的一些状态（搜索与点击事件，用户档案，关注列表），然后当连接另一侧的消息到达时查询该状态。

​ 用于维护状态的事件顺序是很重要的（先关注然后取消关注，或者其他类似操作）。在分区日志中，单个分区内的事件顺序是保留下来的。但典型情况下是没有跨流或跨分区的顺序保证的。

​ 这就产生了一个问题：如果不同流中的事件发生在近似的时间范围内，则应该按照什么样的顺序进行处理？在流表连接的例子中，如果用户更新了它们的档案，哪些活动事件与旧档案连接（在档案更新前处理），哪些又与新档案连接（在档案更新之后处理）？换句话说：你需要对一些状态做连接，如果状态会随着时间推移而变化，那应当使用什么时间点来连接呢【45】？

​ 这种时序依赖可能出现在很多地方。例如销售东西需要对发票应用适当的税率，这取决于所处的国家/州，产品类型，销售日期（因为税率会随时变化）。当连接销售额与税率表时，你可能期望的是使用销售时的税率参与连接。如果你正在重新处理历史数据，销售时的税率可能和现在的税率有所不同。

​ 如果跨越流的事件顺序是未定的，则连接会变为不确定性的【87】，这意味着你在同样输入上重跑相同的作业未必会得到相同的结果：当你重跑任务时，输入流上的事件可能会以不同的方式交织。

​ 在数据仓库中，这个问题被称为 缓慢变化的维度（slowly changing dimension, SCD） ，通常通过对特定版本的记录使用唯一的标识符来解决：例如，每当税率改变时都会获得一个新的标识符，而发票在销售时会带有税率的标识符【88,89】。这种变化使连接变为确定性的，但也会导致日志压缩无法进行：表中所有的记录版本都需要保留。

容错

​ 在本章的最后一节中，让我们看一看流处理是如何容错的。我们在 第 10 章 中看到，批处理框架可以很容易地容错：如果 MapReduce 作业中的任务失败，可以简单地在另一台机器上再次启动，并且丢弃失败任务的输出。这种透明的重试是可能的，因为输入文件是不可变的，每个任务都将其输出写入到 HDFS 上的独立文件中，而输出仅当任务成功完成后可见。

​ 特别是，批处理容错方法可确保批处理作业的输出与没有出错的情况相同，即使实际上某些任务失败了。看起来好像每条输入记录都被处理了恰好一次 —— 没有记录被跳过，而且没有记录被处理两次。尽管重启任务意味着实际上可能会多次处理记录，但输出中的可见效果看上去就像只处理过一次。这个原则被称为 恰好一次语义（exactly-once semantics） ，尽管 有效一次（effectively-once） 可能会是一个更写实的术语【90】。

​ 在流处理中也出现了同样的容错问题，但是处理起来没有那么直观：等待某个任务完成之后再使其输出可见并不是一个可行选项，因为你永远无法处理完一个无限的流。

微批量与存档点

​ 一个解决方案是将流分解成小块，并像微型批处理一样处理每个块。这种方法被称为 微批次（microbatching） ，它被用于 Spark Streaming 【91】。批次的大小通常约为 1 秒，这是对性能妥协的结果：较小的批次会导致更大的调度与协调开销，而较大的批次意味着流处理器结果可见之前的延迟要更长。

​ 微批次也隐式提供了一个与批次大小相等的滚动窗口（按处理时间而不是事件时间戳分窗）。任何需要更大窗口的作业都需要显式地将状态从一个微批次转移到下一个微批次。

​ Apache Flink 则使用不同的方法，它会定期生成状态的滚动存档点并将其写入持久存储【92,93】。如果流算子崩溃，它可以从最近的存档点重启，并丢弃从最近检查点到崩溃之间的所有输出。存档点会由消息流中的 壁障（barrier） 触发，类似于微批次之间的边界，但不会强制一个特定的窗口大小。

​ 在流处理框架的范围内，微批次与存档点方法提供了与批处理一样的 恰好一次语义 。但是，只要输出离开流处理器（例如，写入数据库，向外部消息代理发送消息，或发送电子邮件），框架就无法抛弃失败批次的输出了。在这种情况下，重启失败任务会导致外部副作用发生两次，只有微批次或存档点不足以阻止这一问题。

原子提交再现

​ 为了在出现故障时表现出恰好处理一次的样子，我们需要确保事件处理的所有输出和副作用 当且仅当 处理成功时才会生效。这些影响包括发送给下游算子或外部消息传递系统（包括电子邮件或推送通知）的任何消息，任何数据库写入，对算子状态的任何变更，以及对输入消息的任何确认（包括在基于日志的消息代理中将消费者偏移量前移）。

​ 这些事情要么都原子地发生，要么都不发生，但是它们不应当失去同步。如果这种方法听起来很熟悉，那是因为我们在分布式事务和两阶段提交的上下文中讨论过它（参阅 恰好一次的消息处理 ）。

​ 在 第 9 章 中，我们讨论了分布式事务传统实现中的问题（如 XA）。然而在限制更为严苛的环境中，也是有可能高效实现这种原子提交机制的。 Google Cloud Dataflow【81,92】和 VoltDB 【94】中使用了这种方法，Apache Kafka 有计划加入类似的功能【95,96】。与 XA 不同，这些实现不会尝试跨异构技术提供事务，而是通过在流处理框架中同时管理状态变更与消息传递来内化事务。事务协议的开销可以通过在单个事务中处理多个输入消息来分摊。

幂等性

​ 我们的目标是丢弃任何失败任务的部分输出，以便能安全地重试，而不会生效两次。分布式事务是实现这个目标的一种方式，而另一种方式是依赖 幂等性（idempotence） 【97】。

​ 幂等操作是多次重复执行与单次执行效果相同的操作。例如，将键值存储中的某个键设置为某个特定值是幂等的（再次写入该值，只是用同样的值替代），而递增一个计数器不是幂等的（再次执行递增意味着该值递增两次）。

​ 即使一个操作不是天生幂等的，往往可以通过一些额外的元数据做成幂等的。例如，在使用来自 Kafka 的消息时，每条消息都有一个持久的，单调递增的偏移量。将值写入外部数据库时可以将这个偏移量带上，这样你就可以判断一条更新是不是已经执行过了，因而避免重复执行。

​ Storm 的 Trident 基于类似的想法来处理状态【78】。依赖幂等性意味着隐含了一些假设：重启一个失败的任务必须以相同的顺序重放相同的消息（基于日志的消息代理能做这些事），处理必须是确定性的，没有其他节点能同时更新相同的值【98,99】。

​ 当从一个处理节点故障转移到另一个节点时，可能需要进行 防护（fencing） （参阅 领导和锁 ），以防止被假死节点干扰。尽管有这么多注意事项，幂等操作是一种实现 恰好一次语义 的有效方式，仅需很小的额外开销。

失败后重建状态

​ 任何需要状态的流处理 —— 例如，任何窗口聚合（例如计数器，平均值和直方图）以及任何用于连接的表和索引，都必须确保在失败之后能恢复其状态。

​ 一种选择是将状态保存在远程数据存储中，并进行复制，然而正如在 流表连接 中所述，每个消息都要查询远程数据库可能会很慢。另一种方法是在流处理器本地保存状态，并定期复制。然后当流处理器从故障中恢复时，新任务可以读取状态副本，恢复处理而不丢失数据。

​ 例如，Flink 定期捕获算子状态的快照，并将它们写入 HDFS 等持久存储中【92,93】。 Samza 和 Kafka Streams 通过将状态变更发送到具有日志压缩功能的专用 Kafka 主题来复制状态变更，这与变更数据捕获类似【84,100】。 VoltDB 通过在多个节点上对每个输入消息进行冗余处理来复制状态（参阅 真的串行执行 ）。

​ 在某些情况下，甚至可能都不需要复制状态，因为它可以从输入流重建。例如，如果状态是从相当短的窗口中聚合而成，则简单地重放该窗口中的输入事件可能是足够快的。如果状态是通过变更数据捕获来维护的数据库的本地副本，那么也可以从日志压缩的变更流中重建数据库（参阅 日志压缩 ）。

​ 然而，所有这些权衡取决于底层基础架构的性能特征：在某些系统中，网络延迟可能低于磁盘访问延迟，网络带宽可能与磁盘带宽相当。没有针对所有情况的普世理想权衡，随着存储和网络技术的发展，本地状态与远程状态的优点也可能会互换。