第二章 -- 流处理基本概念

写在前面：

大家好，我是强哥，一个热爱分享的技术狂。目前已有 12 年大数据与 AI 相关项目经验， 10 年推荐系统研究及实践经验。平时喜欢读书、暴走和写作。

业余时间专注于输出大数据、AI 等相关文章，目前已经输出了 40 万字的推荐系统系列精品文章，强哥的畅销书「构建企业级推荐系统：算法、工程实现与案例分析」已经出版，需要提升可以私信我呀。如果这些文章能够帮助你快速入门，实现职场升职加薪，我将不胜欢喜。

想要获得更多免费学习资料或内推信息，一定要看到文章最后喔。

内推信息

如果你正在看相关的招聘信息，请加我微信：liuq4360，我这里有很多内推资源等着你，欢迎投递简历。

免费学习资料

如果你想获得更多免费的学习资料，请关注同名公众号【数据与智能】，输入“资料”即可！

学习交流群

如果你想找到组织，和大家一起学习成长，交流经验，也可以加入我们的学习成长群。群里有老司机带你飞，另有小哥哥、小姐姐等你来勾搭！加小姐姐微信：epsila，她会带你入群。

第二章--流处理基本概念学完第一章的内容，你应该了解了流处理引擎是如何解决传统批处理存在的瓶颈，以及如何在流处理引擎架构中运行应用程序，知晓了开源分布式流处理引擎一步步演进的历程，通过本章内容的学习，你将对流处理引擎有更深的认知。

本章的目的是介绍流处理的基本概念及其框架与组件，我们希望你在阅读完本章之后，能够充分理解主流的流处理引擎的相关特性。

数据流编程简介

在我们开始介绍流处理的基本概念之前，我们先了解一下数据流编程（Dataflow Programming）的背景，这是一个非常重要的概念，会贯穿整本书。

数据流图（Dataflow Graphs）

顾名思义，Dataflow Graph 描述了数据流程序算子之间的关系图，类似于 spark 的 DAG 有向无环图，数据流图通常表现为有向图，由节点（nodes）和边（edge）构成，也有人叫结点或者顶点，本质上是一个意思，节点表示逻辑算子，边则代表算子之间的依赖关系，其中，算子是数据流应用程序计算的基本单元，可以通过消费输入的数据进行一系列逻辑计算，不仅如此，还可以对计算后的数据做进一步处理。一个完整的数据流图包括数据源（source）、算子（operation）和数据接收端（sink）这几个要素，最基本的数据流图必须要包含一个数据源和一个数据接收端，图 2-1 展示是数据流程序对数据源 Tweets source 做标签提取转换和计数的流程。

图 2-1 描绘的 Dataflow Graph 实际上是一个计算过程的逻辑图，清晰分解了执行过程的细节，在真正执行计算的时候，流处理引擎会将逻辑视图的操作转换为物理执行计划，包括了整个流应用程序执行的详细步骤，在实际进行分布式处理时，每个算子可能会在不同机器上运行多个并行计算任务。图 2-2 的物理执行计划对应的是图 2-1 的计算逻辑，其中“Extract hashtags”和“Count”都指定了两个并行度，每个并行任务负责计算输入流的一部分数据，最终才进行结果汇总。

数据并行度和任务并行度

在 Dataflow Graphs 中，你可以通过不同的方式去指定程序的并行度。第一种方式是针对数据本身，你可以通过对输入数据进行分区，然后针对不同的分区执行相同的处理逻辑，这种方式称为数据并行度，提高数据并行度有很多的好处，可以有效实现数据负载均衡，将数据分散到多个计算节点上去，避免数据集中在一个节点上，有利于充分利用计算资源和降低计算节点负载。第二种方式是针对算子操作，通过指定算子的并行度，不同的算子可以对相同或者不同的数据进行计算，这种方式我们称之为任务并行度，并行度类似并发，通过提高任务并行度可以更好地利用集群资源，更加高效地获取最终结果。

数据交换策略

数据交换策略是用来体现在数据流的物理执行计划中数据是如何分发到不同的 task 上的，task 是计算的基本单元，下面简单介绍一下几种常见的交换策略，如图 2-3 所示。

前向策略（Forward）：将数据从上游的 task 发送到下游的 task。如果恰巧两个 task 均在一个机器上，则可以避免网络传输，减少 IO 损耗。

广播策略（Broadcast）：数据发送到所有并行 task 中。广播策略涉及到数据复制以及网络传输，所以较为消耗 IO 资源和内存。

Key-based 策略：根据 key 做分区，使具有相同 key 的条目可以被同一个 task 处理。

随机策略（Random）：将数据随机均匀分布到 task 中，以此均衡集群计算负载。

