快手 Flink 的稳定性和功能性扩展

摘要：本文整理自快手技术专家刘建刚，在 Flink Forward Asia 2022 生产实践专场的分享。本篇内容主要分为四个部分：

1. 快手 Flink 平台

2. 稳定性保障和智能运维

3. 复杂场景下的功能扩展

4. 批处理的定制优化

Tips：点击「阅读原文」查看原文视频&演讲 ppt

快手 Flink 平台

1.1 计算平台架构

如上图所示，Flink 主要运行在 Yarn 和 K8s 上，存储组件使用 HDFS 和快手自研的 kwaistore。在 Runtime 层，同时支持流处理和批处理。在用户结构层，主要包括 Data Stream 和 SQL。最上面一层是快手的作业管理平台，包括前端和后端。

基于 Flink 计算生态，周边组件囊括万千。既有 Kafka、RocketMQ等中间件，又有 Clickhouse、Druid 等 OLAP 分析，还有 Hudi、Hive 等存储层。最外层是应用场景，涵盖了音视频、商业化、数据工厂、湖仓一体等应用场景。

1.2 架构演进

架构演进

在 2018 年~2019 年，我们搭建了实时计算平台，在 Data Stream、SQL 计算、大状态存储等方面，为用户提供生产可用的功能。
在 2020 年~2021 年，我们对 SQL 进行了进一步的优化和推广，对 Flink Runtime 进行了深度改造。
在 2021 年~2022 年，我们开始探索流批一体。与此同时，在稳定性和功能性方面，进行了深度改造。
未来，我们会探索更加智能的流批融合，进一步丰富计算生态。

1.3 未来规划

上图是未来规划的大一统架构。从用户的角度来看，不再需要关心流和批，以及复杂的流批配置。对用户暴露的将是简单的资源延迟和吞吐配置，底层计算引擎会根据这些配置智能执行。

Flink 将作为统一的计算层，为用户暴露统一的 SQL 和 Data Stream API。统一的调度器，会解决用户的资源调度和算子调度。增量和全量计算对应用户的延迟要求。如果用户希望尽快得到结果，可以使用更短的增量触发。

在批处理方面，我们希望能在 Streaming 内核的基础上，达到业界领先。除此之外，Shuffle 对用户的吞吐至关重要。我们希望能够在 Pipeline 和 Blocking 之间智能切换，实现最大的吞吐。

在流批的智能切换和无缝融合方面，需要在性能和平滑过渡上多做一些工作，为用户屏蔽具体的细节。在存储层通过统一的管理，为用户展现一致的 Table 表。

稳定性保障和智能运维

2.1 自动迁移

稳定性保障和智能运维。由于我们经常会遇到机器下线、队列下线、集群下线的情况。此时，如果让用户自己迁移作业会非常麻烦，主要有三个方面。

机器下线是常态化的事情，会频繁干扰用户。
当涉及的用户作业数较多，手动操作非常繁琐。
沟通成本较大，会浪费大量的时间。

如上图所示，先来看一下单作业的自动迁移。对于用户来说，只需要配置 AutoMigrate 参数即可，该参数可以配置三个值。

Normal 值表示自动驱逐 Container 并恢复。
Exactly-Once 表示 stop-with-savepoint 并恢复。
Manual 表示通知用户 Dealine 前手动处理，否则自动处理。

当用户配置完参数，Flink 平台会在机器下线时，自动帮助用户操作。主要分成以下四步。

第一步，将下线的机器标记为不再调度。
第二步，找到对应的 Flink 作业。
第三步，按照用户的配置，进行自动化迁移。
第四步，当所有作业都迁移完后，机器就可以正式下线。

除了单作业的自动迁移，我们还具备集群迁移的经验。集群迁移的复杂性主要体现在以下三点。

Flink 及其上下游的数据一致性很难得到保证。
上千个作业，操作复杂，稳定性难以保障。
用户众多，沟通运维成本极高。

为此我们开发了一套自动化迁移程序。如上图所示，单个作业会在新集群，自动拷贝一个新作业。然后，通过 stop-with-savepoint 停止老作业，在新集群启动新作业。在启动时，我们会跨集群读取老作业的 Savepoint。除此之外，我们会通过各项指标检查作业的健康度。一旦出现问题，立刻回滚。

