关于flink:Flink-容错恢复-20-2022-最新进展

摘要：本文整顿自阿里云 Flink 存储引擎团队负责人，Apache Flink 引擎架构师 & PMC 梅源在 FFA 核心技术专场的分享。次要介绍在 2022 年度，Flink 容错 2.0 这个我的项目在社区和阿里云产品的停顿。内容包含：

1. Flink 容错复原 2.0 我的项目简介及思考
2. 2022 年度 Flink 容错 2.0 我的项目停顿

点击查看直播回放 & 演讲 PPT

一、Flink 容错复原 2.0 我的项目简介及思考

首先来概括的看一下 Flink 容错的链路，次要包含四个过程：做快照（Checkpointing），发现失败节点（Failure Detection），从新调度（Re-Scheduling）和状态复原（State Recovery）。

在一个 Flink 作业中，数据从数据源通过各个算子的解决最终写入 Sink。其中有些算子须要记录数据处理的两头后果，会临时把两头后果缓存在算子外部，即算子的状态中。如果这个 Flink 作业因为各种起因呈现失败或者谬误，就须要从新复原这些状态，因而咱们须要对状态进行周期性快照，即 Checkpointing 过程。因为快照很频繁，所以须要 Checkpointing 过程稳固、轻量并保障胜利。

如果一个节点挂了，须要疾速发现失败节点，并实现相应的清理工作。

而后生成新的作业，并从新调度，实现部署。

当作业从新调度之后，须要从最新的快照中复原算子的中间状态，也就是 State Recovery 状态复原。

从下面的形容不难看出，容错复原是一个全链路的过程，包含给状态做快照，发现节点失败，从新调度部署以及复原状态。另一方面容错复原也是个须要从多个维度来思考的问题，包含容错老本、对失常解决的影响、数据一致性保障的水平、以及在整个容错复原过程中的行为表现（比方是否须要满速 TPS 的保障和一直流的保障）等等。另外须要指出的是在云原生背景下，容器化部署带来了一些新的限度和便当，这些都是咱们在设计新一代容错时不能漠视的中央。无关这个局部更具体的内容能够参考去年的 talk“Flink 新一代流计算和容错——阶段总结和瞻望”

2022 年咱们的工作次要集中在 Checkpointing，Scheduling 和 State Recovery 这三个局部。在 Checkpointing 这个局部，咱们实现了分布式快照架构的降级：

• Flink 1.16 修复了和 Unaligned Checkpoint 相干的几个要害 bug，特地是 Unaligned 和 Aligned Checkpoint 之间转换这个局部，使得 Unaligned Checkpoint 能够真正做到生产可用；
• Flink 1.16 公布的一个重要 feature 是通用增量 Checkpoints，通过增量快照和 State Store 快照过程的拆散，能够保障稳固疾速的 Checkpointing 过程。这个后续咱们会有相干 Blog Post 具体的剖析通用增量 Checkpoints 在各种 Benchmark 下的各项指标以及实用场景。

在 Scheduling 这个局部，Approximate 和 At-least-once 单点重启性能曾经在阿里云实时计算企业级服务里实现，次要由阿里云实时计算 Flink Runtime 团队实现，他们还通过引入作业热更新性能，极大缩短了扩缩容的断流工夫。这两个性能在阿里云实时计算 VVR 6.0.4 版本上线，欢送大家试用。

在 State Recovery 这个局部，社区在 1.15 引入了工作目录的概念，联合 K8S 的 PV 挂载，能够实现状态的本地复原。为了适应云原生部署，咱们在状态存储局部进行了分层存储架构降级，对复原和扩缩容都有很大的改良。特地是 Lazy-Load 的引入，使得疾速复原不再受状态大小的影响，联合下面提到的作业热更新性能，根本能够做到扩缩容无断流。Lazy-Load 这个性能也已在阿里云 VVR 6.x 版本中上线。

上面我次要从快照生成，作业复原 / 扩缩容，快照治理这三个方面具体介绍 2022 年度 Flink 在容错局部的停顿。

二、2022 年度 Flink 容错 2.0 我的项目停顿

2.1 优化快照生成