并行流处理

在我们了解完以上关于数据流编程的基本概念之后，接下来我们来看一下这些概念是如何应用到并行数据流上的，数据可以被分为无界数据流和有界数据流。

1. 无界数据流: 有始无终，有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。无界流的数据必须持续处理，意思是数据被摄取后需要快递处理。我们不能等到所有数据都到达再处理，因为这种情况下输入流是无穷无尽的，在任何时候输入都不会结束。处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性，无界流处理通常代表实时计算。

2. 有界数据流: 有始有终，有定义流的开始，也有定义流的结束。有界流可以在摄取所有数据后再进行计算，有界流的所有数据可以被排序，所以并不需要按顺序进行摄取。有界流处理通常被称为批处理。

任何类型的数据都可以形成一种事件流。如信用卡交易、传感器测量、机器日志、网站或移动应用程序上的用户交互记录，所有这些数据都可以形成数据流。在本章的内容中，你将学习到如何通过数据流编程对无界流进行并行处理。

延迟和吞吐量

在第一章的学习中，相信你已经知道了流式计算和批处理在应用场景和底层 API 实现上存在着很大的差别，包括对实时性和性能的要求也各不相同，我们一般关注的是一个作业的执行时间，或是处理引擎需要多长时间读数据、计算、以及输出结果，而流式处理则对作业时间有严格的要求，力求把时间延迟降到最低，追求实时性。通常来说，批处理用于处理海量的历史数据，属于有界数据流，总的时间长度是可以衡量的，相反，流处理程序提供在短时间内处理源源不断输入的海量数据，属于无界数据流，在这里我们用延迟和吞吐量来衡量两者的需求。

延迟

延迟表示处理事件所需要的时间，实际上就是从接收事件到事件结束这个过程的时间长度。举个生动的例子，假如你是一个特别喜欢喝咖啡的人，每天必去的地方就是咖啡馆，每次当你去店里买咖啡的时候，经常会遇到咖啡店里有其他顾客的情况，这时候你就需要排队等候，等到其他顾客下完单的时候才轮到你，当你下完单给完钱之后，收银员会把你的订单信息给到咖啡师，然后就开始了一段时间的等待，当你的咖啡煮好的时候，咖啡师会通知你过来拿咖啡，终于，你如愿以偿的拿到了心爱的咖啡，并享受了第一口咖啡。在这个过程中，从你走进咖啡店到你喝上第一口咖啡这个过程耗费的时间就是延迟。

在数据流中同样是以时间延迟作为实时性衡量的标准，对于不同的应用程序，你可能会在意平均延迟、最大延迟或者延迟率，例如平均延迟为 10ms 的应用程序意味着处理事件的时间大约需要 10ms，假如延迟率为 95%的应用程序的延迟时间为 10ms，则说明只有 95%的事件能在 10ms 左右被处理完，平均延迟值隐藏了处理延迟的分布情况，可能会难以定位问题。例如：如果咖啡师在为你准备咖啡时发现牛奶用完了，这时候你需要付出额外的时间等待咖啡师去拿牛奶，这意味你的咖啡会有更大的延迟，但是其他大部分用户并不会受到影响，因为他们可能已经喝上咖啡了或者正在在咖啡店的路上。

对于很多流式应用场景来说，保证低延迟是必要且重要的，比如系统告警、欺诈检测、网络监控等。低延迟是流处理的一个关键特性，是实现实时应用的基础要求，像 Apache Flink 这样的主流流处理器可以实现几毫秒的延迟，与之相反的是，传统的批处理延迟通常从几分钟到几个小时不等；在批处理中，首先你需要把事件放到批次里面，然后才能处理它们，因此，延迟受到批次中的最后一个事件到达时间的限制，而且还受到批次中数据量大小的影响。真正的实时流处理是无法接受这种模式的，通常是来一条就处理一条，或者进行微批处理，粒度很小，因此可以保证低延迟。

吞吐量

吞吐量是衡量一个系统处理数据能力的重要指标，也就是常说的处理率。表明了系统在每个时间片可以处理多少事件的能力，举个例子，如果咖啡店从早上 7 点营业到晚上 7 点，一天下来服务了 600 个顾客，则咖啡店的平均吞吐为 50 个顾客/每小时。在流系统中，我们需要保证时间延迟尽可能的低，而吞吐量尽可能的高。

