摘要：本文整理自美团数据平台计算引擎组工程师冯斐，在 Flink Forward Asia 2022 生产实践专场的分享。本篇内容主要分为四个部分：
1. 相关背景和问题 2. 解决思路分析 3. 资源调度优化实践 4. 后续规划
Tips：点击「阅读原文」免费领取 5000CU*小时 Flink 云资源
在计算规模方面，目前我们有 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 监听并完成异常节点的处理。3.5 规避慢节点场景
接下来，介绍下规避慢节点场景。我们对部分并发慢，产生慢节点的原因进行了分类。
数据倾斜、逻辑倾斜都是业务侧的问题，引擎无法控制和应对。但资源不均是黑名单可以应对的。应对这种原因的慢节点，核心是如何感知慢节点。因为它们感知后的处理能力是相同的。
慢节点判定策略。首先，观察某个作业是否有部分并发的吞吐，明显高于其他并发。如果有，说明存在数据倾斜。如果没有，继续查看是否有部分并发的 processTime，比其他并发高。其中，processTime 是我们新增的单条消息处理耗时指标。如果 processTime 比其他并发高，我们需要判断是逻辑倾斜，还是存在慢节点。如果某个 TM 里存在消费慢的 task，那么这个节点的慢节点票数+1。如果一个机器上，超过一半的 TM 都认为该节点慢。那我们会认为消费慢的原因是，遭遇慢节点，这个节点会被发送给黑名单服务处理。3.6 其他优化
除了基于多作业的感知处理，一些明确异常可以直接闭环在引擎内部感知处理，提升处理时效。例如磁盘故障，是有明确特征的，不存在误判。这种可以直接在引擎内部完成感知和拉黑处理。
黑名单机制上线后，也有效解决了很多问题。首先是，应对故障节点。当节点出现磁盘故障时，作业的 restart 次数从之前的 10 多次降低到了 1 次。对于节点宕机的情况，我们可以在 10s 内发现和规避宕机的节点，作业 restart 的耗时从之前的 5 分钟降低至正常水平。在慢节点场景里，对于运行在慢节点上的 TM，黑名单使其在健康节点重新启动后，作业消费吞吐可恢复正常。
在资源和调度方向，后续的建设重点有两方面。
▼ 「活动推荐」免费领取 5000CU*小时 Flink 云资源 ▼
▼ 关注「Apache Flink」，获取更多技术干货 ▼
点击「阅读原文」，免费领取 5000CU*小时 Flink 云资源