在快照生成局部，咱们对 Flink 分布式快照架构进行了整体降级。

先来看看降级前的架构有什么问题：

问题 1：对齐工夫长，反压时被齐全阻塞

Flink 的 Checkpoint 机制是通过从 Source 插入 Barrier，而后在 Barrier 流过每个算子的时候给每个算子做快照来实现的。为了保障全局一致性，如果算子有多个输出管道的时候，须要对齐多个输出的 Barrier。这就产生了问题 1，因为每条链路的处理速度不一样，因而 Barrier 对齐是须要工夫的。如果某一条链路有反压，会因为期待对齐而使得整条链路齐全被阻塞，Checkpoint 也会因为阻塞而无奈实现。

问题 2：Buffer 数目固定，管道中有多余的解决数据

因为算子间的上下游 Buffer 数目是固定的，它们会缓存比理论所需更多的数据。这些多余的数据不仅会在反压时进一步阻塞链路，而且会使得 Unaligned Checkpoint 存储更多的上下游管道数据。

问题 3：快照异步上传工夫较长且不可控

快照的过程包含两局部：同步状态刷盘和异步上传状态文件，其中异步文件上传的过程和状态文件大小相干，工夫较长且不可控。

咱们从 Flink 1.11 开始着手逐个的解决这些问题，如下图所示。

Flink 1.11、Flink 1.12 引入了 Unaligned Checkpoint，使得 Checkpoint Barrier 不被迟缓的两头数据阻塞。Flink 1.13、Flink 1.14 引入了 Buffer Debloating，让算子与算子间的管道数据变得更少。Flink 1.15、Flink 1.16 引入了通用增量 Checkpoints，让异步上传的过程更快、更稳固。

降级后的分布式快照架构如下图所示：

对于问题 1，在 Flink 1.16 版本中，Unaligned Checkpoint 容许透支 Buffer，解决了在 Buffer 有余时，不能及时响应 Unaligned Checkpoint 的问题。此外，全局计时超时机制的引入可能无效改良 Unaligned 和 Aligned Checkpoint 之间主动转换的触发条件。

对于问题 2，Buffer debloating 的引入能够动静调整缓存的数据量，动静缓存 1 秒内须要解决的数据。

上面咱们来重点看一看第 3 个问题是如何用通用增量 Checkpoint 来解决的

Flink 的算子状态更新会反映在状态表中。在之前的设计当中，Flink 算子做快照的过程分为两步：第一步是同步的对状态表进行快照，内存中的数据刷盘，筹备好上传到长久存储的文件；第二步是异步的上传这些文件。

异步上传文件这个局部有两个问题：

问题 1：异步上传的文件大小依赖 State Backend 的实现

问题 2：异步过程须要等到同步过程完结能力开始，因为同步快照完结前是没法筹备好须要上传的文件的

咱们来别离看一下这两个问题。对于第一个问题，以 RocksDB 为例，尽管 Flink 对 RocksDB 也反对增量 Checpoint，然而 RocksDB 出于本身实现思考，它须要对文件做 Compaction。每次 Compaction 会产生新的比拟大的文件，那这个时候即便是增量 Checkpoint，须要上传的文件也会因而时不时变大。在 Flink 作业并发比拟大的状况下，上传文件时不时变大的问题就会变得很频繁，因为只有等所有并发的文件上传完毕，一个残缺的算子状态才算快照实现。

对于第二个问题，在同步快照完结前，Flink 无奈筹备好须要上传的文件，所以必须要等快照完结时能力开始上传。也就是说，上图中的红色斜条纹这个时间段齐全被节约了。如果须要上传的状态比拟大，会在很短时间内对 CPU 和网络产生较大的压力。

为了解决上述两个问题，咱们在 Flink 社区实现了通用增量快照。在新架构下，状态更新不仅会更新状态表，而且会记录状态的更新日志。上图中状态表会和架构降级前一样周期性的刷到长久存储，然而这个周期能够比拟大（比方 10 分钟）在后盾缓缓上传，该过程称为物化过程。同时状态更新日志也会继续上传到远端长久存储，并且在做 Checkpoint 时 Flush 残余全副日志。