吞吐量是通过每个时间片内处理事件的能力来度量的，但是要注意的是，吞吐量的高低还取决于事件到达的状态，古语有云：巧妇难为无米之炊，如果你都没有事件过来，我也无法体现出我的能力，因此某段时间内吞吐量低并不意味着系统的性能差。在流式系统中，通常希望能够确定系统在运行过程中在某个时间片能处理的事件的极值，这里的极值也就是我们在性能测试中经常提到的“峰值”和“谷值”，通常我们希望知道系统在最大负载下的性能以及最小负载下的性能分别是什么情况。假如说，事件一进来就立马被系统处理掉，这种情况的延迟是非常低的，就好比你是咖啡店刚营业的第一个顾客，咖啡师会第一个为你服务，这样你喝到第一口咖啡的时间就是最短的。在理想的情况下，我们希望保证自己实现低延迟的同时也能确保其他事件达到低延迟。但是系统资源是有限的，事件流是无限制的，一旦输入的事件需要的资源达到了系统资源的阈值，这种情况下就必须要设置缓冲区，用于缓冲事件，比如咖啡店的排队机制，当你去吃完饭想去喝杯咖啡，此时刚好有很多人和你有同样的想法，当你们同时出现在咖啡店的时候发现咖啡店已经爆满了，好多人都在店门口排队，为了喝上心爱的咖啡，你也加入了排队大军中，如果门口等待的人是暴脾气，死活都要闯进去，这样导致的后果就是咖啡店没法正常营业，你们也无法喝上心爱的咖啡，对于系统来说就会出现缓冲区不可用，数据丢失的情况，这种情况通常称为反压，有不同的策略供你选择，在第三章中我们会详细讨论 Flink 的反压机制。

延迟 VS 吞吐量

在这里需要明确的一点是，延迟与吞吐并不是两个互相独立的指标。如果事件到达数据处理管道的时间比较长，便无法保证高吞吐。同理，如果系统的资源不足、性能较低，则事件会被缓存并等待，直到系统有能力处理。还是以咖啡店为例，首先比较好理解的是，在负载低的时候，资源充足，时间延迟很低。例如你是咖啡店里第一个顾客，只需要很短的时间就能喝上咖啡。但是在咖啡店较忙的时候，顾客就需要排队等候，这种情况下延迟很高。另外一个影响延迟的因素是处理一个事件的时间，跟业务的处理逻辑有关系。例如咖啡师为每个顾客做咖啡所消耗的时间。假设在一个圣诞节，咖啡师需要在每杯咖啡上画一个圣诞老人。也就是说，每杯咖啡制作的时间会增加，导致每个顾客在咖啡点等待的时间边长了，最终使得整体吞吐量下降。

那是否可以同时达到低延迟与高吞吐？在咖啡店的例子中，你可以招聘大师级的咖啡师，提高制作咖啡的效率。这里主要考量的地方是：减少延迟并提高吞吐量。如果一个系统执行的操作更快，则它就可以在同一时间内处理更多的事件。另外的方法是招聘更多的咖啡师，在同一时间服务更多顾客。在流处理管道中，通过增加流的并行度处理事件，既保证了低延迟，也保证了高吞吐，两全其美。

数据流算子

流处理引擎内置了丰富的算子用于加载数据、数据转换和输出数据流，这些算子可以组合起来构成数据流程图，形成流应用程序的处理逻辑，下面介绍最常用的数据流算子。

算子分为无状态算子和有状态算子，无状态算子不保存任何内部状态，也就是说，处理此事件时，并不依赖于任何过去的事件，也不需要保存本身的信息。无状态算子操作简单，易于并行化，独立于其他事件，不需要考虑事件到达的顺序。在出现错误时，可以简单地重新执行无状态算子，从它丢失出现异常的地方开始继续执行即可。相反，有状态算子依赖于以前接收到的事件，可以把当前状态用于下次事件更新或者其他计算逻辑处理，比如说累加或计数操作；有状态流处理应用程序在并行化和容错操作方面非常具有挑战性，需要对状态进行有效分区，在发生故障时能够可靠地进行恢复。在本章的最后，你将了解更多关于有状态流处理、故障场景和一致性的内容。

数据加载和数据输出

数据加载和数据输出操作允许流处理引擎与外部系统通信。数据加载一般是指从外部源获取原始数据并将其转换成适合处理的格式。实现数据加载逻辑的操作称为数据源（source）算子。支持从很多渠道加载数据，如数据源可以从 TCP Socket、文件、Kafka Topic 或传感器数据接口获取数据，数据输出是指在数据转换之后以适合外部系统使用的形式输出。执行数据输出逻辑的操作称为数据接收端(sink)算子，例如文件、数据库、消息队列和监控接口等。

转换操作

转换操作是一个单次操作，每次单独处理一个事件，用于对事件做一系列转换后输出一个新的输出流，转换逻辑可以整合在算子中，或是由用户定义的函数（UDF）实现。如图 2-4 所示：

