小米基于 Flink 的实时计算资源治理实践-阿里云开发者社区

摘要：本文整理自小米高级软件工程师张蛟，在 Flink Forward Asia 2022 生产实践专场的分享。本篇内容主要分为四个部分：

发展现状与规模

框架层治理实践

平台层治理实践

未来规划与展望

一、发展现状与规模

如上图所示，下层是基础服务，包括：统一元数据服务、统一权限管理、统一任务调度、统一数据集成。

在此之上是各类分布式引擎，包括数据源、数据采集、消息中间件、数据计算和数据查询。Flink 主要位于数据计算模块，目前已经是实时计算事实上的标准，并且正在不断发力离线计算场景，向着更快更稳更易用的批处理引擎迈进。

目前，小米 Flink 平台运行着 5000+的用户作业和约 1 万 2 千个数据集成作业，他们共使用了 13 万左右的 CPU cores 和 460TB 的内存，我们可以看到资源消耗还是非常巨大的。

目前，用户在使用 Flink 开发实时作业时，存在着各种各样的问题。我将这些问题统一归纳总结为两大类问题，分别是经验税问题和非经验税问题。

经验税主要是指，用户在开发 Flink 作业时，因为经验不足造成的资源浪费。包括无法准确预估作业真实需要的资源、不当资源设置导致积压造成的运维压力，以及部分用户为稳定性和减少运维而设置大资源冗余等。
非经验税则是指，已经有一定经验的 Flink 作业开发者可能遇到的问题。比如由于内部 Flink 框架未支持细粒度资源管理导致的资源浪费、为应对短时流量高峰而不得不长期设置较大的资源、以及无法针对流量波动场景动态调整资源等。

介绍了用户开发 Flink 实时作业时存在的各种问题之后。下面就来看下这些问题导致的资源浪费的结果。虽然不同资源配置的集群，其堆内存使用率和 CPU 利用率都不同，但整体都是比较低的。用户作业所在集群的平均资源利用率仅在 35%左右，最低的甚至只有 20%左右，这造成了巨大的资源浪费。

上图所示，最近半年内，无论是用户的 Flink 作业还是数据集成作业，都增长了将近一倍左右。如果按照这个趋势继续增长下去，集群资源将会存在巨大的缺口和更巨大的资源浪费，因此对集群资源进行治理迫在眉睫。

Flink 实时作业的稳定性是非常重要的，因此我们提出资源治理的基本原则是，降本但不能将质。我们需要在保证稳定性不受大影响的前提下，达成资源节约的目标。因此，我们围绕这个基本原则，提出了以数据驱动，用价值量化，不断深入业务，持续进行业务推广和收集业务反馈的方式，形成渠道闭环。

二、框架层治理实践

弹性调度的主要逻辑都集中在 JobMaster。我们开发了一个全新的模块 DynamicSchedulerManager，作为弹性调度的控制器。它主要负责拉取和聚合从各个 TaskManager 收集到的各类弹性相关的指标。然后将这些指标和从 HDFS 拉取到的规则，统一由 Drools 进行处理和触发，并根据触发结果，按照垂直伸缩和水平伸缩两个大类进行调整。垂直伸缩主要针对单个 Container 的资源进行调整。目前，调整结果不支持持久化。水平伸缩主要针对并行度进行调整，调整结果可以持久化。Drools 是一个开源的规则引擎，而规则可以按照需要动态调优和更新。

弹性关键指标主要来源是两类。

TaskManager 和 Task 自带的指标，包括用于进行 CPU 调整的 CPU Load 和 Task 空闲指标等。
用于进行内存和并发调整的堆内/堆外内存利用率、GC 次数和频率、来自第三方 Connector 的流量以及积压情况等指标。

接下来将以一个具体的例子，来讲述实现的内存调整规则。上图左侧是 Flink 1.10 后的 TaskManager 的内存模型图。相信对于 Flink 有一定了解的同学，对这个图应该比较熟悉。根据 Java 堆大小的计算规则，假设 FullGC 后老年代剩余大小空间为 M。整个堆的大小建议值是 3~4 倍的 M。假设取堆大小建议值为 3M，结合 Flink 内存模型图，我们可以推算出建议的 TaskManager 内存大小，相关公式如上图所示。这个计算结果只是一个初略值，并不精确。真实的 TaskManager 的内存预估过程远比这个过程要复杂。

接下来，分享一下扩资源的原地重启的整个流程。

首先，AppMaster（实质上是 JobMaster 内的 DynamicSchedulerManager），它会向 ResourceManager 发出增加资源的请求。这个请求指定了 ContainerID 和目标资源值。Scheduler 会在调度周期内进行分配，返回新资源的 token，并启动一个监听器。

然后，AppMaster 会用新的 token，向 NodeManager 发出扩资源请求。ContainerManager 会以同步的方式，通知 Containers Monitor 更新资源监控并执行。

同时，它还会更新容器资源的记账和 metrics 信息。NodeStatusUpdater 会以心跳的方式，将资源更新的消息发送给 RM。于是，scheduler 就取消掉自己先前注册的监听器，整个扩资源的流程至此完成。

缩资源的流程跟扩资源略有不同，主要原因是，它不需要新的 token 来访问扩的资源。因此，它在 RM 已经缩资源后，就直接将被缩容 Container 的信息，通过心跳通知了 AM 和 NM。NM 获得被缩容 Container 的信息后，就通知 ContainerManager 并更新自己的相关 metrics 信息。ContainerManager 会通知 ContainersMonitor 更新其资源监控并执行，然后更新其内部的资源记账和 metrics。

接下来分享并行度调整相关的实践。它需要依赖于 Flink 1.13 版本提出的 AdaptiveScheduler。在 Drools 通过规则和指标进行处理，确定了需要伸缩的并行度大小后，需要先通过校验确定是否能够进行并行度的伸缩，如果校验不通过，则会直接撤回调整，否则就通知 Executing 进行调整。Executing 声明新的资源需求，并触发重启。如果伸缩成功，则会将新的并行度进行持久化。如果是缩并行度，还需要释放资源。实现上是启动了一个定时任务，定期检查并释放空闲 slot。

前面提到并行度调整时需要有一个校验，这个校验会进行伸缩条件判断。比如并行度是否已经到了最大并行度无法伸缩、伸缩比例是否合适、是否需要增大伸缩的比例以便能快速的消费积压、以及伸缩是否会造成数据的倾斜等。如果伸缩造成数据分布不均匀，很可能会影响作业的稳定性。此外，必须对并发伸缩前后的 DAG 图进行对比，避免 DAG 图发生变化，导致计算逻辑有误，或是有状态作业无法正常恢复。

除此之外，资源是否足够对扩并发也非常重要。因此，必须事先进行判断，否则在扩并发的过程中，出现资源申请超时，可能会严重影响作业。

目前，我们设计实现了多种调度策略以应对各种各样的弹性场景。根据调度周期来分，有固定时间的定时调度以应对流量波动比较规律的场景，有周期性调度，以应对通用的场景，还有能够根据某些阈值进行判断以自动触发的主动调度场景。根据触发主体的不同，实现了框架自动触发，无须人工干预的自动策略；以及由用户人工进行干预触发的手动策略。