这样的设计就比拟好的解决了后面提到的两个问题：通过将快照过程和物化过程齐全独立开来，能够让异步上传的文件大小变得很稳固；同时因为状态更新是继续的，所以咱们能够在快照之前就始终继续的上传更新日志，所以在 Flush 当前咱们理论须要上传的数据量就变得很小。

架构降级后的一个 Checkpoint 由物化的状态表以及增量更新的日志组成。物化过程完结后，绝对应的更新日志就能够被删除了。上图中的蓝色方框局部，是通用增量快照和之前架构的区别，这个局部被称为 Changelog Storage（DSTL）。

DSTL 是 Durable Short-term Log 的缩写。咱们从这个英文名就能看进去 DSTL 是有针对性需要的

• 须要短期长久化增量日志，物化后即可删除
• 须要反对高频写，是一个纯 append 写操作，仅在复原时须要读取
• 须要 99.9% 的写申请在 1 秒内实现
• 须要和现有的 Checkpoint 机制提供同一级别的一致性保障

社区当初的版本是用 DFS 来实现的，综合考量下来根本能够满足需要。同时 DSTL 提供了规范的接口也能够对接其余的存储。在本次 Flink Forward 美团的分享中，能够看到美团在应用 Bookkeeper 实现 DSTL 以及通用增量快照方面获得的性能的晋升。

这个局部的最初咱们来看一下应用通用增量快照的 Trade-off

通用增量快照带来的益处不言而喻：

• 能够让 Checkpoint 做的更稳固，平滑 CPU 曲线，安稳网络流量应用（因为快照上传的工夫被拉长了，并且单次上传量更小更可控）
• 能够更疾速的实现 Checkpoint（因为缩小了做快照 Flush 的那个局部须要上传的数据）
• 也因而，咱们也能够取得更小的端到端的数据提早，减小 Transactional Sink 的提早
• 因为能够把 Checkpoint 做快，所以每次 Checkpoint 复原时须要回滚的数据量也会变少。这对于对数据回滚量有要求的利用是十分要害的

通用增量快照也会带来一些额定的 Cost，次要来自两个方面：Checkpoint 放大和状态双写：

• Checkpoint 放大的影响次要有两点。第一，远端的存储空间变大。但远端存储空间很便宜，10G 一个月大概 1 块钱。第二，会有额定的网络流量。但个别做 Checkpoint 应用的流量也是内网流量，费用简直能够忽略不计。
• 对于状态双写，双写会对极限性能有一些影响，但在咱们的试验中发现在网络不是瓶颈的状况下，极限性能的损失在 2-3% 左右（Flink 1.17 中优化了双写局部 FLINK-30345，也会 backport 到 Flink 1.16），因而性能损失简直能够忽略不计。

对于通用增量 Checkpoint 这个局部咱们近期会有更详尽的测试剖析报告，敬请期待。

2.2 优化作业复原和扩缩容

接下来讲一讲 Flink 社区在作业复原和扩缩容局部的优化，次要包含优化本地状态重建，云原生背景下的分层状态存储架构降级，以及简化调度过程。

作业扩缩容和作业容错复原有很多共性，比方都须要根据上一次快照来做复原，都须要从新调度，但他们在轻微之处又是有些区别的。

本地状态重建

以状态复原本地重建来讲，对于容错复原，将状态文件原样加载进本地数据库就能够了，然而如果是扩缩容复原就会更简单一些。举例来说上图中的作业并发从 3 扩容到 4，新作业 task 2 的状态有一部分来自原先作业的 task 1，还有一部分来自原先作业的 task 2，别离是橙色和黄色局部。

Flink 作业算子的状态在 Rescaling 做状态重新分配时，新调配的状态来自原先作业相邻的并发，不可能呈现跳跃的有距离的状态调配。在缩容时，有可能有多个状态合成一个新状态；在扩容的时候，因为状态肯定是变小的，所以新的变小的状态肯定最多来自相邻的两个原先的并发。

接下来具体讲一讲状态是如何做本地重建的，以 RocksDB 为例。

