美团 Flink 资源调度优化实践

摘要：本文整理自美团数据平台计算引擎组工程师冯斐，在 Flink Forward Asia 2022 生产实践专场的分享。本篇内容主要分为四个部分：

相关背景和问题
解决思路分析
资源调度优化实践
后续规划

点击查看原文视频 & 演讲PPT

一、相关背景和问题

在计算规模方面，目前我们有 7w 多作业，部署在 1.7w 台机器上，高峰期流量达到每秒 9 亿条。在部署方式上，目前我们主要还是在 Yarn 上使用 Session 模式部署作业。

大量的作业和机器也带来很多资源相关的问题，我们把问题分成两类。一类是硬件问题，比如磁盘故障、机器宕机、内存故障导致的机器卡顿等等。另一类是软件问题，包括磁盘 IO 被打满、作业间相互竞争影响等等。这两类问题，都会影响作业的部署和运行。

对于作业部署，最典型的问题就是，资源被调度到宕机节点，导致资源不能及时就绪，作业至少需要 5 分钟才能完成启动；或者调度到慢节点，导致 TM 启动耗时很长，作业启动慢。

对于作业运行，如果机器有问题，可能会导致这个机器上的作业处理慢，导致个别分区有消费延迟，甚至产生反压。

二、解决思路分析

如何解决这些问题？先看下问题的来源，异常的节点分成两类。

故障节点。通常是这个节点上出现了严重的故障，无法继续使用。比如磁盘损坏、机器宕机。
慢节点。虽然机器可用，但存在性能问题。例如网卡降速，导致作业处理能力下降；或者这个节点上有很多高负载作业。

当前，Flink 和 Yarn 都有一定机制来处理异常资源，但是也有缺陷不足。

首先，Flink 的心跳机制只能作为一个兜底机制。它无法感知节点的健康和负载情况。然后，Yarn 有心跳和健康检查两种机制。心跳检查的问题在于，超时时间过长。它需要 5 分钟才能感知到机器失联，这期间 Yarn 会认为机器正常可用。健康检查的问题是，感知机器故障的耗时达到分钟级别，而且不能发现所有的机器故障问题。

因此，我们希望通过加强 Flink 应对异常节点的能力，来保障资源能够健康及时地就绪。

首先，对于重启后遭遇其他故障节点的作业。我们通过复用 Session 集群资源的思路进行规避。这样不仅可以规避新的故障节点，而且能加快作业重新部署。其次，对于作业自动重启的场景。一个简单有效的思路就是冗余申请，通过申请过量资源的方式，使作业所需的资源全部就绪，从而规避节点故障导致的资源就绪慢或者无法就绪的问题。这需要用户的队列有足够的资源余量。

如果没有足够资源余量的队列，我们的思路是采用黑名单。当系统识别出异常节点后，进行规避。期望用这个思路来解决普遍的机器故障或者机器慢的问题。

三、资源调度优化实践

3.1 资源冗余申请

冗余申请和黑名单机制。首先介绍下资源冗余申请。我们在 Scheduler 中新增了一个 RedundantSlotAllocator 组件，负责发起冗余资源的申请。当作业完成调度后，我们会释放冗余的资源，这里主要复用了现有的清理空闲资源的能力。

下面介绍下冗余申请策略。首先需要要考虑的问题是，如何保障冗余申请是有效的？我们需要额外申请多少个冗余 container，才能确保能规避故障节点？

我们抽象了机器故障后的调度过程，得出如上图所示的模型。这个公式的含义是：加上冗余申请后，实际会就绪的 TM 数量，要大于等于作业部署所需的 TM 数量。化简后，可以得出，一个作业应该冗余的 TM 数量，要大于或等于作业的总 TM 数量除以队列机器数乘以机器数减一。

这个公式虽然简单，但也有一些前提。首先，队列中同一时间只有 1 个机器故障。其次，调度策略要保障调度均匀。