Flink 算子可以接收多个输入流并产生多个输出流，可以用于调整数据流图的结构，例如将一个流拆分为多个流，亦或是将多个流整合为一个流。我们将在第 5 章中讨论 Flink 中所有可用算子的语义。

滚动聚合

滚动聚合是一个聚合操作，例如 sum，minimum 以及 maximum，它会对每个输入的事件做持续的更新。聚合操作是有状态的，它将输入的数据与当前的状态信息进行整合，从而产生一个更新后的聚合值。为了高效地与当前状态进行整合，并输出一个单一的值，聚合操作必须进行状态关联和交换。否则有状态算子需要保存整个流的历史记录。图 2-5 是一个滚动聚合求最小值的示例，它持有当前最小值，并根据输入的事件更新当前最小值：

窗口操作

转换操作与滚动聚合每次处理一个事件并产生一个新的事件，继而更新状态。然而，某些操作需要收集并缓存记录然后再计算它们的结果。举个例子，对流做 join 操作或整体聚合，例如求中位数，为了在无界流上有效地计算这些操作，需要限制这些操作所维护的数据量。在本节中，我们会讨论提供这种机制的窗口语义。

窗口操作除了具有实际应用价值外，窗口还提供不同的时间语义对流进行查询，从上面可以看到滚动聚合是如何对流的历史记录执行聚合计算并得到最终结果的，并保证了事件结果的低延迟。这对于某些应用程序来说是可以的，但是如果你只对最近的数据感兴趣呢?假设这样的一个场景，你想通过应用程序向司机提供实时交通信息，这样一来他们就可以避开拥挤的路线。在这种情况下，现在你想知道在过去几分钟内某个位置是否发生了事故。另一方面，如果你想了解过去发生过的所有事故就没有那么容易了。更重要的是，如果仅是对流历史记录做一个单聚合，则会损失有关数据随时间变化的信息。例如，你可能想知道每 5 分钟有多少辆车通过十字路口。

窗口操作不断地从无界事件流创建有界的事件集，我们称之为 bucket，基于这些有界的事件集进行计算。事件通常根据数据本身的属性或时间分配给 bucket，为了正确使用窗口语义，我们需要聚焦两个主要问题:“事件是如何分配给 bucket ”以及“窗口多久产生一次结果”。窗口的行为是由一组策略定义的，窗口策略决定何时创建新 buckets、将哪些事件分配给哪些 buckets、以及何时计算 buckets 的内容。何时计算 buckets 的内容基于一个触发条件，当满足触发条件时，bucket 的内容会被发送到一个计算函数，基于 bucket 中的元素进行逻辑计算，比如做 sum 或者 minimum 这样的聚合操作，或者是自定义函数。窗口的策略可以是基于时间(例如最近 5 秒内接收到的事件)、基于数量(例如最近 100 个事件)或数据属性。下面我们介绍几种常用的窗口类型：

滚动窗口（Tumbling windows）：滚动窗口具有固定的窗口大小，并且窗口之间不会重叠，有两种触发计算的策略，第一种是基于事件数量，当满足设置的数量时会执行计算，如图 2-6 所示：

第二种是算子基于固定的时间窗口执行计算，不管的固定时间区间里的数量是多少，只要超出时间边界就执行计算，如同 2-7 所示。

滑动窗口（Sliding windows）：滑动窗口以一个步长（Slide）不断向前滑动，窗口的长度固定。定义一个滑动窗口需要设置步长（Slide）和滑动窗口的长度（Size）。步长的大小决定了 Flink 创建新的窗口的频率，步长设置比较小的话，会产生大量的时间窗口，数据存在重复消息的情况；步长设置大于窗口长度的话，会出现丢失数据的情况；图 2-8 表示了一个长度为 4，步长为 3 的滑动窗口。

会话窗口（Session windows）：会话窗口在某些实际的应用场景中比滚动窗口和滑动窗口更加适用，假设有这样的一个应用场景，需要分析在线用户的行为，在这样的场景中我们希望将来自同一个用户的行为信息根据不同时间内的会话进行分组，通常来说，这样的会话活动有可能是连续的，也有可能是断断续续的，例如，用户在浏览新闻时点击不同的页面，可以被看作一个会话。因为一个会话的时长没有办法预先定义好，而是取决于用户的实际活动时长。对这样的场景来说，滚动窗口以及滑动窗口显得并不适用。我们需要的是一个窗口操作可以将所有属于同一个会话的事件，分发到同一个 bucket 中，会话窗口可以根据一个“会话间隔”（session gap）定义一个会话的过期时间，超过了会话间隔，则关闭当前的会话窗口，直至下一个会话窗口被触发，如图 2-9 所示：