• 第一步，须要下载相干的状态文件。
• 第二步，重建初始的 RocksDB 实例，并删除对实例无用的 Key，即删除上图中灰色的局部，留下橙色局部。
• 第三步，将长期 RocksDB 实例中的 Key 插入到第二步重建的 RocksDB 中，也就是黄色的局部插入到橙色的 DB 中。

咱们在 Flink 1.16 中，对本地重建的第二步进行了优化。通过引入 DeleteRange，使得整个删除无用 Key 的操作变成 O（1），并且因为最多只可能有 2 个 Range 须要删除，因而额定的数据结构（TombStone 表）对失常读写的影响微不足道。

从上图左边的试验后果能够看出，对于状态大小 122GB 的 Word Count 作业，从并发 3 扩容到并发 4，Flink 1.16 比之前的版本扩容速度晋升 2 – 10 倍。同时咱们在 Flink 1.16 引入了规范的 Rescaling Micro Benchmark，在此之前社区没有一个规范来测试 Rescaling 的性能。

在阿里云实时计算企业版中，咱们对本地状态重建的第一步和第三步也进行了优化。咱们只须要下载须要的状态文件，并且能够进行文件粒度的间接合并，防止创立长期 DB 实例。阿里云实时计算版本和 Flink 1.16 社区版本比照，缩容速度也有 7 倍的晋升。

分层状态存储架构

为了更好的适应云原生的大背景，咱们对分层状态存储架构也进行了初步摸索，也就是说咱们把远端盘也作为 State Backend 的一部分。这种分层架构能够解决 Flink 状态存储在云原生背景下面临的大部分问题：

• 解决容器化部署本地磁盘大小受限的问题
• 解决外置状态老本高，数据一致性难以保障的问题
• 解决小状态须要额定落盘的问题
• 解决大状态拜访速度慢的问题

这些问题和容错没有太大关系，当前有机会专门讲一讲这个局部。

远端盘作为 State Backend 的一部分，状态加载策略能够变得更灵便。这样状态在没有齐全加载复原实现之前，就能够开始数据处理。在优化前，用户状态复原时，读写被齐全阻塞。在优化后，用户状态复原的过程中，能够进行半速读写，而后逐步复原到全速，如上图所示。

可配置的状态加载策略的引入，极大的缩短了状态复原的初始提早和业务断流工夫。咱们从上面两个试验能够看到，单并发状态大小 7GB 左右的作业启动后延迟时间在优化后从 4.5 分钟降到了 1.25 分钟，状态复原局部提速 75%。同样的作业，在优化后能够发现从状态复原开始时，就会有 TPS，极大的升高了业务断流的工夫。

在作业调度这个局部，阿里云 Flink Runtime 团队在阿里云实时计算 VVR 6.0.4 版本中引入了作业热更新这个性能，其核心思想是在扩缩容的时候简化作业从新调度的步骤，并且在新的作业生成后再停掉老的作业，这样能够进一步缩短作业断流的工夫。在没有资源预申请的状况下，作业热更新能够使无状态作业扩容和缩容工夫升高三倍左右。

综上所述，通过提早状态加载策略，配合作业热更新，根本能够保障在状态复原和调度层面，做到扩缩容无断流或极短时间断流。

2.3 优化快照治理

前两个局部都是讲性能上的优化，最初一个局部我想聊一聊快照治理局部的一些梳理。Flink 倒退到现今曾经是一个很成熟的零碎了，清晰化的概念以及简略易用性是掂量一个零碎成熟度的很重要的局部，所以这里聊一聊快照概念和治理。

Flink 的快照 Snapshot 分为两种：Savepoint 和 Checkpoint。

Savepoint 个别由用户触发，所以它归属用户所有，因而由用户负责创立和删除。正因而，Flink 零碎引擎层是不可能去删除 Savepoint 相干文件的。所以 Savepoint 不和 Flink 作业强绑定，不同的 Flink 作业能够从同一个 Savepoint 启动。Savepoint 是自蕴含的：本人蕴含所须要的所有。

Checkpoint 正好相同，它的次要作用是零碎容错自愈，所以它由 Flink 引擎周期性触发，并且所属权归属 Flink 引擎。Checkpoint 文件的组织构造都由 Flink 引擎决定和治理，所以引擎负责按需清理 Checkpoint 文件。正因而，Checkpoint 和生成该 Checkpoint 的作业强绑定，并且是非自蕴含的，比如说 Incremental Checkpoint 之间会有依赖关系。