涉及集群操作时，我们会进行批量的灰度操作。如右图所示，我们通过批量的灰度，操作这些作业。一旦出现问题，会立刻回滚，并且告知用户尽快介入。

为了保证上下游 Kafka 数据的一致性，Flink 针对 Kafka 进行了深度改造和适配。在消费端，Kafka 会 Mirror 最近一段数据到新集群，确保 Offset 和数据都一样，Flink 自动切换到新集群消费。

在产出端，Kafka 也会自动 Mirror 数据到新集群，当 Flink 自动切换后，新数据会写到新集群，确保 Offset 和数据一致。

2.2 故障归因

故障归因主要讨论不确定性的硬件故障，主要有以下三点。

磁盘故障，比如磁盘坏道、内核故障等引起的读写卡顿问题。
内存故障，比如部分数据脏写、gc 频繁等。
网络故障，比如网卡异常导致无法连接等问题。

这些问题的共性是，导致作业卡顿或者表现异常，但又不是彻底的失败，定位非常困难，甚至无法查出问题所在。当规模比较大时，这种问题会频繁出现。

接下来，介绍一下应对方案。我们的应对方案主要包括以下四点。

自动拉黑。既包括在作业粒度，将机器拉黑。也包括在平台维度，将一个机器彻底拉黑。
智能归因。我们会监控 Flink 算子的异常 Task，比如某个 Task 的延迟、吞吐、快照等有明显异常。我们会把它迅速拉黑。
人工快速识别。当我们怀疑一批机器有问题时，会快速计算共同的机器，并进行指标分析。一旦确定异常，立刻拉黑。
建立实施故障指标库。虽然离线计算能够容忍很多机器问题，但实时计算不能容忍。我们需要识别这些 Case，并且快速的处理，防止作业出现问题后处理。

故障归因。故障归因的目的是，将人工运维的经验，沉淀到自动化程序里，进一步解放开发人员、运维人员和使用人员。主要分为作业失败和作业性能两个诊断。

在作业失败方面，查询单个用户作业失败的原因时，首先去 ES 获取作业失败的原因。如果是用户的问题，直接返回给用户。如果是心跳超时，去 Yarn 查询 Host 和 Container 的存活性，然后再去 metric 系统查询是否 gc 等信息。

如果是其他问题，先看一下是否在系统诊断库里，如果是就可用直接返回。如果不是，直接给用户返回异常。与此同时，人工会排查原因，并把它加入诊断库。

作业性能问题的排查。首先，进行宽泛检查。比如资源是否打满，数据是否倾斜。然后，找到第一个有问题的 Task。如果这个作业有快照，我们就会找到第一个快照失败的 Task。

如果没有快照，我们会递归查询反压，并根据 Task 的 Input Queue 来判断它是不是第一个出现问题的 Task。

如果确定了第一个出现问题的 Task，会为用户返回资源、线程等相关信息，辅助用户确定根本原因。

2.3 分级保障

分级保障。分级保障是指给予不同优先级作业，不同的保障级别，其背景如下。

在资源紧张的情况下，我们会优先保障高优的作业。
我们会为高优的作业提供平台化的保障措施，比如热备、冷备等方案。
我们会根据优先级统筹规划，提高整个集群的利用率。

如上图所示，我们为作业划分了四个等级，其中 P0 为双 AZ 容灾（热备）、P1 为双 AZ 容灾（冷备）、P2 表示单集群常规作业、P3 表示不重要的作业资源，随时可能被抢占。

下面重点介绍一下冷备方案和抢占方案。Flink 的冷备方案既支持 Flink 冷备，也支持 Kafka AZ 容灾，主要指消费两个同名的 Topic 和写出两个同名的 Topic。同名 Topic 在不同的 AZ 下，两个同名的 Topic 共同组成一份完整的数据。

这时如果上游的一个 Kafka 集群挂掉，Flink 会自动容灾，并推动 watermark 的前进，整个作业不受影响。Flink 在常规情况下，通过轮转写的方式，将数据写到下游的两个 Topic。如果一个 Topic 挂掉，数据会全部导到另一个 Topic。

