高效稳定的通用增量 Checkpoint 详解之二：性能分析评估

2023-03-16
浙江
本文字数：9020 字
阅读完需：约 30 分钟

作者：雷颜菲、夏瑞、俞航翔、梅源｜阿里云 Flink 存储引擎团队

摘要：我们在“Flink 1.15 新功能架构解析：高效稳定的通用增量 Checkpoint”【1】一文介绍了通用增量 Checkpoint 的原理和背后的思考以及执行性能、空间放大等方面的初步测试结果。该功能在 Flink 1.16 中经过优化，已达到生产可用状态。本文将从理论和实验两个部分详细论述通用增量 Checkpoint 的收益与开销，并分析其适用场景。

一、概述

在进行详细的性能分析之前，我们简单回顾一下通用增量 Checkpoint 的设计思考。Flink Checkpoint 过程包括同步刷盘和异步上传文件两个部分，一个算子的 Checkpoint 需要算子的所有并发完成异步过程并确认成功后才算完成。因此，在大规模作业中，Checkpoint 异步耗时通常是影响 Checkpoint 稳定性和延迟的瓶颈点。异步上传文件耗时部分有两个不确定性因素：

异步上传的文件大小依赖于状态存储的具体实现
异步过程需要等到同步过程结束才能开始

第一个问题，以 RocksDB 为例，虽然 Flink 对 RocksDB 也支持增量 Checkpoint，但是异步上传文件的数据量会受到 RocksDB Compaction 影响，因为在 Compaction 发生后可能会导致大量相对较大的新文件需要重新上传。目前，RocksDB Compaction 触发不具有确定性，大规模作业在每次 Checkpoint 时，会经常出现部分节点异步耗时过长而影响 Checkpoint 的完成速度。

第二个问题，是因为在原先的 Checkpoint 架构下，同步快照结束前是没法准备好需要上传的文件的，因此异步过程需要等到同步过程结束后才能开始。这种情况下，两个 Checkpoint 间隔之间很大一部分时间占比就被浪费了。如果需要上传的状态比较大，会在很短时间内对 CPU 和网络产生较大的压力。

通用增量 Checkpoint 通过引入 State Changelog（状态变化日志）实现了在架构上 Checkpoint 执行过程与状态存储快照的解耦，很好的解决了上述两个问题。在新架构下，状态更新不仅会更新状态表，而且会记录状态的变化日志。状态表会周期性持久化到远端存储，称为物化过程（Materialization）。这个过长通常周期比较长（比如 10 分钟），并独立于 Flink 引擎的 Checkpoint 过程。另一方面，状态的变化日志也会持续上传到远端持久存储，并且在执行 Checkpoint 时 Flush 剩余全部日志【1】。

这样的设计就比较好的解决了前面提到的两个问题：通过将 Checkpoint 过程和物化过程解耦，可以让异步上传的文件大小变得稳定；同时因为状态变化更新是持续的，所以我们可以在触发 Checkpoint 之前就一直持续的上传增量的更新日志，所以在 Flush 以后我们实际需要上传的数据量就变得很小。

因此，使用通用增量 Checkpoint 可以更稳定快速地完成 Checkpoint；从而达到更稳定更低的端到端延迟、更少的数据回追和更平稳的资源消耗；与此同时在开销方面，通用增量 Checkpoint 带来的极限性能下降几乎可以忽略，所带来的额外资源消耗也是可预估并且可控的。本文将从这几个方面详细分析通用增量 Checkpoint 的优势和代价，并分享相应的实验结论。

二、收益与开销

2.1 更稳定更低的端到端延迟

端到端延迟是 Flink 流处理系统非常重要的性能指标。端到端延迟是指开始处理输入数据到输出该数据产生的结果所需的时间，而更低的端到端延迟则意味着下游系统可以更快的得到更新结果。Flink 基于 Checkpoint 机制可以提供 Exactly-once 语义保障，当 Source 支持重放、Sink 支持事务后，可以进一步提供端到端的 Exactly-once 语义保障。如图 1 所示，Checkpoint 完成后，事务型 Sink 才可以更新提交位点。因此，Checkpoint 执行的速度越快、间隔越短，事务型 Sink 就可以更频繁地提交，从而为作业提供更低的端到端延迟。