那有什么问题呢？因为 Savepoint 次要指标服务对象是用户，为了对用户敌对，Savepoint 应用用户可读的规范格局，也正因而 Savepoints 做得十分慢，常常状况下状态略微大一点就会超时，同样复原也很慢。另一方面，Checkpoint 应用的是增量零碎原生格局，所以做得很快。

这种状况下，用户会把 Retained Checkpoint 当成 Savepoint 来应用。Retained Checkpoint 是在作业停掉后保留的 Checkpoint，这样 Retained Checkpoint 就变成了 Savepoint 和 Checkpoint 的混合体。造成的问题是用户负责删除 Retained Checkpoint，然而用户并不知道如何平安的删除 Retained Checkpoint。

为了解决上述问题，Flink 1.15 引入了两种状态恢复模式，即 Claim 模式和 No-Claim 模式。

在 Claim 恢复模式下，引擎申明 Retained Checkpoint 的所属权，Retained Checkpoint 归引擎所有，引擎负责删除。

在 No-Claim 恢复模式下，引擎放弃 Retained Checkpoint 的所属权。Retained Checkpoint 中所有的文件都不会被 Flink 引擎应用，用户能够很平安的删除 Retained Checkpoint。

在 No-Claim 的根底上，咱们引入了 Native Savepoint，来减速 Savepoint 的创立和复原。Native Savepoints 应用和 Checkpoint 一样的存储格局，其实现原理和 No-Claim 相似。Savepoint 不会应用之前的 Checkpoint 文件，相当于做一个全量的 Checkpoint。咱们的企业版本通过进一步优化，让 Native Savepoint 也真正能做到增量 Savepoint。

上图是 Flink 社区引入 Native Savepoint 之前，之后以及企业版进一步优化后的 Savepoint 性能比照图。咱们能够看到在引入 Native Savepoint 之前，Savepoint 做的很慢。在 Savepoint 总大小 5GB 的状况下（状态中等大小），做一次 Savepoint 超过 10 分钟（超时）。这意味着一旦用户应用 stop-with-savepoint 来进行作业（也就是在进行作业前做个 Savepoint），就得等超过 10 分钟，齐全没法用。在引入 Native Savepoint 之后，须要等 2 分半钟，也比拟长，勉强能用。企业版进一步优化后，等待时间变成 5s 钟，这个基本上在可期待的范畴之内了。

最初咱们小结回顾一下 Flink 容错复原在 2022 年的次要停顿

• 在分布式快照架构方面，Unaligned Checkpoint 引入全局计时器，能够通过超时机制主动从 Aligned Checkpoint 切换成 Unaligned Checkpoint，这个对于 Unaligned Checkpoint 生产可用是十分重要的一步
• 通用增量 Checkpoint 生产可用，这对于 Checkpoint 稳定性和实现速度有很大的晋升，同时能够平滑 CPU 和网络带宽的应用
• 这里值得一提的是，不仅仅是阿里巴巴在 Checkpoint 这个局部奉献了大量的代码，很多其余的公司也踊跃的投入到社区当中，比方 Shopee 和美团。他们在社区中奉献代码同时，也踊跃推动这些性能在公司外部的落地和延展，获得了不错的成果
• 在状态存储方面，咱们进行了分层状态存储的初步摸索，扩缩容速度有 2 – 10 倍的晋升
• 阿里云实时计算平台推出了扩缩容无断流的组合性能：提早状态加载和作业热更新，别离从状态加载和作业调度这两个方面来实现扩缩容无断流
• 引入增量 Native Savepoint，全面晋升 Savepoint 的可用性和性能。

点击查看直播回放 & 演讲 PPT

更多内容

流动举荐

阿里云基于 Apache Flink 构建的企业级产品 - 实时计算 Flink 版现开启流动：
99 元试用 实时计算 Flink 版（包年包月、10CU）即有机会取得 Flink 独家定制卫衣；另包 3 个月及以上还有 85 折优惠！
理解流动详情：https://www.aliyun.com/produc…