目前我们所看到的所有窗口类型都是基于单一流的处理，但是在实际中并非如此，你更希望将流划分成多个逻辑流分发到并行窗口中，举个例子，假如你通过不同的传感器接收测量数据，对于你来说，为了提高计算效率，你希望在进行窗口计算之前先根据传感器 ID 对数据做分组，这样一来，在并行窗口中，每个分区都独立于其他分区的窗口策略，图 2-10 显示了一个基于计数的并行滚动窗口，数量为 2，图中不同的颜色对应不同的分区。

       Figure 2-10. A parallel count-based tumbling window of length 2 

在流处理中需要特别注意两个密切相关的概念：时间语义和状态管理，相对而言，时间是流处理中最重要的指标，在旁人看来，低延迟是流处理引擎一个非常吸引人的特性，事实上，它的意义已经远远超出了快速分析的范畴了，在真实的应用场景中，往往会因为系统本身的问题、网络不稳定以及各种各样的不稳定因素引起数据出现乱序或者迟到的情况，这种情况是经常发生的，那么一旦发生了这种情况，我们要如何保证数据的准确性呢？这是我们要思考的问题。除此之外，流处理应用除了处理当前产生的事件外，还应具备处理历史事件的能力，如此一来就可以用作离线分析或者时间旅行分析（time travel analyses）。这一切都要基于系统能够实现对状态信息容错的基础上，否则这些功能均毫无意义。到目前为止，我们所看到的所有窗口类型都需要先将数据做缓存，继而用于计算得到结果。事实上，如果你想在一个流应用上做任何简单或复杂的逻辑操作，例如简单的计数操作，也需要维护状态信息。一个流处理应用可能会运行几天、几个月、乃至几年的时间，我们需要确保在发生任何故障之后，都能够有效可靠地恢复状态信息，并能够保证在状态恢复完成之后仍然能够提供准确的结果。在本章接下来的内容中，我们将更深入地了解当数据流处理异常情况下的时间和状态保证的概念。

时间语义

在本节内容中，我们会介绍时间语义，以及描述流处理中不同的时间概念，同时，我们将讨论流处理引擎如何能够为无序事件提供准确的结果，以及如何用流处理历史事件和实时计算。

一分钟在流处理中意味着什么？

当我们要处理持续不断输入的无界数据流时，这时候时间就成了一个至关重要的指标，假如你希望连续得到计算结果，比如说每一分钟内的计算结果，一分钟对于我们来说就是时间上的感知，那么在流应用程序中，你知道一分钟意味着什么吗？

假设这样的一个场景，一个程序用于分析手机网游用户的行为信息，每个用户分属不同的队伍。系统通过收集小队的信息，根据小队成员达成游戏设定目标的速度，在游戏中给出不同的奖励，例如提升游戏等级、增加游戏经验值等。例如，如果一个小队的所有成员在一分钟内生产了 500 个泡泡，就可以给小队或者队员提升一个等级。小吴是该游戏的骨灰级玩家，他每天早上利用坐公交车去上班的时间玩游戏，由于地域荒凉，经常会出现网络不好的情况，小吴住在大山里，在出大山的路上网络特别差，有一天小吴在做游戏任务，开始在游戏中生产泡泡，刚开始手机网络网络正常，游戏也没出现异常，系统后台能够正常拿到小吴的游戏信息，突然，公交车进入了一个隧道，网络信号很差，导致小吴的手机断网了，但是小吴并没有意识到这个问题，小吴继续玩着游戏，断网期间的游戏数据会缓存到他的手机里，当公交出了隧道，网络恢复后，缓存待发送的游戏数据会被发送到分析程序。此时分析程序怎么做？在这里一分钟意味着什么呢？是否考虑到了小吴断网的时间？图 2-11 形象地描述了上面的问题：

在线游戏是一个简单的场景，更多地是取决于事件实际发生的时间，而不是应用程序接收到事件的时间。在上面的情景中，对于手机网游用户来说，如果说在断网期间内丢失了游戏数据，无疑是一种非常差的用户体验，往往会导致游戏用户退游或者游戏热情大大受挫；对于那些非常注重时间的系统应用，引起的后果可能会更严重。如果只是考虑在一分钟内接收到多少个事件，那么结果会直接取决于网络连接状况、或是处理的速度等。相反，真正定义一分钟内的事件是事件本身的时间。通过小吴的游戏例子，体现出了流处理应用程序的两个时间概念：处理时间（processing time）和事件时间（event time），接下来我们对这两个时间概念展开描述。

处理时间（Processing Time）