图 1：事务型 Sink 的端到端延迟示意图

通用增量 Checkpoint 可以带来更稳定更低的端到端延迟。如前所述，通用增量 Checkpoint 机制通过将状态存储快照与 Flink 引擎 Checkpoint 执行过程解耦，可以解决不同状态存储实现（例如 RocksDB Compaction）在 Checkpoint 期间异步状态上传量对 Checkpoint 异步耗时的影响。同时因为可以持续上传 State Changelog，极大降低了 Checkpoint 触发后需要上传的数据量，Checkpoint 触发后只需要上传少量增量日志即可，从而大幅提升了 Checkpoint 完成的稳定性和速度，降低作业端到端延迟，并且让延迟稳定在一定范围内。

2.2 更少的数据回追

在没有使用 Exactly-once Sink 的情况下，数据回追会导致重复输出。如果下游不支持完全去重，会在一定程度上影响数据正确性。因此，如何保证流计算系统在故障恢复后回追更少的数据，尽快恢复到故障前的状态是流计算系统设计需要考虑的问题。Flink 在从故障恢复时，首先会定位到最新的完整可用的 Checkpoint，每个算子节点基于这个 Checkpoint 重建本地状态后，从该 Checkpoint 保留的输入位点开始回追数据。需要回追的数据越少，数据重复输出也少，恢复也越快。

图 2：以 RocksDB 状态存储为例，作业失败后，数据回追示意图（Checkpoint 简写为 CHK）。

通用增量 Checkpoint 可以减少作业恢复时的重放数据量。如图 2 所示，数据回追量是作业失败时间点到最近一次 Checkpoint 位点的数据量。由于通用增量 Checkpoint 可以加快 Checkpoint 的完成速度，因此可以将 Checkpoint 做得更高频，需要回追的数据也会变得更少，减少数据重复输出，加快恢复。这对于对数据重复度有较高要求的业务来说是很关键的提升。

2.3 更平稳的资源消耗

Flink 快照过程会引起 Flink 作业 CPU 和网络资源使用出现尖刺，这点对资源量消耗多的大状态作业尤为明显。我们仍以 RocksDB 作为状态存储为例，Flink Checkpoint 过程会间接地触发 RocksDB Compaction 的执行条件，使 CPU 使用量在短时间内激增，形成尖刺（Compaction 是 RocksDB 状态存储主要的 CPU 消耗来源之一）。另外，Checkpoint 触发快照制作之后，作业中全部算子并发的状态存储会集中在这个时间段内向远端存储上传数据，导致网络资源使用量在短时间内也出现激增，很容易达到网络带宽上限。如图 3 所示，Checkpoint 触发之后，所有算子中的 RocksDB 状态存储几乎同时向远端存储上传状态。对于大状态作业来说，CPU 尖刺会使得资源预留估算和实际使用出现较大偏差，影响作业正常运行。虽然现在集群部署都采用容器化，但出于资源利用率的考虑，很多资源还是会共享的（比如 CPU 和网络），因此 CPU 和网络尖刺会影响集群内其他作业的正常运行，进一步影响整个集群的稳定性。

图 3：RocksDB 增量 Checkpoint 的 RocksDB 状态存储上传过程

通用增量 Checkpoint 可以有效降低作业的 CPU 和网络流量的消耗，提升 Checkpoint 稳定性，平滑集群 CPU 和网络资源使用。一方面，通用增量 Checkpoint 通过将增量状态变化日志（State Changelog）持续上传到远端持久化存储，可以将网络流量的消耗均摊在整个 Checkpoint 过程中；另一方面 Checkpoint 过程与存储后端快照过程解耦可以使得 RocksDB 低频、错峰执行物化过程，解决 RocksDB 频繁执行 Compaction 操作和集中向远端存储上传数据的问题，从而降低 CPU 和网络资源使用量和波动性。如图 4 所示，不同算子的 RocksDB 文件上传过程均匀打散到物化间隔内，使得 RocksDB 的物化过程可以低频、错峰。

图 4：通用增量 Checkpoint 的 RocksDB 状态存储上传过程

