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

发展现状与规模

如上图所示，下层是基础服务，包括：统一元数据服务、统一权限管理、统一任务调度、统一数据集成。
在此之上是各类分布式引擎，包括数据源、数据采集、消息中间件、数据计算和数据查询。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 图发生变化，导致计算逻辑有误，或是有状态作业无法正常恢复。
除此之外，资源是否足够对扩并发也非常重要。因此，必须事先进行判断，否则在扩并发的过程中，出现资源申请超时，可能会严重影响作业。

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

在这个过程当中，我们也遇到了各种各样的问题。比如，周期性触发无法应对流量突增的问题。尤其是当周期触发的弹性减少作业资源后更是导致问题出现得非常频繁。我们最后通过 FullGC Monitor 监控来根据阈值判断，从而触发主动伸缩容进行处理。
触发场景的多样，也导致作业频繁发生框架内重启，甚至是 APP 级别的重启，影响了作业的稳定性。除此之外，频繁报警也增加了我们的运维压力。
通过增加触发的稳定间隔，以及增加伸缩编排功能，让针对相同 Container 的多次调整进行合并，我们很好的保持了作业的稳定性。
此外，每次伸缩资源都会导致框架内的重启，熟悉 Flink 的同学都知道，这类重启通常并不能保证端到端的 Exactly once 语义，可能导致数据的重复。针对这类场景，我们通过引入增量 savepoint，让需要保证端到端数据一致性的场景，从增量 savepoint 恢复来解决。

由于 Flink 并行度的调整，不能超过最大并行度设置。因此，为了更好的发现这类场景，我们对作业各个算子的最大并行度进行了计算和处理，在页面上展示给用户。除此之外，我们还将默认最小的最大并行度值，设置为 512 来避免无法调整的情况。

经过弹性伸缩这一系列调整，我们在资源治理上取得了一些成效。如上图所示，展示了某个 Kafka（Talos）Topic 的流量随时间发生着周期性动态变化。

如上图所示，在开启垂直弹性后，随着前面流量的变化，TaskManager 的内存也随之发生着调整。这个作业默认配置的 TaskManager 内存总大小是 13 万 1 千左右，可以看到伸缩效果符合预期。

针对流量比较平稳的作业，在开启垂直弹性后，内存直接就下降到了合理值，并且持续稳定的运行，没有发生变化。

从整个弹性的结果来说，我们已经节约了用户作业配置内存的 34%，即节约了用户作业配置总内存的三分之一，同时提升了集群堆内存利用率约 10%。
目前，我们的调整策略比较保守，并未推广到所有的作业。随着进一步的推广，这个数据会有更大的提升，优化效果和收益将会更加明显。

平台层治理实践

内存智能建议主要是根据作业历史堆内存使用量及作业画像，为作业推荐合适的 JobManager 和 TaskManager 的内存建议。
目前，我们已经对线上约四分之一的用户作业给出了内存建议。累计可节约内存超过 21TB，预估每月节约成本超过 42w 人民币。

通常，用户在开发 Flink 作业时，需要申请一个实时队列，并配置好所需资源。这个队列按照用户或用户组提供资源。由于每个用户或用户组都有自己的独立队列，因此，用户作业运行更稳定，平台运维成本也较低。
在新的架构设计下，我们让实时队列都统一使用同一个队列。统一实时队列在所有用户和用户组之间共享资源，提升了资源利用率。同时，它也能将整个队列剩余资源作为 buffer 供弹性调度使用。此外，它也降低了新用户开发实时作业时的理解成本。

统一实时队列使所有作业都使用公共队列提交作业，但是由于代码配置的并行度默认是优先级最高的，导致用户在平台上设置的并行度不会生效，进而导致用户作业资源用量不受控。
为了应对这种情况，我们在保证 Chain 不变的前提下，强制作业运行时的并行度不能超过在平台上配置的并行度，从而可以提前校验集群资源是否充足，避免资源的滥用。