处理时间是正在执行流处理逻辑所在的服务器上本地时钟的时间，简而言之就是系统时间，处理时间窗口包括了在指定时间区间内所接收到的所有事件，以本地机器时间衡量。如图 2-12 所示，在小吴的示例中，当手机断网时，没有数据上传，这种情况下，处理时间窗口的时间会持续计时，不会理会在这段时间内没有进入时间窗口的游戏数据。

事件时间（Event Time）

事件时间顾名思义指的是事件本身发生的真实时间，而不是进入机器执行计算的系统时间，事件时间是基于事件发生时被打上的时间戳，会被附带到数据流上，一般来说，在事件进入处理管道之前事件的时间戳就已经存在了，比如说事件被创建出来的时候就有了时间戳，图 2-13 展示了事件时间窗口将事件正确地进行放置，反映了事件实际发生的情形，就算事件存在延迟也不影响。

事件时间实现了处理速度与结果之间完全解耦，基于事件时间的操作是可预测的，并且结果是确定的。使用事件时间窗口计算时，无论流处理的速度有多快，或是事件到达处理管道的速度有多慢，最终都能得到相同的结果。

事件时间可以解决很多棘手的问题，处理延迟事件只不过是其中的一个方面，它还可以解决其他不同应用场景的问题，包括处理乱序数据。延续上面小吴的例子，小刘也是在线手游的骨灰级玩家，这一天恰巧和小吴在同一班公交车上，两个人玩着同一款游戏，但是小吴和小刘使用的是不同的运营商网络，当小吴的手机在隧道里断网的时候，小刘的手机网络却不受任何影响，小吴断网期间，小刘的游戏数据正常发送给了系统后台。基于事件时间进行回放，就算数据是无序的，我们也可以保证结果的准确性，而且，通过关联可进行数据流回放，基于明确的时间戳也可以进行快速计算分析，也就是说，通过回放数据流并分析历史数据，你也可以把流里面的事件当作实时发生的一样，因为每个事件本身都带有真实的时间。另外，你可以将计算快进到当前时间，这样一来，当程序同步到了现在正在发生的事件，就可以作为实时计算的程序，使用完全相同的处理逻辑。

水位线（Watermarks）

到目前为止，在我们关于事件时间窗口的讨论中，我们忽略了一个非常重要的问题：我们如何决定何时触发事件时间窗口？也就是说，我们需要等多久才能确定我们已经接收到了在某个时间点之前发生的所有事件？我们怎么知道数据有延迟呢？考虑到分布式系统的不可预测的各种问题，以及可能由于外部组件本身或者其他因素引起的延迟，这些问题还没有很完美的答案。在本节中，我们将看到如何使用水位线来配置事件时窗口行为。

水位线是一个全局进度的指标，通过水位线我们确信在某个时间点不会再有延迟事件到来。本质上，水位线提供了一个逻辑时钟，告知系统当前事件的时间。当算子收到时间为 T 的水位线时，它可以假定不会再收到时间戳小于 T 的事件。对于事件时间窗口和处理无序事件的算子来说，水位线都是必不可少的。一旦达到水位线，算子就会收到信号，表示一定时间间隔内的所有时间戳都已被扫描到了，开始触发计算或者对接收的事件进行排序。

水位线作为一个用于权衡结果可信度与时间延迟的配置，逻辑上就是时间容忍度，水位线设置比较小的时候可以保证低延迟，但是结果可信度就不是很高，因此可能有数据在触发的时候还没达到水位线的要求。在这种情况下，延迟的事件可能会在水位线之后到达，我们需要处理这些在水位线之后到达的事件。另一方面，如果水位线设置得特别大，虽然能尽可能保证结果的准确性，但是延迟会比较高。

在很多实际的应用中，系统往往没办法完美地准确算出水位线的值。在手机游戏的例子中，基本上我们无法预测用户丢失连接的时间有多长，他们可能要穿过隧道，登上飞机，或者再也不玩了。无论是自定义的水位线或是自动生成的水位线，在分布式系统中，由于有落后的任务，跟踪整个分布式系统的进度可能仍是个问题。所以，如果只是单纯地依赖于水位线，可能并不是一个很好的方法。相反，流处理系统还得提供一些机制去处理这些落后于水位线的事件。结合实际应用的需求，可以选择丢弃这些不合要求的事件，或是通过日志记录下来，或者使用它们做结果比对。

处理时间 VS 事件时间

