22、Flink 背压下的 Checkpoint处理

1.概述

通常，对齐 Checkpoint 的时长主要受 Checkpointing 过程中的同步和异步两个部分的影响；但当 Flink 作业正运行在严重的背压下时，Checkpoint 端到端延迟的主要影响因子将会是传递 Checkpoint Barrier 到所有的算子/子任务的时间；可以通过高 alignment time and start delay metrics 观察到，解决方案如下：

解决背压问题。优化 Flink 作业，调整 Flink 或 JVM 参数，通过扩容。
减少 Flink 作业中缓冲在 In-flight 数据的数据量。
启用非对齐 Checkpoints。

上述选项并不是互斥的，可以组合使用。

2.缓冲区 Debloating

Flink 1.14 引入了缓冲区 Debloating 机制，用于自动控制在 Flink 算子/子任务之间缓冲的 In-flight 数据的数据量；可以通过将属性taskmanager.network.memory.buffer-debloat.enabled设置为true来启用。

此特性对对齐和非对齐 Checkpoint 都生效，并且在这两种情况下都能缩短 Checkpointing 的时间，不过 Debloating 的效果对于对齐 Checkpoint 更明显；当在非对齐 Checkpoint 情况下使用缓冲区 Debloating 时，好处是 Checkpoint 大小会更小，并且恢复时间更快 (需要保存和恢复的 In-flight 数据更少)。

3.非对齐 Checkpoints

a）概述

从 Flink 1.11开始，Checkpoint 可以是非对齐的；Unaligned checkpoints 包含 In-flight 数据（例如，存储在缓冲区中的数据）作为 Checkpoint State的一部分，允许 Checkpoint Barrier 跨越这些缓冲区，使 Checkpoint 时长变得与当前吞吐量无关，因为 Checkpoint Barrier 实际上已经不再嵌入到数据流当中。

如果 Checkpointing 由于背压导致周期非常的长，应该使用非对齐 Checkpoint，这样 Checkpointing 时间基本上就与端到端延迟无关；但是非对齐 Checkpointing 会增加状态存储的 I/O，因此当状态存储的 I/O 是整个 Checkpointing 过程当中真正的瓶颈时，不应当使用非对齐 Checkpointing。

启用非对齐 Checkpoint：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 启用非对齐 Checkpoint
env.getCheckpointConfig().enableUnalignedCheckpoints();

或者在 flink-conf.yml 配置文件中增加配置：

execution.checkpointing.unaligned: true

b）对齐 Checkpoint 超时

在启用非对齐 Checkpoint 后，依然可以通过编程的方式指定对齐 Checkpoint 的超时时间

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getCheckpointConfig().setAlignedCheckpointTimeout(Duration.ofSeconds(30));

或是在 flink-conf.yml 配置文件中配置：

execution.checkpointing.aligned-checkpoint-timeout: 30 s

在启动时，每个 Checkpoint 仍然是 aligned checkpoint，但是当全局 Checkpoint 持续时间超过 aligned-checkpoint-timeout 时，如果 aligned checkpoint 还没完成，那么 Checkpoint 将会转换为 Unaligned Checkpoint。

c）限制

并发 Checkpoint

Flink 当前并不支持并发的非对齐 Checkpoint；Savepoint 也不能与非对齐 Checkpoint 同时发生，因此它们将会花费更长的时间。

与 Watermark 的相互影响

非对齐 Checkpoint 在恢复的过程中改变了关于 Watermark 的一个隐式保证；目前，Flink 确保 Watermark 作为恢复的第一步，而不是将最近的 Watermark 存放在 Operator 中，以方便扩缩容。

在非对齐 Checkpoint 中，当恢复时，Flink 会在恢复 In-flight 数据后再生成 Watermark，如果 Pipeline 中使用了对每条记录都应用最新的 Watermark 的算子将会相对于使用对齐 Checkpoint 产生不同的结果；如果 Operator 依赖于最新的 Watermark 始终可用，解决办法是将 Watermark 存放在 OperatorState 中；此时 Watermark 应该使用单键 group 存放在 UnionState 以方便扩缩容。

与长时间运行的记录处理交互

尽管有未对齐的检查点，但 barrier 能够越过队列中的其它记录；如果当前记录需要大量时间进行处理，则此 barrier 的处理仍可能延迟，例如在窗口操作中同时触发多个定时器时。

当系统处理单个输入记录时被阻止，在等待多个网络缓冲区可用时，可能会出现 Flink 不能中断对单个输入记录的处理，未对齐的检查点必须等待当前处理的记录被完全处理；

原因要么是由于不适合单个网络缓冲区的大型记录的串行化，要么是在 flatMap 操作中，为一个输入记录生成许多输出记录；此时背压可以阻止未对齐的检查点，直到处理单个输入记录所需的所有网络缓冲区可用；当单个记录的处理需要一段时间时，也可能发生在其它情况下，导致检查点的时间可能高于预期。

某些数据分布模式不是检查点式的

有一部分包含属性的的连接无法与 Channel 中的数据一样保存在 Checkpoint 中；为了保留这些特性并且确保没有状态冲突或非预期的行为，非对齐 Checkpoint 对于这些类型的连接是禁用的，所有其他的交换仍然执行非对齐 Checkpoint。

点对点连接

目前没有任何对于点对点连接中有关数据有序性的强保证；由于数据已经被前置的 Source 或是 KeyBy 相同的方式隐式组织，一些用户会依靠这种特性在提供的有序性保证的同时将计算敏感型的任务划分为更小的块。

只要并行度不变，非对齐 Checkpoint(UC) 将会保留这些特性，但是如果加上 UC 的伸缩容，这些特性将会被改变。

针对如下任务：

如果想将并行度从 p=2 扩容到 p=3，那么需要根据 KeyGroup 将 KeyBy 的 Channel 中的数据划分到3个 Channel 中去；通过使用 Operator 的 KeyGroup 范围和确定记录属于某个 Key(group) 的方法可以实现；但对于 Forward 的 Channel，没有 KeyContext，Forward Channel 里也没有任何记录被分配了任何 KeyGroup；也无法计算它，因为无法保证 Key 仍然存在。

广播 Connections

广播 Connection 无法保证所有 Channel 中的记录都以相同的速率被消费，可能导致某些 Task 已经应用了与特定广播事件对应的状态变更，而其他任务则没有。