在冗余策略里，第二个问题就是，能否尽可能的节省资源？因为资源常驻式的冗余，虽然能最带来最快的资源就绪时效，但资源放着不用，是比较浪费的。

最终选择在作业部署或重启时，防御性的发起冗余资源申请，保障作业所需的资源，能够正常按时就绪。当作业部署或重启完成后，及时释放冗余申请的资源。通过这样的策略，我们在资源就绪时效性和资源成本中，取得平衡。

当冗余申请上线后，效果非常明显。SLA 作业的 tp99 的资源申请耗时从 30s 降到了 15s，tp9999 的耗时从 300s 降到了 20s。由此可见，资源就绪耗时被控制在正常范围内。

3.2 黑名单机制

黑名单机制分为感知和处理两部分。在感知部分，需要快速准确，它是黑名单机制有效的前提。在处理部分，需要灵活有效，从而应对各种类型的异常。

在设计黑名单时，看到社区和业界都有相关的思考和实践。因此，我们也进行了相关调研。

社区黑名单，主要用于在批计算推测执行中，规避慢节点。业界的黑名单机制，主要用于在实时作业调度过程中，规避故障节点。社区黑名单，通过对比任务执行耗时，来发现慢节点。业界黑名单，主要通过异常的次数累计，来识别节点故障。由此可见，社区和业界利用不同策略解决不同场景的问题。

接下来，介绍下美团的黑名单。如上图所示，左侧是黑名单的感知部分。我们收集作业运行或调度过程中的异常事件和运行指标。然后，根据一些策略识别出慢节点和故障节点。我们从应用层的视角感知异常，不需要明确完整的原因，也能快速准确的发现异常节点。

右侧是黑名单的处理部分，我们通过维护一个外围的黑名单服务，统一接受上一步识别出的异常节点，并把它们发送给资源管理服务或 Flink 作业来处理。我们从资源管理的视角出发，简化处理流程，支持流批两种执行模式、支持不同的资源管理服务。

3.3 故障节点感知策略

在前篇提到，我们需要快速准确的发现故障节点，那我们是怎么做到的呢？通常如果机器有问题，这个机器上的作业都可能受影响。如果多个作业的异常，来自同一个节点，那我们有理由相信这个节点有问题。

基于上述思路，我们通过 track-service 收集所有作业的异常信息。然后，用一个 Flink 作业判断，是否在同一时间的某个节点上，多个作业都有异常。如果有这样的节点，我们就把它发送给黑名单服务来处理。相比单个作业积累多次异常，这种方式能更快更准的发现故障节点。

3.4 异常节点处理机制

上图所示，这里罗列了一些我们主要关注的异常。在启动时，我们关注 JM 和 TM 的启动是否成功、是否及时。在运行过程中，我们关注 TM-JM 间的心跳超时异常、TM 被 Kill 的异常、Task 运行异常。通过聚合这些异常信息，我们就能找出哪些节点有异常。

如何有效处理不同类型的异常节点。目前，我们支持两种处理方式。即可以让 TM 立即从异常节点上退出，也可以先运行，等下次 restart 时，再退出异常节点。在处理粒度方面，既支持处理单个作业，也可以直接处理整个节点。

Flink 和 Yarn 如何处理异常节点？在 Flink 内部，我们新增一个组件 Unhealthy Node Manager，负责对异常节点的管理。

这个组件定义在 Flink 的资源管理层，与上层任务调度的逻辑解耦。这样可以支持流和批两种执行模式，而且不依赖作业的调度状态。

对于下层物理资源管理，通过抽象核心接口，可以适配不同的资源管理服务。除此之外，通过提供对外交互的 API，可以跟外部系统联动。

在 Yarn 侧，我们在原有健康检查的基础上，新增了 FREEZE 状态，表示节点不再接受调度，但也不 Kill 正在运行的 container。与此同时，我们打通了 Yarn 的健康检查机制，因为一些人力和成本的原因，我们使用了基于 zk 的共享存储，黑名单服务发布异常节点信息，Yarn 监听并完成异常节点的处理。