Introduction to Apache Flink

人们希望流处理不仅做到“低延迟”和“高吞吐”，还可以处理中断。优秀的流处理技术应该能使系统在崩溃之后重新启动，并且产出准确的结果；换句话说，优秀的流处理技术可以容错，而且能保证“exactly-once”。

Apache Storm 是“流处理”的先锋。Storm 提供了低延迟（“毫秒级”）的流处理，但它没有实现高吞吐。

Spark Streaming 的吞吐率很高，不过它本质上还是用“批处理”模拟“流处理”：对一个时间段内的数据收集起来，作为一个 RDD，再处理，这种处理方式被称为“微批处理”。Spark Streaming 的延迟一般是“秒级”。

Flink 具有诸多优势，如图 1 所示。

Flink 起源于 Stratosphere 项目，Stratosphere 是在 2010~2014 年由 3 所地处柏林的大学和欧洲的一些其他的大学共同进行的研究项目。2014 年 12 月一跃成为 Apache 软件基金会的顶级项目。2015 年 10 月，第一届 Flink Forward 研讨会在柏林举行。

采用批处理架构可以解决上面提到计数问题。在该架构中，持续摄取数据的管道每小时创建一次文件。这些文件通常被存储在 HDFS 或 MapR-FS 等分布式文件系统中。像 Apache Flume 这样的工具可以用于完成上述工作。由调度程序安排批处理作业（如 MapReduce 作业）分析最近生成的一个文件（将文件中的事件按 key 分组，计算每个 key 对应的事件数），然后输出计数结果。对于每个使用 Hadoop 的公司来说，其集群都有多个类似的管道。

这种架构完全可行，但是存在以下问题。

• 对时间的处理方法不明确。 假设需要改为每 30 分钟计数一次。这个变动涉及工作流调度逻辑（而不是应用程序代码逻辑），从而使 DevOps 问题与业务需求混淆。
  
• 乱序事件流。 在现实世界中，大多数事件流都是乱序的，即事件的实际发生顺序（事件数据在生成时被附上时间戳，如智能手机记录下用户登录应用程序的时间）和数据中心所记录的顺序不一样。这意味着本属于前一批的事件可能被错误地归入当前一批。批处理架构很难解决这个问题，大部分人则选择忽视它。
  
• 预警不方便。 假设除了每小时计数一次外，还需要尽可能早地收到计数预警（比如在事件数超过 10 时预警）。为了做到这一点，可以在前面介绍的定期运行的批处理作业之外，引入 Storm 来采集消息流，Storm 实时提供近似的计数。但这增加了系统的复杂性。
  
• 批处理作业的界限不清晰。 在该架构中，“每小时”的定义含糊不清，分割时间点实际上取决于不同系统之间的交互。充其量也只能做到大约每小时分割一次，而在分割时间点前后的事件既可能被归入前一批，也可能被归入当前一批。
  

采用流处理架构解决计数问题，可以克服上一节提到的采用批处理架构所遇到的问题。

Flink 可以以时间为单位把事件流分割为一批批任务（称作“窗口”），这种逻辑完全嵌入在 Flink 程序的应用逻辑中。预警可由同一个程序生成，乱序事件可由 Flink 自行处理。Flink 应用程序用来计数的代码非常简单，如下：

DataStream<LogEvent> stream = env
    // 通过Kafka生成数据流
    .addSource(new FlinkKafkaConsumer(...))
    // 分组
    .keyBy("country")
    // 将时间窗口设为60分钟
    .timeWindow(Time.minutes(60))
    // 针对每个时间窗口进行操作
    .apply(new CountPerWindowFunction());

流处理区别于批处理最主要的两点是：1、流即是流，不必人为地将它分割为文件；2、时间的定义被明确地写入应用程序代码（如以上代码的时间窗口），而不是与摄取、计算和调度等过程牵扯不清。

在流处理中，主要有三个时间概念（重点关注前两个时间）：

• 事件时间，即事件实际发生的时间。更准确地说，每一个事件都有一个与它相关的时间戳，并且时间戳是数据记录的一部分（比如手机或者服务器的记录）。事件时间其实就是时间戳。
  
• 处理时间，即事件被处理的时间。处理时间其实就是处理事件的机器所测量的时间。
  
• 摄取时间，也叫作进入时间。它指的是事件进入流处理框架的时间。缺乏真实事件时间的数据会被流处理器附上时间戳，即流处理器第一次看到它的时间。
  

图 3 说明了事件时间和处理时间的区别。以《星球大战》系列电影为例。首先上映的 3 部电影是该系列中的第 4、5、6 部（这是事件时间），它们的上映年份分别是 1977 年、1980 年和 1983 年（这是处理时间）。之后按事件时间上映的第 1、2、3、7 部，对应的处理时间分别是 1999 年、2002 年、2005 年和 2015 年。由此可见，事件流的顺序可能是乱的（尽管年份顺序一般不会乱）。

