flink checkpoint 测试（实时计算-深入理解）

我是LakeShen，专注大数据技术分享，程序员经验分享，互联网科技分享。如果我的文章对你有帮助，希望你能点赞或者关注我，你的鼓励，就是我前进的最大动力。关注就完事了。

前言

为了保证程序的容错恢复以及程序启动时其状态恢复，几乎所有公司的实时任务都会开启 Checkpoint 或者触发 Savepoint 进行状态保存。为了使得用户更加理解这两点区别，本文结合 Flink 1.9 版本，重点讲述 Flink Checkpoint，Savepoint 相关概念以及注意事项，同时也会讲述实时任务启动时读取 Kafka 偏移量问题，使得用户能够更好的开发实时任务。

1. Checkpoint，Savepoint 异同

首先，为什么会在文章开头对这两点进行介绍，因为有时候用户在开发实时任务时，会对这两点产生困惑，所以这里直接开门见山对这两点进行讲解。

Flink Checkpoint 是一种容错恢复机制。这种机制保证了实时程序运行时，即使突然遇到异常也能够进行自我恢复。Checkpoint 对于用户层面，是透明的，用户会感觉程序一直在运行。Flink Checkpoint 是 Flink 自身的系统行为，用户无法对其进行交互，用户可以在程序启动之前，设置好实时程序 Checkpoint 相关参数，当程序启动之后，剩下的就全交给 Flink 自行管理。当然在某些情况，比如 Flink On Yarn 模式，某个 Container 发生 OOM 异常，这种情况程序直接变成失败状态，此时 Flink 程序虽然开启 Checkpoint 也无法恢复，因为程序已经变成失败状态，所以此时可以借助外部参与启动程序，比如外部程序检测到实时任务失败时，从新对实时任务进行拉起。

Flink 会定时在任务的 Source Task 触发 Barrier，Barrier是一种特殊的消息事件，会随着消息通道流入到下游的算子中。只有当最后 Sink 端的算子接收到 Barrier 并确认该次 Checkpoint 完成时，该次 Checkpoint 才算完成。所以在某些算子的 Task 有多个输入时，会存在 Barrier 对齐时间，我们可以在Web UI上面看到各个 Task 的Barrier 对齐时间

2.2 Flink Checkpoint 语义

Flink Checkpoint 支持两种语义：Exactly Once 和 At least Once，默认的 Checkpoint 模式是 Exactly Once. Exactly Once 和 At least Once 具体是针对 Flink 状态 而言。具体语义含义如下：

Exactly Once 含义是：保证每条数据对于 Flink 的状态结果只影响一次。打个比方，比如 WordCount程序，目前实时统计的 "hello" 这个单词数为5，同时这个结果在这次 Checkpoint 成功后，保存在了 HDFS。在下次 Checkpoint 之前， 又来2个 "hello" 单词，突然程序遇到外部异常容错自动回复，从最近的 Checkpoint 点开始恢复，那么会从单词数 5 这个状态开始恢复，Kafka 消费的数据点位还是状态 5 这个时候的点位开始计算，所以即使程序遇到外部异常自我恢复，也不会影响到 Flink 状态的结果。

At Least Once 含义是：每条数据对于 Flink 状态计算至少影响一次。比如在 WordCount 程序中，你统计到的某个单词的单词数可能会比真实的单词数要大，因为同一条消息，你可能将其计算多次。

Flink 中 Exactly Once 和 At Least Once 具体是针对 Flink 任务状态而言的，并不是 Flink 程序对其处理一次。举个例子，当前 Flink 任务正在做 Checkpoint，该次Checkpoint还么有完成，该次 Checkpoint 时间端的数据其实已经进入 Flink 程序处理，只是程序状态没有最终存储到远程存储。当程序突然遇到异常，进行容错恢复，那么就会从最新的 Checkpoint 进行状态恢复重启，上一部分还会进入 Flink 系统处理：

上图中表示，在进行 chk-5 Checkpoint 时，突然遇到程序异常，那么会从 chk-4 进行恢复，那么之前chk-5 处理的数据，会再次进行处理。

Exactly Once 和 At Least Once 具体在底层实现大致相同，具体差异表现在 Barrier 对齐方式处理：

如果是 Exactly Once 模式，某个算子的 Task 有多个输入通道时，当其中一个输入通道收到 Barrier 时，Flink Task 会阻塞处理该通道，其不会处理这些数据，但是会将这些数据存储到内部缓存中，一旦完成了所有输入通道的 Barrier 对齐，才会继续对这些数据进行消费处理。

对于 At least Once,同样针对某个算子的 Task 有多个输入通道的情况下，当某个输入通道接收到 Barrier 时，它不同于Exactly Once,At Least Once 会继续处理接受到的数据，即使没有完成所有输入通道 Barrier 对齐。所以使用At Least Once 不能保证数据对于状态计算只有一次影响。

2.3 Flink Checkpoint 参数配置及建议

1. 当 Checkpoint 时间比设置的 Checkpoint 间隔时间要长时，可以设置 Checkpoint 间最小时间间隔 。这样在上次 Checkpoint 完成时，不会立马进行下一次 Checkpoint，而是会等待一个最小时间间隔，然后在进行该次 Checkpoint。否则，每次 Checkpoint 完成时，就会立马开始下一次 Checkpoint，系统会有很多资源消耗 Checkpoint。