在这一点上，你可能会有疑问，既然事件时间可以解决我们所有的问题，那为什么还要研究处理时间呢？事实上，在某些特定的场景下，处理时间比事件时间更合适，处理时间窗口能尽可能地降低时间延迟，因为不需要等待，事件进入处理管道就执行计算，在不关心延迟和数据乱序的情况下，窗口只是用来缓存数据并在达到指定时间长度后触发计算而已，相当于一个临时的仓库，其容量是有限的，并且有时间规定的，所以对那些严格要求时间且对数据准确性要求没那么高的应用场景来说，处理时间显得非常适用。还有另一种情况，假如你需要定期实时汇报结果，而且对结果的准确性不做要求，例如一个实时监控的仪表盘，会将接收到的事件进行聚合计算，并把结果展示在仪表盘上。最后，处理时间窗口本身对流提供了可靠性保证，这在某些场景下是一个非常好的特性。总之，处理时间保证低延迟，但是输出的结果取决于处理事件的速度，并且结果准确性没有很好的保证。而事件时间能最大程度保证结果的准确性，并可以处理延迟或者乱序事件，但是延时相对较大。

状态与一致性模型

前面说到了时间概念，这里我们介绍流处理另一个极其重要的概念：状态。状态在数据处理中显得无处不在，几乎所有的重要计算都会涉及到状态，为了得到结果，函数通常会计算一段时间内事件累积的状态，例如计算 10 分钟内某个十字路口经过了多少辆车，就需要对 10 分钟内的所有车辆进行累加。有状态的算子使用流的输入事件以及内部状态，从而计算出它们的输出结果。例如，一个滚动聚合算子输出当前它读入的所有的事件的总和。算子拿到当前聚合的值作为它的内部状态，并在每次读入新值时对它做更新。例如，当算子在 10 分钟内检测到“高温”事件之后，紧接着就会出现“冒烟”事件，然后就会发出警报。这种情况下算子需要将“高温”事件存储在其内部状态，直到看到“冒烟”事件或超过了 10 分钟的过期时间就发出警报。

如果考虑到使用批处理系统分析无界数据集的情况，则状态就显得更加重要了。在现代流处理引擎崛起之前，一种处理无界数据的常用方法是将数据划分为多个小批次，然后在批处理系统上执行重复调度作业，即微批处理。作业完成后，将结果持久化到存储系统中，然后清空算子的状态。如此一来，当执行下一个作业时，它将无法获取到上一个作业的状态，为了解决这样的问题，通过引入外部系统来做状态管理，例如数据库。相反，在连续运行的流作业中，状态会一直存在于事件中，在编程模型中是公开透明的。理论上，我们还是可以通过引入外部存储系统来做状态管理，只不过会增加额外的延迟，这是我们要考量的问题。

由于流处理引擎通常会用来处理无界数据，所以需要注意的是，不要让内部状态无限地增长下去。为了控制状态值大小，一般来说，flink 算子会维护一些到目前为止看到的所有事件的概要信息。这个概要信息可以是计数、总和、 截至目前的事件样本、或者是用于存储某些带有属性的自定义数据结构等等。

正如你可能想象的那样，有状态的算子的实现会存在一些挑战：

l 状态管理

系统需要高效地管理状态，并确保它不会受到并行更新处理的影响。

l  状态分区

在并行处理时，场景较为复杂，因为输出的结果依赖于前面事件的状态以及持续输入的事件。在大部分应用场景下，我们可以通过 key 将状态进行分区，对每个状态进行单独管理。例如处理的输入流作为一组传感器的事件，这时可以使用分区的算子状态独立地维护每个传感器的状态信息。

l 状态恢复

最大的挑战是：在发生故障时，有状态的算子如何能够确保状态可以被恢复，并且仍能输出正确的结果。

接下来我们详细地讨论一下任务失败以及结果保证。

任务失败

在流处理作业中，状态是非常重要的信息，需要具备容错的能力。如果在一次失败的作业中，导致状态丢失，就算最终作业恢复正常了，最终得到的结果也可能是不对的。流处理作业一般会长时间持续运行，所以可能会每隔几天甚至几个月才收集一次状态信息，如果说由于作业失败导致状态丢失，需要通过重新执行丢失状态的部分逻辑去生成丢失的状态，这无疑是一个费时费算力的过程，代价很大。

在本章的开头，你了解了如何将流程序建模为数据流图。在执行之前，它们被转换为成千上百个并发任务的物理执行计划，而在流作业长期运行的过程中，程序无法预测每个任务什么时间会出现失败的情况。如何确保这些故障可以被正确地处理，并让流作业继续运行？事实上，我们需要流处理引擎做到的是不仅能在任务失败时继续处理，而且还需要保证结果以及状态的正确性。在本节中我们会讨论所有的这些问题。

什么是任务失败？

对于输入流中的每个事件，任务执行以下步骤:

(1)接收事件，将其存储在本地缓冲区中;

(2)更新内部状态;

(3)生成输出记录。