2.4 可控的性能与资源开销

状态双写对性能影响很小（2～3%）

通用增量 Checkpoint 需要双写状态数据（写状态存储和 State Changelog），因此有额外的性能消耗。如图 5 所示，通用增量 Checkpoint 引入 Durable Short-term Log（DSTL，短存 Log）管理 State Changelog，状态变化日志需要同时写入状态存储和 DSTL 中。实测中，在网络带宽不是瓶颈的情况下，其对极限吞吐量（作业能达到的最高 TPS）仅有很小（2～3%）的影响。状态变化日志以 log append 的形式顺序写入攒批上传，影响微乎其微。

图 5：以 RocksDB 状态存储为例，通用增量 Checkpoint 状态双写示意图

额外的存储和网络流量费用可控且可估算

存储空间方面，引入通用增量 Checkpoint 后，Checkpoint 由两部分组成：状态存储的物化部分和 DSTL 中的增量状态变化日志。DSTL 在状态存储后端完成物化且 Checkpoint 过期（subsume）后可以删除 State Changelog 对应的部分。因此理论上来说，在状态存储即将完成物化、清理 State Changelog 之前，DSTL 占用的远端额外存储空间最大。在作业稳定运行的情况下，算子的读写模式基本固定，因此可以估算出在一个物化间隔中写入 DSTL 的 State Changelog 数据规模，并通过参数调整其大小。例如，通过减小物化间隔，可以降低 DSTL 占用远端存储的最大值。如图 6 所示，在相邻的两个物化触发点之间，Checkpoint 中状态存储的物化部分保持不变，而 State Changelog 部分持续增加。因此，State Changelog 累积最大值出现在清理 State Changelog 前。图 6 中，CHK-7 是 MID-1 之后、MID-2 之前的最后一个 Checkpoint，包含了 t₁ 至 t₇ 之间全部的 State Changelog，此时 State Changelog 累积值最大。

图 6：通用增量 Checkpoint 中，算子-1 的全量 Checkpoint 大小变化示意图

另外值得一提的是远端存储的费用占流式作业计费中的比例很小。以阿里云 OSS 服务为例，其 1 GB 存储空间的每月费用是 0.12 RMB 左右【2】。单算子需要的 Checkpoint 存储规模通常在 10 GB 以内，一个月折算 1.2 RMB，远低于算子的计算资源费用。

网络流量方面，State Changelog 会有额外的网络流量开销，但这个部分因为状态存储后端快照频率（物化过程）降低会有一定的抵消，并且一般 Checkpoint 使用的是内网流量，费用几乎可以忽略不计。

三、收益与开销实验结果与分析

得益于 Checkpoint 完成耗时的缩短，作业可以达到更低的端到端延迟，同时回追更少的数据，加快作业恢复；得益于 Checkpoint 完成的更稳定，整个集群的资源消耗可以更平稳；得益于通用增量 Checkpoint 机制的轻量可控，其对极限吞吐量和空间资源的开销是我们可以预估且接受的。

因此，我们分成六组实验验证通用增量 Checkpoint 机制的收益与开销。

3.1 测试环境与配置

测试基于 Flink 1.16 版本，使用 K8s application 部署模式，state backend 选择 RocksDB（FRocksDB 6.20.3 版本）并开启 state backend 层面的增量 Checkpoint，远端存储使用阿里云 OSS 标准存储。Flink TM 和 JM 的配置使用 1 Core 4 GB RAM。测试作业默认并发度为 50。方便起见，下文将通用增量 Checkpoint （General Incremental Checkpoint）简写为 GIC。

为了测量极致情况下 Checkpoint 完成的速度和稳定性，Checkpoint 间隔设置为 1 秒，从而端到端延迟和失效后的数据回追量可降为秒级。通用增量 Checkpoint 的物化间隔设置为 3 分钟，该设置与 RocksDB 增量 Checkpoint 的常用 Checkpoint 间隔对齐。

测试作业涵盖了实际应用中常见的两类有状态作业，分别是指标聚合型和明细型作业。