2. 如果Flink状态很大，在进行恢复时，需要从远程存储读取状态恢复，此时可能导致任务恢复很慢，可以设置 Flink Task 本地状态恢复。任务状态本地恢复默认没有开启，可以设置参数`state.backend.local-recovery`值为`true`进行激活。

3. Checkpoint保存数，Checkpoint 保存数默认是1，也就是保存最新的 Checkpoint 文件，当进行状态恢复时，如果最新的Checkpoint文件不可用时(比如HDFS文件所有副本都损坏或者其他原因)，那么状态恢复就会失败，如果设置 Checkpoint 保存数2，即使最新的Checkpoint恢复失败，那么Flink 会回滚到之前那一次Checkpoint进行恢复。考虑到这种情况，用户可以增加 Checkpoint 保存数。

4. 建议设置的 Checkpoint 的间隔时间最好大于 Checkpoint 的完成时间。

下图是不设置 Checkpoint 最小时间间隔示例图，可以看到，系统一致在进行 Checkpoint，可能对运行的任务产生一定影响：

3. Flink Savepoint

3.1 Flink Savepoint 原理

Flink Savepoint 作为实时任务的全局镜像，其在底层使用的代码和Checkpoint的代码是一样的，因为Savepoint可以看做 Checkpoint在特定时期的一个状态快照。

Flink 在触发Savepoint 或者 Checkpoint时，会根据这次触发的类型计算出在HDFS上面的目录:

如果类型是 Savepoint，那么 其 HDFS 上面的目录为：Savepoint 根目录 savepoint-jobid前六位 随机数字，具体如下格式：

Checkpoint 目录为 chk-checkpoint ID,具体格式如下：

一次 Savepoint 目录下面至少包括一个文件，既 _metadata 文件。当然如果实时任务某些算子有状态的话，那么在 这次 Savepoint 目录下面会包含一个 _metadata 文件以及多个状态数据文件。_metadata 文件以绝对路径的形式指向状态文件的指针。

社区方面，在以前的 Flink 版本，当用户选择不同的状态存储，其底层状态存储的二进制格式都不相同。针对这种情况，目前 [FLIP-41](FLIP-41: Unify Binary format for Keyed State) 对于 Keyed State 使用统一的二进制文件进行存储。这里的 Keyed State 主要是针对 Savepoint 的状态，Checkpoint 状态的存储可以根据具体的状态后端进行存储，允许状态存储底层格式的差异。对于 Savepoint 状态底层格式的统一，应用的状态可以在不同的状态后端进行迁移，更方便应用程序的恢复。重做与状态快照和恢复相关的抽象，当实现实现新状态后端时，可以降低开销，同时减少代码重复。

3.2 Flink Savepoint 触发方式

Flink Savepoint 触发方式目前有三种：

1. 使用 flink savepoint 命令触发 Savepoint,其是在程序运行期间触发 savepoint。

2. 使用 flink cancel -s 命令，取消作业时，并触发 Savepoint。

3. 使用 Rest API 触发 Savepoint，格式为：**/jobs/:jobid /savepoints**

3.3 Flink Savepoint 注意点

1. 使用 flink cancel -s 命令取消作业同时触发 Savepoint 时，会有一个问题，可能存在触发 Savepoint 失败。比如实时程序处于异常状态(比如 Checkpoint失败)，而此时你停止作业，同时触发 Savepoint,这次 Savepoint 就会失败，这种情况会导致，在实时平台上面看到任务已经停止，但是实际实时作业在 Yarn 还在运行。针对这种情况，需要捕获触发 Savepoint 失败的异常，当抛出异常时，可以直接在 Yarn 上面 Kill 掉该任务。

2. 使用 DataStream 程序开发时，最好为每个算子分配 `uid`,这样即使作业拓扑图变了，相关算子还是能够从之前的状态进行恢复，默认情况下，Flink 会为每个算子分配 `uid`,这种情况下，当你改变了程序的某些逻辑时，可能导致算子的 `uid` 发生改变，那么之前的状态数据，就不能进行复用，程序在启动的时候，就会报错。

3. 由于 Savepoint 是程序的全局状态，对于某些状态很大的实时任务，当我们触发 Savepoint，可能会对运行着的实时任务产生影响，个人建议如果对于状态过大的实时任务，触发 Savepoint 的时间，不要太过频繁。根据状态的大小，适当的设置触发时间。

4. 当我们从 Savepoint 进行恢复时，需要检查这次 Savepoint 目录文件是否可用。可能存在你上次触发 Savepoint 没有成功，导致 HDFS 目录上面 Savepoint 文件不可用或者缺少数据文件等，这种情况下，如果在指定损坏的 Savepoint 的状态目录进行状态恢复，任务会启动不起来。

5. 总结

本文没有过多的讲述源码，考虑大家的都能够读懂，其语言竟可能通俗一一点。如果有需要改进的地方，希望大家能够指出。后续我会不断的和大家一起大数据相关的技术，和大家一起交流学习。

参考资料

1. FLIP-41: Unify Binary format for Keyed State

2. FLIP-47: Checkpoints vs. Savepoints

3. Apache Kafka Connector

4. Apache Flink 1.9 Documentation: Savepoints

5. Apache Flink 1.9 Documentation: Checkpoints

6. Apache Flink 1.8 Documentation: Checkpointing

,