针对 Flink 作业，我们会定期将快照写到备集群。一旦作业管理平台监测到 Flink 所在的 AZ 挂掉，会自动在备集群拉起一个一样的 Flink 作业。

未来，我们将实现 HDFS、Kafka 的双 AZ 部署，到时它们会自动 AZ 容灾并为 Flink 呈现逻辑视图。

资源不足时的抢占方案。抢占策略主要有三点。

高优作业抢占低优作业资源。
优先抢占不健康的作业，比如 lag 严重的作业。
实时作业会优先抢占同一个作业的资源。

右图展示了我们的抢占效果。作业通过改造，在资源不足时，也能启动。

常见的保障措施，主要包括以下四点：

资源隔离。高优作业可以单独划定队列，实现物理隔离，同时不与离线作业混部。
资源抢占。在资源紧张情况下，高优作业可自动抢占低优作业的资源。
AZ 容灾。高优作业可实现 AZ 容灾，包括冷备和热备。
智能监控报警。高优作业配套的报警更加完善，一旦出现预期之外的问题可快速人工介入。

复杂场景下的功能扩展

3.1 弹性伸缩

在复杂场景下的功能性扩展。这里主要讲三个部分，即弹性伸缩、Remote Shuffle Service、云盘存储。接下来，来看下弹性伸缩的背景。

第一，Flink 作业当前的静态资源分配一般都是按照最高峰申请的，导致了其他时间段的资源，大量浪费。
第二，PerJob 模式调整并发度慢，需要停止作业、修改并发度、启动作业等繁琐流程。
第三，用户不知道该配置多少资源，有的延迟严重、有的资源浪费严重，带来了各种运维问题。

基于以上背景，我们开发了更轻量级的弹性伸缩方案，用户或者平台决定好并发度后，直接发给 Flink 作业，Flink 作业在不停止作业的情况下快速完成调整。

先来看一下整体的架构实现，用户可以直接触发扩缩容或者配置扩缩容的条件。比如当 CPU 或 IO 超过多少之后，进行调整。AutoController 作为自动控制的组件，会根据用户的配置，自动完成作业的弹性伸缩，比如根据 metric 自动触发调整。

在 Flink 内部，我们实现了快速 rescale 的接口。伸缩原理如下，如果是扩容，会提前申请资源，然后将并发度持久化到 ZK 里，来防止 Master Failover，然后重新生成执行图。在停止时，默认使用 stop-with-savepoint。

弹性伸缩的效果。扩缩容时间，常规聚合作业从分钟级别降低到 10s 左右，常规 ETL 作业 3s 内可完成调整。通过平台调整，作业资源占有量显著下降，可以有效整顿集群的资源利用率。将用户从作业资源调整的繁琐运维中解放出来，极大地减少了人工运维工作。

3.2 Remote Shuffle Service

Remote Shuffle Service。Flink 由 TaskManager 来管理 Shuffle 数据，计算和存储耦合，存在以下问题：

一旦 TaskManager 挂掉就会造成 Shuffle 数据丢失，存在重跑整个任务的风险。
空闲 TaskManager 因为 Shuffle 数据的存在而无法退出，导致资源浪费。
Shuffle 代价大，无论是网络连接、还是磁盘读取，开销都比较大。

快手内部自研的 Shuffle Service 主要采用 Master slave 架构。接下来，介绍一下 Flink 和 Shuffle Service。Flink 的 StreamShuffleMaster 负责跟 Shuffle Service 的 ShuffleManager 交互。

Task 的 Reader 和 Writer 通过心跳从 StreamShuffleMaster 处获取读写地址，将数据读写到 Shuffle Service。Shuffle 数据会持久化到 HDFS，防止数据丢失。

数据交互流程。针对上下游的交互，中间数据以 ReducePartition 的形式存在。Shuffle Service 负责将上游 Task 的数据聚合到一块，下游 Task 到时候只需要一对一的顺序读取即可。Shuffle Service 在网络和磁盘方面做了很多的优化，确保了整个 Stage 的快速数据传递。