指标聚合型作业的特征是包含有聚合函数，例如，min、max、count 等。本测试选择 WordCount 作业作为典型的指标聚合型作业。我们将 WordCount 的单并发状态规模控制在 500 MB 左右（中等状态大小），每一个 Word 的长度设置为 200 Bytes（聚合类型作业 Key 的长度）。
明细记录型作业的特征是需要保存一段完整的原始数据流，例如，join、window。本测试选择 Sliding window 作业（简写为 Window）作为典型的明细记录型作业。我们将 Window 的单并发状态规模控制在 2 GB 左右，Window 中每一个 record 的大小设置为 100 Bytes。

3.2 更快速的 Checkpoint 制作

本实验使用 Checkpoint 完成数量和完成时间来衡量 Checkpoint 制作速度。

从表 1 中可以看到，开启通用增量 Checkpoint，作业可以更高频地完成 Checkpoint。在开启 GIC 之后，WordCount 作业的 Checkpoint 完成数量提升了 4.23 倍，Sliding Window 作业的 Checkpoint 完成数量提升了近 40 倍。

从表 2 中可以看到，开启通用增量 Checkpoint 可以极大加速 Checkpoint 的完成。WordCount 和 Window 作业的 Checkpoint 平均完成时间分别下降 89.7% 和 98.8%。在状态规模较大的 Window 测试作业中，开启 GIC，Checkpoint 完成时间可以从分钟级降为秒级。

表 1：12 小时内成功完成 Checkpoint 数量

表 2：开启/关闭通用增量 Checkpoint，Checkpoint 完成时间对比

Checkpoint 速度提升的主要原因是快照制作过程中异步上传到远端存储的数据量大幅降低。从图 7 和图 8 中可以看到，开启通用增量 Checkpoint，增量快照的大小可以降低超过 95%。这是因为 State Changelog 会持续写入到 DSTL 中。在触发 Checkpoint 的时候，通用增量 Checkpoint 只需要把当前还未上传的 State Changelog （小于 5MB）写入到远端存储，因此通用增量 Checkpoint 在 Checkpoint 期间异步上传远端存储的数据量非常少，大幅提升了 Checkpoint 的制作速度。更快速的 Checkpoint 可以支持更短的 Checkpoint 间隔，从而降低作业运行时的端到端时延。为了保证数据处理的 Exactly-once 语义，事务性 Sink 算子需要在 Checkpoint 完成之后才可以触发事务提交。因此，通用增量 Checkpoint 可以将事务性算子的端到端时延从分钟级降至秒级。

图 7：WordCount 运行时增量 Checkpoint 大小

图 8：Window 运行时增量 Checkpoint 大小

3.3 更稳定的 Checkpoint 制作

本实验使用 Checkpoint 完成时间的波动范围来衡量 Checkpoint 制作的稳定性。

从图 9 和图 10 中可以看到，开启通用增量 Checkpoint，Wordcount 和 Window 作业的 Checkpoint 完成时间能够稳定在 1 - 5 秒内。相反没有 GIC 时，RocksDB 增量快照的完成时间波动范围大很多。在 Sliding Window 的情况，波动范围超过 100 秒，极端情况下会超过 400 秒。

图 9：WordCount 作业 Checkpoint 完成时间

图 10：Window 作业 Checkpoint 完成时间

Checkpoint 稳定性提升的主要原因是 Checkpoint 执行过程与存储后端快照解耦。针对本实验使用的 RocksDB 存储后端，RocksDB 增量 Checkpoint 依赖于 RocksDB 的快照机制，而 RocksDB 内部的 Compaction 会影响 RocksDB 快照产生的增量快照大小。而 RocksDB Compaction 触发受到多种因素影响，具有随机性，这可能会导致 RocksDB 增量快照大小波动范围很大。在开启通用增量 Checkpoint 之后，Checkpoint 期间只需上传增量 State Changelog，有效规避了 RocksDB Compaction 对增量快照制作的负面影响。

3.4 更少的数据回追

本测试使用 Window 作业，并使用 Kafka Source 恢复后的业务延时【3】来衡量数据回追量。在作业运行时，通过向 Kafka 注入指定的 record，触发作业失效。如图 11 所示，开启通用增量 Checkpoint，数据回追时间从 180 秒减少到 105 秒，减少 41.7%（状态下载时间由于开启了 local recovery 可以忽略）。数据回追量下降的主要原因是 Checkpoint 做得更快让 Checkpoint Interval 可以设得更短。

