超越Storm，SparkStreaming——Flink如何实现有状态的计算

流式计算分为无状态和有状态两种情况。无状态计算观察每个独立的事件，Storm就是无状态的计算框架，每一条消息来了以后和前后都没有关系，一条是一条。比如我们接收电力系统传感器的数据，当电压超过240v就报警，这就是无状态的数据。但是如果我们需要同时判断多个电压，比如三相电路，我们判断三相电都高于某个值，那么就需要将状态保存，计算。因为这三条记录是分别发送过来的。

Storm需要自己实现有状态的计算，比如借助于自定义的内存变量或者redis等系统，保证低延迟的情况下自己去判断实现有状态的计算，但是Flink就不需要这样，而且作为新一代的流处理系统，Flink非常重视。

一致性

其实就是消息传递的正确性。在流处理中，一致性分为 3 个级别。

• at-most-once：最多一次，可能会丢失。

• at-least-once：最少一次，可能会重复，而计算的时候可能就会多次运算影响结果。

• exactly-once：恰好保证一次，这样得到的结果是最准确的。

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

但是，Flink解决了这种问题。

检查点机制

检查点是 Flink 最有价值的创新之一，因为它使 Flink 可以保 证 exactly-once，并且不需要牺牲性能。

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 算子在接收到每个元素后， 将输入记录的第二个字段的数据加到现有总数中，再将更新过的元素发射出去。

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

如果输入流来自消息传输系统Kafka，这个相互隔离的位置就是偏移量。

检查点屏障像普通记录一样在算子之间流动。当 map 算子处理完前 3 条记录 并收到检查点屏障时，它们会将状态以异步的方式写入稳定存储.

当没有出现故障时，Flink 检查点的开销极小，检查点操作的速度由稳定存储的可用带宽决定。

如果检查点操作失败，Flink 会丢弃该检查点并继续正常执行，因为之后的 某一个检查点可能会成功。

在这种情况下，Flink 会重新拓扑（可能会获取新的执行资源），将输入流 倒回到上一个检查点，然后恢复状态值并从该处开始继续计算。

Flink 将输入流倒回到上一个检查点屏障的位置，同时恢复 map 算子的状态值。 然后，Flink 从此处开始重新处理。这样做保证了在记录被处理之后，map 算子的状 态值与没有发生故障时的一致.

Flink 检查点算法的正式名称是异步屏障快照（asynchronous barrier snapshotting）。

保存点

1062 状态版本控制

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

保存点与检查点的工作方式完全相同，只不过它由用户通过 Flink 命令行工 具或者 Web 控制台手动触发，而不由 Flink 自动触发，用户可以从保存点重启作业，而不用从头开始。对保存点的另一种理解是，它在明确的时间点保存应用程序状态的版本。

在图中，v.0 是某应用程序的一个正在运行的版本。我们分别在 t1 时刻和 t2 时刻触发了保存点。因此，可以在任何时候返回到这两个时间点，并且重 启程序。更重要的是，可以从保存点启动被修改过的程序版本。举例来说， 可以修改应用程序的代码（假设称新版本为 v.1），然后从t1 时刻开始运行 改动过的代码。

使用保存点更新Flink 应用程序的版本。新版本可以从旧版本生成的一个 保存点处开始执行.

端到端的一致性

在该应用程序架构中，有状态的Flink 应用程序消费来自消息队列的数据， 然后将数据写入输出系统，以供查询。

输入数据来自Kafka，在将状态内容传送到输出存储系统的过程中，如何保证 exactly-once 呢？这 叫作端到端的一致性。本质上有两种实现方法，用哪一种方法则取决于输 出存储系统的类型，以及应用程序的需求。

(1) 第一种方法是在 sink 环节缓冲所有输出，并在 sink 收到检查点记录时， 将输出“原子提交”到存储系统。这种方法保证输出存储系统中只存在 有一致性保障的结果，并且不会出现重复的数据。从本质上说，输出存 储系统会参与 Flink 的检查点操作。要做到这一点，输出存储系统需要 具备“原子提交”的能力。

(2) 第二种方法是急切地将数据写入输出存储系统，同时牢记这些数据可能 是“脏”的，而且需要在发生故障时重新处理。如果发生故障，就需要将 输出、输入和 Flink 作业全部回滚，从而将“脏”数据覆盖，并将已经写 入输出的“脏”数据删除。注意，在很多情况下，其实并没有发生删除 操作。例如，如果新记录只是覆盖旧纪录（而不是添加到输出中），那么 “脏”数据只在检查点之间短暂存在，并且最终会被修正过的新数据覆盖。

根据输出存储系统的类型，Flink 及与之对应的连接器可以一起保证端到端 的一致性，并且支持多种隔离级别。

更多Flink相关文章：

Streaming-大数据的未来

实时计算大数据处理的基石-Google Dataflow

数据架构的未来——浅谈流处理架构

穿梭时空的实时计算框架——Flink对时间的处理

Flink快速入门--安装与示例运行

大数据实时处理的王者-Flink

更多实时计算,Flink,Kafka等相关技术博文，欢迎关注实时流式计算：