在这些步骤中任何一个步骤都有可能发生故障，系统必须在故障场景中清楚地定义其行为。如果任务在第一步中失败，事件会丢失吗?步骤二如果更新内部状态失败，在恢复后会再次更新吗?在这些情况下，最终输出的结果是正确的吗?

在批处理场景中，你可以轻松地回答这些问题。最简单的方法就是重新启动作业，回放所有数据，所有的状态从零开始，这样就不会丢失事件，然而，在流处理场景中，处理失败任务并不是一个微不足道的问题。流处理引擎通过提供结果保证来定义它们在出现故障时的行为。接下来，我们将介绍流处理引擎提供的数据处理的可靠机制，用来表明在实际生产运行中会对数据处理提供哪些保障。

结果保证

在介绍不同类型的保证机制之前，我们需要明确几个要点，在说到流处理引擎中的任务失败时，这些要点经常会引起混淆。在本章的其余部分，当我们讨论“结果保证”时，指的是流处理引擎内部状态的一致性。也就是说，我们关心的是从失败的任务恢复后的状态值。值得注意的是，流处理引擎通常只能保证程序内部状态的一致性而不是结果一致性。但是，一旦数据被发送到数据接收端，就很难保证结果的正确性，除非数据接收端支持事务。一般来说，流处理引擎通常为用户的应用程序提供以下几种数据处理语义。

最多一次（at-most-once）

当一个任务失败时，最简单的办法就是不做任何操作来恢复丢失的状态和重放丢失的事件。At-most-once 是保证对每个事件进行最多一次处理。换句话说，数据可能不是非常重要，允许出现丢失，对结果的正确性要求不高，如果你可以接受结果存在误差且要求低延迟的话，那可以采用 at-most-once 这种处理机制。

至少一次（at-least-once）

在大多数实际应用场景中，是无法容忍出现数据丢失的，如金融业务、银行交易等等。这种类型的保证称为“至少一次”，系统会保证数据或事件至少被处理一次。如果发生错误或者丢失，那么会从数据源头重新发送数据执行到流处理程序中，意味着同一个事件或者消息可能会被处理很多次。为了确保至少一次的结果正确性，您需要有一种机制来重放事件，可以是数据源或者缓冲区。将所有事件写入事件日志做持久化，以便当任务失败时进行重放。另一种方式就是使用记录确认（record acknowledgements），这种方法的原理是把每个事件存储在缓冲区中，直到事件确认被任务处理过了，这时候就可以选择清理掉该事件。

有且只有一次（exactly-once）

exactly-once 是最精确也是最具挑战性的一种保障机制。exactly-once 意味着不会有事件丢失，而且对于每个事件的内部状态只会更新一次。从本质上讲，准确地说，一旦采用 exactly-once 机制，我们的应用程序就会提供正确的结果，就好比任务失败从来没有发生过一样。

提供 exactly-once 保证需要满足 at-least-once 保证，因此需要同时支持数据重放。此外，流处理引擎需要确保内部状态的一致性。也就是说，在恢复之后，它应该知道事件更新是否作用到了状态上。事务性更新是其中一种实现的方式，但是它可能带来大量的性能开销。相反，Flink 使用轻量级的快照机制来实现 exactly-once 的结果保证。我们将在第三章讨论 Flink 的容错算法，包括 Checkpoints、Savepoints 和 State Recovery。

端到端有且只有一次（End-to-end exactly-once）

到目前为止，你看到的保证类型都是基于流处理引擎本身而言的，但是在实际流用中，一个完整的流架构通常要有数据源、流处理引擎和数据接收端这几个核心组件。端到端保证是指整个数据处理管道的结果正确性，每个组件都有自身的保证机制，而对于完整的管道来说，端到端保证是架构中最弱的一环。需要注意的是，有时候可以通过简单的语义来实现可靠性保证，比如说幂等操作，求最大值或最小值，使用 at-least-once 语义也可以做到 exactly-once 语义的效果。

总结

在本章中，你已经学习了数据流处理的基本概念和原理，熟悉了数据流编程模型，并了解了如何将流应用程序表示为分布式数据流图。紧接着了解了并行处理无界数据流的需求，并认识到延迟和吞吐量对流应用程序的重要性。并且学习了一些基本的流算子，以及如何使用时间窗口对无界数据流进行计算。明确了流处理中时间语义，对事件时间和处理时间的概念有了清晰地认识。最后，你也知道了为什么“状态”是流处理应用程序时至关重要的概念，以及如何实现任务失败保障和结果的正确性。

到目前为止，我们已经学习了独立于 Apache Flink 的流概念。在本书的其余部分，我们将学习 Flink 是如何实现这些概念的，以及如何使用 DataStream api 编写具备我们目前介绍的所有功能的应用程序。