图 11：Window 作业中 Kafka Source 的业务延迟

值得一提的是 Source 的延迟和作业恢复速度也有关系，表 3 展示了 Window 作业在恢复中各个阶段的 P99 耗时。通用增量 Checkpoint 恢复 State Changelog 需要额外使用 16 秒左右。由于开启了 local recovery，无需从远端存储下载状态文件，所以下载部分时间可以忽略。对于 Window 作业，根据 3.2 节中的数据结果，Checkpoint 间隔可以从分钟级降低到秒级。因此，综合考虑上述两个方面，通用增量 Checkpoint 以增加微小的额外恢复时间为代价（16 秒），将重放数据量从分钟级降到秒级，因此整体上有更快的恢复速度。

表 3：Window 作业恢复的各个阶段 P99 耗时

3.5 更平稳的资源消耗

本测试使用 WordCount 作业，单并发状态大小设置在 2GB 左右，对比测试整个作业的 CPU 使用量和网络流量实时变化情况。通过调整 Checkpoint 间隔和通用增量 Checkpoint 的开关设置三个实验：

关闭 GIC，间隔 10min：以 10min 为 Checkpoint 间隔，关闭通用增量 Checkpoint
关闭 GIC，间隔 10s：以 10s 为 Checkpoint 间隔，关闭通用增量 Checkpoint
开启 GIC，间隔 10s：以 10s 为 Checkpoint 间隔，开启通用增量 Checkpoint

图 12：作业的全部 TM CPU 总使用率

图 13：作业的网络流量

结合图 14 对比三个实验在 CPU 和网络流量上的数据，可以发现：

关闭 GIC，间隔 10min 的作业，其 CPU 使用率和网络流量有非常明显的尖峰，每次的峰值数据相比平均值有数十倍的增长，且该波动具有一定周期性。周期性波动主要是由于所有节点在 Checkpoint 同步阶段会触发 RocksDB 强制 Flush，进而可能导致多个并发节点同时 Compaction，造成 CPU 和网络流量明显增加。这种周期性会影响整个集群作业的稳定性，也让集群资源的估算变得困难，导致资源浪费。
关闭 GIC，间隔 10s 的作业，其 CPU 使用率和网络流量的峰值相对关闭 GIC，间隔 10min 的作业下降了 37.8% 和 34.9%，但其均值提升了 128.9% 和 730.2%，且其波动性依然较大。峰值下降是因为 Checkpoint 间隔变短，在一个 Checkpoint 间隔期间消费的数据量也随之变少，因此每次 Checkpoint 时参与 Compaction 的数据量也变少，要上传的数据量也随之变少；均值提升是因为 Compaction 的频次会随着 Checkpoint 间隔的减少而增加。
开启 GIC，间隔 10s 的作业，其 CPU 使用率和网络流量相对另外两者明显稳定和减少很多，其峰值相对关闭 GIC，间隔 10min 的作业下降了 62.2% 和 69.4%，相对关闭 GIC，间隔 10s 的作业下降了 39.3% 和 53.0%；其均值相对关闭 GIC，间隔 10s 的作业下降了 46.8% 和 67.7%。峰值和均值下降主要来源于 Checkpoint 间隔减少和前文提到的错峰的物化过程、运行时的持续上传。

图 14：三个实验的峰值/均值对比结果

如 1.3 节所述，通用增量 Checkpoint 通过低频、错峰的 Checkpoint 特性，让集群的资源使用变得非常平稳。在降低 Checkpoint 间隔的同时开启通用增量 Checkpoint，可以在降低 Flink 作业的 CPU 和网络流量的消耗的同时，提升 Flink 作业的 CPU 利用率和网络流量的稳定性。对单个作业来说，通用增量 Checkpoint 可以有效降低计算和流量成本，同时在资源受限的云原生环境下，其更稳定的 CPU 利用率和网络流量也可以让作业的 TPS 更加稳定；对整个 Flink 集群来说，稳定的 CPU 利用率和网络流量也使得作业间的资源竞争更加可控，进一步提升 Flink 集群的稳定性，同时也更易于估算集群资源，避免资源浪费。

