OPPO实时计算平台基于云原生的作业弹性伸缩设计与实践

文章大纲：

一、背景

二、技术方案

2.1 整体架构
2.2 方案详述：

1.纵向伸缩：

2.横向伸缩

3.云原生独立部署模式

4.资源伸缩协调器

三、方案实践及效果

3.1 弹性伸缩
3.2 自动诊断

四、后续规划

一、背景

Flink目前是业界主流的流批一体的计算引擎。OPPO基于Flink构建的实时计算平台总共运行5000+作业，服务于公司二十几条业务线。我们从2021年开始着手计算上云的工作，目前已经有1/5的作业稳定运行在k8s上。

实时计算任务有以下特点：

资源初始固定。任务需要在提交之前明确资源用量且作业运行过程中不会自动调整
任务常驻。由于数据源多为无界流式数据。一旦有新的流数据进入任务，它会立刻发起并进行一次计算任务，因此整个过程是持续进行的；
负载呈周期性变化。实时作业流量和负载会随着时间的变化出现明显的波峰波谷

由于以上特点带来了以下问题：

调优困难。通常，用户需要花费大量的时间进行作业调优。例如，新上线一个作业，需要考虑如何配置该作业的资源、并发数、TaskManager个数及大小等
资源用量无法匹配负载的变化。由于实时作业的负载往往随着流量的变化而变化，初始设置的资源量容易过多或太少，从而造成资源浪费或者资源不足而导致作业延时。而解决这个问题用户往往需要手动重启作业再次设置资源用量，而这种操作繁琐的同时也是滞后的

下图展示的是某作业流量和资源使用情况：

由图中看出作业呈明显的波峰波谷的情况。

二、技术方案

2.1 整体架构

为了解决上述问题，我们设计实现了一种基于云原生的Flink自动扩缩容的方案。整体方案以kubernete为基座，自研的"自动诊断 + 弹性伸缩"为核心。其中自动诊断模块由 SmartResource 负责，弹性伸缩的能力承接由 Flink 计算框架负责。

SmartResource负责根据作业上报的指标、用户自定义的监控规则、全链路诊断规则来判断当前作业的运行的健康度，作业是否需要伸缩均源于此。
Flink 计算框架使用自研自适应调度器RescaleMonitor，自动感知资源的变化，动态改变算子的并行度，并重新调度作业。

架构图如下：

整体流程如下：

用户配置自定义的伸缩策略到 SmartResource
SmartResource 将相关的策略转化为告警规则并配置到云监控上
云监控的数据来自 Flink 集群上报的作业相关的各种指标
云监控触发告警时回调指定的 SmartResource 接口，SR 根据告警信息，在任务链路打上优化建议Tag；此外，需要调整资源的，结合用户配置的伸缩规则自动计算需要伸缩的副本数，并告知后端服务
后端服务收到伸缩请求后，再次将此请求在诊断联调上应用一遍来确认是否是个有效的请求，如果是会在数据库中存储一个状态为 pending 的伸缩请求
当作业的 Checkpoint 完成后，会通知 ostream backend 然后开始执行Flink计算资源的伸缩

2.2 方案详述：

Flink 集群往往由一个 JobManager 配合多个 TaskManager 构成。在纵向上也就是在单个 TaskManager的参数配置上，主要关注 CPU、内存两个方面。在横向上主要关注 TM 的数量。同时 TM 数量也代表了整个集群可用 slot 个数，我们在伸缩的时候也是从纵横向两个方面考虑。

1. 纵向伸缩：

纵向的伸缩主要依赖于 Pod 在声明资源的时候设置 request 和 limit。当我们在创建 TaskManager的时候，在用户配置的基础资源量上额外设置最小资源量(降低下限)，最大资源量会略大于用户配置的基础资源量(提高上限)。作业的负载波动的时候，单个 Pod(TM)占用的资源也会在 request 和 limit 之间波动。这样在纵向上，减少资源的固定占用。也能很好的解决堆外内存占用突高引起的容器OOM的问题。

示例配置如下：

2. 横向伸缩

横向的伸缩主要表现在 TaskManager 数量的增减上，TaskManager的增减同样反应整个集群可用的资源数上(slot)，但是这里两个问题需要解决：

当 TaskManager增减的时候 JobManager 以及已经申请的 TaskManager不能丢失，也就是怎么保持已申请的资源？
当 TaskManager增减的时候 JobManager 要快速感知新增的 slot，也就是如何感知资源变化并快速调度Job？