在现实世界中，许多因素（如连接暂时中断，不同原因导致的网络延迟，分布式系统中的时钟不同步，数据速率陡增，物理原因，或者运气差）使得事件时间和处理时间存在偏差（即事件时间偏差）。 事件时间顺序和处理时间顺序通常不一致，这意味着事件以乱序到达流处理器。

根据应用程序的不同，两个时间概念都很有用。有些应用程序（如一些预警应用程序）需要尽可能快地得到结果，即使有小的误差也没关系。它们不必等待迟到的事件，因此适合采用“处理时间”语义。其他一些应用程序（如欺诈检测系统或者账单系统）则对准确性有要求：只有在时间窗口内发生的事件才能被算进来。对于这些应用程序来说，“事件时间”语义才是正确的选择。也有两者都采用的情况，比如既要准确地计数，又要提供异常预警。

Flink 允许用户根据所需的语义和对准确性的要求选择采用事件时间、或处理时间、或摄取时间定义“窗口”。

当采用“事件时间”定义窗口时，应用程序可以处理乱序事件流以及变化的事件时间偏差，并根据事件实际发生的时间计算出有意义的结果。

节 2.2 说明了如何在 Flink 中定义时间窗口并以小时为单位生成聚合结果。 窗口是一种机制，它用于将许多事件按照时间或者其他特征分组，从而将每一组作为整体进行分析（比如求和）。

流处理架构的一个核心能力是时空穿梭（Time Travel）。时空穿梭意味着将数据流倒回至过去的某个时间，重新启动处理程序，直到处理至当前时间为止。这要求数据源（如 Kafka）必须能按照你的要求重放数据，Kafka 支持重放 Topic 中的消息。

支持事件时间对于流处理架构而言至关重要，因为事件时间能保证结果正确，并使流处理架构拥有重新处理数据的能力。当计算基于事件时间时，如何判断所有事件是否都到达，以及何时计算和输出窗口的结果呢？换言之，如何追踪事件时间，并知晓输入数据已经流到某个事件时间了呢？为了追踪事件时间，需要依靠由数据驱动的时钟，而不是系统时钟。

以图 4 中的一分钟滚动窗口为例。假设第一个窗口从 10:00:00 开始，需要计算从 10:00:00 到 10:01:00 的数值总和。当时间就是记录的一部分时，我们怎么知道 10:01:00 已到呢？换句话说，我们怎么知道盖有时间戳 10:00:59 的元素还没到呢？

Flink 通过水印来推进事件时间。 水印是嵌在流中的常规记录，计算程序通过水印获知某个时间点已到。对于上述一分钟滚动窗口，假设水印标记时间为 10:01:00（或者其他时间，如 10:03:43），那么收到水印的窗口就知道不会再有早于该时间的记录出现，因为所有时间戳小于或等于该时间的事件都已经到达。 这时，窗口可以安全地计算并给出结果（总和）。水印使事件时间与处理时间完全无关。迟到的水印（“迟到”是从处理时间的角度而言）并不会影响结果的正确性，而只会影响收到结果的速度。

当在分布式系统中引入状态时，自然也引入了一致性问题。一致性实际上是“正确性级别”的另一种说法，即在成功处理故障并恢复之后得到的结果，与没有发生任何故障时得到的结果相比，前者有多正确？举例来说，假设要对最近一小时登录的用户计数。在系统经历故障之后，计数结果是多少？在流处理中，一致性分为 3 个级别：

• at-most-once：这其实是没有正确性保障的委婉说法——故障发生之后，计数结果可能丢失。
  
• at-least-once：这表示计数结果可能大于正确值，但绝不会小于正确值。也就是说，计数程序在发生故障后可能多算，但是绝不会少算。
  
• exactly-once： 这指的是系统保证在发生故障后得到的计数结果与正确值一致。
  

要保证 exactly-once，其实现比较复杂。最先保证 exactly-once 的系统（Storm Trident 和 Spark Streaming），它们为了保证 exactly-once，这些系统无法单独地对每条记录运用应用逻辑，而是同时处理多条（一批）记录，保证对每一批的处理要么全部成功，要么全部失败。这就导致在得到结果前，必须等待一批记录处理结束。因此，用户经常不得不使用两个流处理框架（一个用来保证 exactly-once，另一个用来对每个元素做低延迟处理），结果使基础设施更加复杂。

Flink 的一个重大价值在于，它既保证了 exactly-once，也具有低延迟和高吞吐的处理能力。后面将介绍相关原理。