3.6 微小的极限吞吐量影响

本测试使用 WordCount 作业，加大 Source 的流量，在作业达到反压的情况下，对比测试不同 Key Length 和 State Size 下开启/关闭通用增量 Checkpoint 后单个算子的极限吞吐量。

值得注意的是，Flink 1.16 的通用增量 Checkpoint 在序列化上有额外的开销，经过 FLINK-30345 【4】的优化后，由表 4 可以看到，算子的极限吞吐量差异稳定在 3% 以内。进一步当开启 Local Recovery 后，从表 5 中可以看到，算子的极限吞吐量差异在 5% 以内。

因此，开启通用增量 Checkpoint 后，双写（state changelog）和三写（local recovery）对极限性能的影响都很小，相对其他方面的提升，基本可以忽略不计。

表 4：Flink 1.16 优化后开启/关闭通用增量 Checkpoint 后极限吞吐量对比

表 5：Flink-1.16 优化后开启 Local Recovery 下开启/关闭通用增量 Checkpoint 极限吞吐量对比

3.7 可预估的远端存储空间开销

在开启通用增量 Checkpoint 后，Flink Checkpoint 包括物化的状态表部分加上增量的 State Changelog，因此需要额外的远端存储空间来存储 State Changelog。【1】中“Benchmark 测试结果分析”一节有实验得到的具体数据，这里我们将从理论入手，根据作业的处理逻辑和参数，对增加的远端存储空间进行分析和估算。

增加的空间主要来自 State Changelog。前面分析过，在状态存储即将完成物化、清理 State Changelog 之前，DSTL 占用的远端额外存储空间最大。因此 DSTL 的最大值近似为相邻物化之间的 State Changelog 总量，计算公式如下：

1 全量 Checkpoint 增加量最大值 ≈ Changelog 写操作速率 × 单次写大小 × 物化间隔

复制代码

我们使用 Sliding Window 作业验证上述计算公式，Window 大小和滑动长度的比值设置为 10 和 5。根据 Sliding Window 算子（Jave Flink 实现【5】）的执行逻辑，其全量 Checkpoint 增加量的预估值为：

1 全量 Checkpoint 增加量 ≈ 2 (Window + Timer 写操作速率) × (Window + Timer 单次写大小) × 物化间隔

复制代码

如图 15 和图 16 所示，实际作业的全量 Checkpoint 增加量和上述公式计算的预估增量基本一致。同时，可以发现 Window 作业的全量 Checkpoint 增加量具有以下特点：当“Window 大小 / 滑动长度”固定的时候，全量 Checkpoint 增加量基本保持不变。在比例为 10 的情况下，增加量约为 1.74 GB；在比例为 5 的时候，增加量约为 887.4 MB。这是因为“Window 大小 / 滑动长度”决定了公式中的“写操作速率”，同时，该作业的单次写大小和物化间隔也是固定的，因此全量 Checkpoint 增加量也是固定的。

图 15：Window 大小 / 滑动长度 = 10，全量 Checkpoint 大小

图 16：Window 大小 / 滑动长度 = 5，全量 Checkpoint 大小

全量 Checkpoint 增加使得作业需要使用更多的远端存储空间，但在实际中对作业整体计费影响很小。以阿里云 OSS 服务为例，存储计费是 0.12 RMB/GB，单算子需要的 Checkpoint 存储规模通常在 10 GB 以内，一个月折算 1.2 RMB，远低于算子的计算资源费用。

四、总结

开启通用增量 Checkpoint 可以显著提高 Checkpoint 的完成速度和稳定性，并以可控的性能与资源开销为作业提供更低的端到端延迟、更少的数据回追和更平稳的资源消耗。

在 Flink 1.16 后，通用增量 Checkpoint 已经达到了生产可用的状态。后续版本中，在实现机制上，我们将重点关注其他可用的 DSTL 存储组件，如 Apache Bookkeeper，来实现更短的延迟；在实践落地上，我们将重点关注与其他技术，如 Unaligned Checkpoint、Buffer Debloating 的结合，打造一整套完整高效稳定的 Checkpoint 流程。