2.2.1 云原生独立部署模式

针对横向伸缩第一个问题我们设计了基于 k8s 的独立的作业部署式：kubernetes-standalone-application。此模式会独立部署副本为1的JobManager，副本为n（根据用户指定的并发自动计算）的TaskManager，两者都是采用kubernetes deploy模式部署。由于TM的生命周期不在由JM控制，所以我们可以在外部控制TM的数量，这位我们后续的弹性伸缩打下了基础。

此模式是我们使用云原生的方式封装了Flink自带的standalone模式，在原生的standalone下，假如要部署JobManager、TaskManager的时候需要构建两个非常复杂的的kubernetes yaml文件(例如使用deploy)，在部署、平台化方面存在诸多不便之处。我们在原有的命令行基础上支持了standalone的部署模式的支持，兼容现有客户端的各种参数命令。

新的部署模式图如下：

特点如下：

云原生一键部署；
自动推断资源参数；
自研KSA Resourcemanager管理已申请资源；
与现有管理平台无缝集成
通过k8s的watch机制自动追踪资源状态；

部署命令如下：

2.2.2 资源伸缩协调器

针对横向伸缩第二个问题我们设计了RescaleMonitor，它是一个伸缩监视器，主要用于快速感知资源的变化并决定是否需要重新调度作业。它首先计算作业所需的资源，计算完资源后，会一直等到可用资源稳定下来，一旦资源稳定，会开始确定作业的实际并行度, 一旦确定了并行度并且执行与可用槽匹配，调度程序就会部署执行。

快速部署。只要有外部资源增加的时候，RescaleMonitor会判断是否满足调度作业的最小资源，如果满足就会立刻部署 jobgraph，不满足会一直等待。
快速失败。当有外部资源减少的时候，依赖于Kubernetes的watch机制可以快速感知资源的变化，无需等待资源超时。此处我们以pod名字作为Resource ID，可以在内部快速定位资源建立连接，进而执行各种操作。
安全恢复。当有外部资源增减的时候不会立即执行，Pending下来等待checkpoint完成的时候会检查是否存在资源增减的Request，如果由才会立即执行重新部署JobGrap并从当前完成的checkpoint恢复，这样在保证作业不丢的情况下，尽量减少重复消费数据的可能。
可固定并发。支持固定某些算子的最大并发，这在作业类似是消费 kafka 的时候特别有用。
UI联动。当资源变化的时候会实时通知web前端实时显示最新的资源用量。

调度流程示意图：

三、方案实践及效果

3.1 弹性伸缩

我们在平台提供了用户开启弹性伸缩的前端页面，也给出了常用的默认设置，如下图所示，在CPU利用率大于70%并持续5分钟的时候，开启扩容，在CPU利用率小于30%并持续5分钟的时候，开启缩容。

弹性伸缩效果图如下，通过途中可以看出，作业在cpu利用率低时，自动降低的并发。

我们记录了每次伸缩的事件，包括时间、触发原因，伸缩前后的资源等，方便平台跟用户跟踪资源的变化和排查问题。

3.2 自动诊断

以某作业为例，此次作业的存在明显的波谷，且资源利用率很低。

通过我们的智能诊断服务可以自动优化整个作业的资源配比，在不重启作业的情况下完成TM的动态调整。

优化后的效果如下

经过我们计算经过自动诊断加弹性伸缩可以为业务带来至少20%的成本降低。

其实成本降低只是收益一方面，让流量更加契合资源，让诊断更加智能，让作业稳定高效的运行，才能更好的服务于下游业务。也是我们实时计算平台服务的宗旨。

四、后续规划

后续我们在继续优化弹性伸缩效果的基础上，继续朝着下面几个方向努力。

batch任务的自适应调度
基于机器学习的自动化诊断
流批一体的错峰调度

附录

https://nightlies.apache.org/flink/flink-docs-master/zh/docs/deployment/elastic_scaling/
https://cwiki.apache.org/confluence/display/FLINK/FLIP-159%3A+Reactive+Mode
https://cwiki.apache.org/confluence/display/FLINK/FLIP-160%3A+Adaptive+Scheduler
https://kubernetes.io/zh-cn/docs/tasks/run-application/horizontal-pod-autoscale/

作者简介

John，OPPO实时计算平台高级研发工程师，Apache Flink Contributor，长期专注于大数据计算领域。