Flink 如何保证 exactly-once 呢？它使用一种被称为“检查点”的特性，在出现故障时将系统重置回正确状态。下面通过简单的类比来解释检查点的作用。

假设你和两位朋友正在数项链上有多少颗珠子，如图 7 所示。你捏住珠子，边数边拨，每拨过一颗珠子就给总数加一。你的朋友也这样数他们手中的珠子。当你分神忘记数到哪里时，怎么办呢？如果项链上有很多珠子，你显然不想从头再数一遍。

于是，你想了一个更好的办法：在项链上每隔一段就松松地系上一根有色皮筋，将珠子分隔开；当珠子被拨动的时候，皮筋也可以被拨动；然后，你安排一个助手，让他在你和朋友拨到皮筋时记录总数。用这种方法，当有人数错时，就不必从头开始数。相反，你向其他人发出错误警示，然后你们都从上一根皮筋处开始重数，助手则会告诉每个人重数时的起始数值，例如在粉色皮筋处的数值是多少。

Flink 检查点的作用就类似于皮筋标记。 数珠子这个类比的关键点是：对于指定的皮筋而言，珠子的相对位置是确定的；这让皮筋成为重新计数的参考点。总状态（珠子的总数）在每颗珠子被拨动之后更新一次，助手则会保存与每根皮筋对应的检查点状态，如当遇到粉色皮筋时一共数了多少珠子，当遇到橙色皮筋时又是多少。当问题出现时，这种方法使得重新计数变得简单。

Flink 检查点的核心作用是确保状态正确，即使遇到程序中断，也要正确。记住这一基本点之后，我们用一个例子来看检查点是如何运行的。 Flink 为用户提供了用来定义状态的工具。 例如，以下这个 Scala 程序按照输入记录的第一个字段（一个字符串）进行分组并维护第二个字段的计数状态：

val stream: DataStream[(String, Int)] = ...

val counts: DataStream[(String, Int)] = stream
  .keyBy(record => record._1)
  .mapWithState((in: (String, Int), count: Option[Int]) =>
    count match {
      case Some(c) => ( (in._1, c + in._2), Some(c + in._2) )
      case None => ( (in._1, in._2), Some(in._2) )
    })

该程序有两个算子： keyBy 算子用来将记录按照第一个元素（一个字符串）进行分组，根据该 key 将数据进行重新分区，然后将记录再发送给下一个算子：有状态的 map 算子（ mapWithState ）。map 算子在接收到每个元素后，将输入记录的第二个字段的数据加到现有总数中，再将更新过的元素发射出去。

图 8 表示程序的初始状态：输入流中的 6 条记录被检查点屏障（checkpoint barrier）隔开，所有的 map 算子状态均为 0（计数还未开始）。所有 key 为 a 的记录将被顶层的 map 算子处理，所有 key 为 b 的记录将被中间层的 map 算子处理，所有 key 为 c 的记录则将被底层的 map 算子处理。

当该程序处理输入流中的 6 条记录时，涉及的操作遍布 3 个并行实例（节点、CPU 内核等）。那么，检查点该如何保证 exactly-once 呢？

检查点屏障和普通记录类似。它们由算子处理，但并不参与计算，而是会触发与检查点相关的行为。 当读取输入流的数据源遇到检查点屏障时，它将其在输入流中的位置保存到稳定存储中。 如果输入流来自消息传输系统（Kafka 或 MapR Streams），这个位置就是偏移量。Flink 的存储机制是插件化的，稳定存储可以是分布式文件系统，如 HDFS、S3 或 MapR-FS。图 9 展示了这个过程。

检查点屏障像普通记录一样在算子之间流动。 当 map（mapWithState）算子处理完前 3 条记录并收到检查点屏障时，它们会将状态（也就是 a 目前计数 0，b 目前计数 5，c 目前计数 1）以异步的方式写入稳定存储， 如图 10 所示。

当 map 算子的状态备份和检查点屏障的位置备份被确认之后，该检查点操作就可以被标记为完成，如图 11 所示。我们在无须停止或者阻断计算的条件下，在一个逻辑时间点（对应检查点屏障在输入流中的位置）为计算状态拍了快照。通过确保备份的状态和位置指向同一个逻辑时间点，后文将解释如何基于备份恢复计算，从而保证 exactly-once。值得注意的是，当没有出现故障时，Flink 检查点的开销极小，检查点操作的速度由稳定存储的可用带宽决定。

如果检查点操作失败，Flink 会丢弃该检查点并继续正常执行，因为之后的某一个检查点可能会成功。 虽然恢复时间可能更长，但是对于状态的保证依旧很有力。只有在一系列连续的检查点操作失败之后，Flink 才会抛出错误，因为这通常预示着发生了严重且持久的错误。

现在，我们来看“检查点操作已经完成，但故障紧随其后”的情况，如图 12 所示。

故障出现后，Flink 会重新拓扑（可能会获取新的执行资源），将输入流倒回到上一个检查点，然后恢复状态值并从该处开始继续计算。 在本例中，["a",2]、["a",2] 和["c",2] 这三条记录将被重播。图 13 展示了这一重新处理的过程。从上一个检查点开始重新计算，可以保证在剩下的记录被处理之后，得到的 map 算子的状态值与没有发生故障时的状态值一致。值得注意的是，输出流会含有重复的数据。具体来说，["a",2]、["a",4] 和 ["c",3] 在输出流中会出现两次。如果 Flink 将输出流写入特殊的输出系统（比如文件系统或者数据库），那么就可以避免这个问题，在节 3.4 中将进一步讨论。

Flink 检查点算法的正式名称是“异步屏障快照”（asynchronous barrier snapshotting）。该算法大致基于 Chandy-Lamport 分布式快照算法。

检查点（Checkpoint）由 Flink 自动生成，用来在故障发生时重新处理记录，从而修正状态。Flink 用户还可以通过另一个特性“主动地”管理状态版本，这个特性叫作保存点（Savepoint）。

保存点（Savepoint）与检查点（Checkpoint）的工作方式完全相同，只不过它由用户通过 Flink 命令行工具或者 Web 控制台手动触发，而不由 Flink 自动触发。 和检查点一样，保存点也被保存在稳定存储中。用户可以从保存点重启作业，而不用从头开始。保存点可以被视为作业在某一个特定时间点的快照（该时间点即为保存点被触发的时间点）。

保存点可用于应对流处理作业在生产环境中遇到的许多挑战：
(1) 应用程序代码升级：假设你在已经处于运行状态的应用程序中发现了一个 bug，并且希望之后的事件都可以用修复后的新版本来处理。通过触发保存点并从该保存点处运行新版本，下游的应用程序并不会察觉到不同（当然，被更新的部分除外）。
(2) Flink 版本更新：Flink 自身的更新也变得简单，因为可以针对正在运行的任务触发保存点，并从保存点处用新版本的 Flink 重启任务。
(3) 维护和迁移：使用保存点，可以轻松地“暂停和恢复”应用程序。这对于集群维护以及向新集群迁移的作业来说尤其有用。此外，它还有利于开发、测试和调试，因为不需要重播整个事件流。
(4) 假设模拟与恢复：在可控的点上运行其他的应用逻辑，以模拟假设的场景，这样做在很多时候非常有用。
(5) A/B 测试：从同一个保存点开始，并行地运行应用程序的两个版本，有助于进行 A/B 测试。

在节 3.2 中提到过故障出现并恢复后，输出流会含有重复的数据，怎么解决这个问题呢？也就是说，在将状态内容传送到输出存储系统的过程中，如何保证 exactly-once 呢？这叫作“端到端的一致性”。

端到端的一致性，本质上有两种实现方法：
(1) 第一种方法是在 sink 环节缓冲所有输出，并在 sink 收到检查点记录时，将输出“原子提交”到存储系统。这种方法保证输出存储系统中只存在有一致性保障的结果，并且不会出现重复的数据。从本质上说，输出存储系统会参与 Flink 的检查点操作。 要做到这一点，输出存储系统需要具备“原子提交”的能力。
(2) 第二种方法是急切地将数据写入输出存储系统，同时牢记这些数据可能是“脏”的，而且需要在发生故障时重新处理。如果发生故障，就需要将输出、输入和 Flink 作业全部回滚，从而将“脏”数据覆盖，并将已经写入输出的“脏”数据删除。注意，在很多情况下，其实并没有发生删除操作。例如，如果新记录只是覆盖旧纪录（而不是添加到输出中），那么“脏”数据只在检查点之间短暂存在，并且最终会被修正过的新数据覆盖。

值得注意的是， 这两种方法恰好对应关系数据库系统中的两种为人所熟知的事务隔离级别：已提交读（read committed）和未提交读（read uncommitted）。 “已提交读”保证所有读取（查询输出）都只读取已提交的数据，而不会读取中间、传输中或“脏”的数据。之后的读取可能会返回不同的结果，因为数据可能已被改变。“未提交读”则允许读取“脏”数据；换句话说，查询总是看到被处理过的最新版本的数据。

某些应用程序可以接受弱一点的语义，所以 Flink 提供了支持多重语义的多种内置输出算子，如支持“未提交读”语义的分布式文件输出算子。用户可以根据输出存储系统的能力和应用程序的需求在端到端的一致性的两种实现方法中做出选择。