01背景

Flink 在网易游戏的资源管理发展主要分为以下三个阶段。第一阶段是在2018年的时候；最开始是使用 Yarn 作为 Flink 作业的资源管理和调度。这个阶段主要是希望能够利用 Yarn 直接复用现有的一些资源，帮助业务快速完成 Flink 作业的部署上线。第二个阶段是在2023年开始了 Flink on K8s 云原生的支持；这个阶段主要是希望能够通过 K8s 接入私有云的弹性资源，同时提高 Flink 作业的资源隔离能力。第三个阶段是在2024年，开始接入公有云的资源，形成了混合云的资源管理体系；这个阶段主要希望在已有的私有云基础上接入公有云资源，满足游戏流量高峰时候突增的资源需求。

早期使用 Yarn 做 Flink 作业的资源管理和调度时，有比较明显的优势和劣势。先看一下它的优势，第一可以复用现有离线资源帮助业务快速的上线。第二它可以做资源的统一管理和调度，因为 Yarn 的所有节点资源，比如 CPU 和 Memory 都会被抽象为 Container ，Flink 在运行任务的时候只需要按需去跟 Yarn 的 Resource Manager 申请对应的 Container 就好了。同时 Yarn 也提供了比较多的调度策略，比如说 Fair 和 Capacity 策略满足不同的调度需求。第三是 Yarn 的容错性也比较好，比如 Flink 作业经常会有偶发的异常退出，在 Yarn 集群里面，基于 Application Master Attempt重试机制可以比较好地帮助 Flink 做快速的 Failover 恢复。

但在使用 Yarn 的过程中也有一些比较明显的劣势和痛点，第一是它的资源粒度比较粗，异构资源调度灵活性低。因为在 Yarn 里，它的最小的资源粒度是 1 core ，有一些小作业是用不满这个资源的。另外如果作业使用 RocksDB StateBackend，需要把作业专门调度到挂载有SSD 的机器上，对 Yarn 集群来说实现比较麻烦。第二个是它的资源隔离能力比较差，在做实时/离线作业混部的时候发现离线作业有一个特点，它启动的时候网络流量特别大，导致机器上其他的 Flink 作业会出现网络异常而退出。第三个是 Yarn 集群环境依赖管理特别复杂，运维成本比较高。比如用户的 PyFlink 作业往往有很多第三方 Python 依赖，这个时候需要管理员帮用户提前把这些依赖部署在节点 Pyenv 环境上帮助用户加速作业的启动。当节点数比较多的时候，运维成本相当高。

所以基于上述 Yarn 的痛点和问题，我们进行了 Flink on K8s 云原生的支持。

02架构演进

接下来分享一下这个阶段在架构上的演进。

Flink on K8s 首先要考虑的是怎么把 Flink 作业在K8s上面部署起来。

社区主要提供了3种模式。第一种是 Standalone 模式，这种模式需要在外部提前创建管理 Flink 的资源，为了避免资源不足往往会从峰值流量去考虑创建比较大的集群，这个时候它的资源利用率是比较低的。同时，这个作业运行完成之后，作业的资源也要自己进行回收，这种方式管理比较复杂，扩展性也比较差。第2种模式是 Native 模式，相比第一种模式，这种模式可以做到让作业按需申请计算资源，但是在这个作业异常或需要升级的时候还是要人工进行处理。第3种模式是 Flink Operator 模式，这种模式可以做到自动管理 Flink 作业以及它的生命周期，作业的启停和升级等。另外 Flink Operator和 Flink 以及 K8s 都做了深度的集成，可以给 Flink 提供比较高级的特性，比如说健康检查以及优化等。这种模式也是社区比较推荐的部署模式，所以最后采用了这种部署模式。

决定了用 Flink Operator 部署模式之后， Flink on K8s 的架构和流程如上图所示。上层是用户的 Flink Streaming、Flink Batch 和 Flink SQL 作业。中间第二层是作业管理，包含有平台的 Lambda 调度器、 Flink Operator服务。第三层，是 K8s 的调度层，同时我们会使用 ZooKeeper 作为高可用服务。因为之前发现在 Flink 作业大规模使用 ConfigMap作为 HA 的时候，它会对 K8s etcd 造成压力从而影响到API service 对外的服务，所以就把这一块单独分离出来使用 ZooKeeper 作为 HA 。第四层是存储层，包括有存储状态数据的 HDFS ，还有存储镜像的镜像仓库以及存储日志的 CubeFS ，最底层的是机器资源池。整体流程如下，Flink 作业在平台启动之后，调度器会收到启动请求，对作业进行编排，也就是 FlinkDeployment CRD的编排。在编码完成之后会提交给 K8s API Server ，由 K8s 的 API Server 把 CRD 信息同步给 Flink Operator 。然后由Flink Operator 基于 Native 的 K8s SDK 去为作业申请创建对应的 Pod 以及其他的 Service、ConfigMap 资源。

在上述 Flink on K8s 架构里，作业其实都是运行在内部的机器资源池上面。但是这部分资源相对来说还是比较有限的，所以和内部的容器团队合作希望接入更多的资源，基于 Virtual Kubelet 去接入了私有云的容器弹性资源。Virtual Kubelet 可以把我们私有云的 K8s 集群作为虚拟节点加入到自己维护的 K8s 控制面上。这样做可以带来以下一些好处；第一是可以兼容原来的 Flink K8s 使用姿势。因为之前在平台上面已经对 Flink 做了 K8s适配，而 Virtual Kubelet 本身又兼容原生的 K8s 接口，所以适配成本比较低；第二是可以享受到低成本高质量的弹性算力；第三是 Serverless 化简化运维, 整个集群里面的计算节点其实只有一个，可以大大减少机器相关的操作，降低运维的压力。

引用了 Virtual Kubelet 之后, Flink 作业的部署架构和流程就发生了一些变化。从图可以看到上层是主控制面，下层是私有云的 K8s 控制面。整体流程会先通过 kubectl去往主控制面提交 FlinkDeployment CRD ，这个时候会被 Flink Operator 服务的 Web hook进行拦截，进行校验的一些操作；没有问题之后会同步回主控制面，主控制面会进行 CRD 资源创建的同时通知到 Operator 服务这边，然后 Operator 继续为 Flink 作业去主控制面申请创建对应的 Deployment、Service、Ingress等。主控制面在为 Deployment 去申请创建对应的 Pod 时会根据平台设置的节点亲和性，把 Pod 调度到  Virtual Kubelet 节点上去。在架构里面可以看到只有 Flink Operator Pod 是运行在主控制面上， Flink 作业的所有 Pod都运行在私有云的 K8s 集群上。

在完成了私有云的接入之后，最后还接入了公有云的资源，形成混合云的资源管理体系。

之所以这样做，其实综合考虑了私有云和公有云的优缺点，私有云的优点比较明显，第一它有比较强的控制权，比如管理员对机器的权限都是比较高，各种操作都很方便。第二是安全性比较高，私有云的机器一般在内网里面，有防火墙还有安全组的安全扫描保护。第三是成本稳定可以预测，在私有云里面资源单价比较固定的，所以计算成本也是能够预测得到。第四是灵活性比较好，可以根据业务的需求做一些定制和优化，但这部分的资源相对来说还是有限的，有赖于 IDC 提供的机器资源，维护成本也比较高。

公有云的优点是理论上它是有无限的弹性资源；其次可以做到按需计费，随用随停；另外就是高效可靠，因为公有云厂商提供的技术支持和运维服务比较高效专业。但如果所有的 Flink 作业都运行在公有云上会涉及到数据安全的问题。所以希望结合两者的优缺点去进行混合云的建设，可以获得以下的优势和能力。第一是弹性伸缩，基于公有云弹性伸缩能力可以把一部分临时的负载弹到公有云上。第二是基于私有云成本稳定可预测以及公有云按时计费的特性做到比较好的成本控制。第三是稳定高效，保证已有作业稳定运行的同时，能够快速满足业务增长对资源的需求。第四是安全敏捷，处理重要敏感的 Flink 作业就可以跑在私有云上去，保障安全，其他作业可以放在公共云上。

接入公有云之后，就形成了混合云的架构。最上层是应用层，还是会有业务的 Flink streaming、 Flink Batch、Flink SQL 这些作业。第二层是作业管理，作业管理包含平台的 Lambda 调度器和 Flink Operator服务；同时会有集群管理，包含有虚拟集群和容量管理，虚拟集群主要是用来满足用户对不同集群以及 Flink 版本的定制化需求和配置。容量管理是对不同项目的配额管理。第三层是容器编排层，现在已经接入了自建 K8s 以及私有云和公有云的 ACK 作为编排调度引擎。用户可以根据业务形态以及它的资源需求选择不同的引擎，比如说业务的上游流量是在阿里云上，他就可以选择 ACK 作为他的编排调度引擎。第四层是共享存储，有 HDFS 以及 OSS，在自建的 K8s 和私有云的环境上，会使用HDFS作为状态存储，在 ACK 环境下会使用 OSS 作为状态的共享存储。最底层的是基础设施，包括网络、计算、存储相关基础设施。同样还会有其他的周边配套服务，包含有不同环境的日志采集、指标上报、告警服务。

03实践挑战

接下来分享一下在演进过程中的实践和挑战。

首先介绍一下在混合云场景下怎么把 Flink 作业给部署起来。前端用户首先要在平台提交作业Jar，然后选择虚拟集群进行启动。 Lambda调度器这时会收到启动请求，把用户提交的作业 Jar上传到 HDFS，同时外部会有定时任务把 Kerberos 的keytab文件同步到 HDFS ，用于后面 Flink Operator 启动作业的时候使用。当 Lambda 调度器在上传完作业Jar之后，就会对作业进行编排，也就是 Flink Deployment 的编排。在这里编排的时候，如果作业选择的是自建 K8s 或者私有云的集群就会在编排里面指定使用 HDFS 作为状态的共享存储，如果是 ACK 的集群就可以选择 OSS 作为存储。 CRD 编排完成后会提交给 K8s 的 API Server 。然后 API Server 把 CRD 的信息同步给 Flink Operator服务，后续就由 Flink Operator服务作为 CRD 的 Controller 为 Flink 作业申请创建对应的Jobmanager Pod， Jobmanager Pod初始化的时候就会通过S3 Proxy 访问到刚刚上传到 HDFS 的作业Jar进行下载和同步。 Jobmanager 启动完成之后会和 API Server 通信去创建作业所需要的 TaskManager Pod ，最后就是通过定时轮询 Flink Rest Service 的方式去跟踪作业的状态。通过这个流程就可以把 Flink 作业统一高效的部署到不同K8s环境上。

Flink 作业在混合云支持之后主要有以下三个应用场景。第一是游戏的 CBT 测试，因为新游戏上线之前，往往要经过比较多轮的 CBT 测试。这些测试数据有流量大周期短的特点，但是都需要用 Flink 作业进行消费，所以每次测试的时候，都需要为这些短期的测试流量，准备较多的计算资源。但在测试完成之后这部分流量又没了，准备的资源就会被闲置，造成浪费。所以有了混合云的支持之后，就可以把这部分的测试流量弹到公有云上，基于公有云的弹性资源去满足测试产生的大量的计算资源的需求，测试完成之后主动释放公有云的资源，减少资源的浪费和机器的成本。 第二个是流量高峰的负载转移，因为游戏流量有比较明显的潮汐现象的，一般是在晚上还有周末的时候流量比较高，其他时候流量比较低。在 Flink 集群水位比较高的时候，如果遇到了游戏搞活动，流量上涨，Flink 作业很有可能会因为资源不够或者资源碎片的问题出现扩容失败，有了混合云的支持之后，我们就可以把高峰时候的一部分负载转移到公有云上，避免 Flink 作业因为资源不足而扩容失败。第三是异地容灾，因为私有云和公有云的机房在不同的地方，当其中一个集群出现比较大规模故障的时候就可以做异地的迁移。

K8s相比于 Yarn，在资源隔离上做得比较好，使得它在服务方面比较有优势。首先介绍一下On K8s 的服务混部。如图所示，我们会对集群上的节点打上不同的标签和污点来划分不同的资源池，比如说这里会划分为计算服务资源池、在线服务资源池。通过 K8s 的节点亲和性、反节点亲和性把 Flink 作业调度到计算服务资源池上。其他的服务，比如说 Hangout 或者 Logstash 就会运行在在线服务资源池上。这样的好处是不同的服务可以共用一套控制面，但是它运行在不同的节点上，资源隔离程度比较高，也有利于 SLA 比较高的在线服务稳定运行。而在计算服务资源池里，在它资源负载比较低的时候，会把一些批作业调度上去。比如说数据湖的优化批作业。充分利用闲时资源，提高资源利用率。

On K8s服务分布只能对运行在 K8s 上的服务进行分布。但还有很多的服务由于各种原因没有运行在 K8s上，比如很多服务没有做容器化的改造，是以进程的方式运行在物理机上。基于裸进程服务混部就可以把这部分资源也利用上。以上图为例子，左上角的节点1，它本身就运行了两个进程服务，把它所消耗的资源，比如说 CPU 、Memory 会映射为同节点上两个优先级比较高的 PlaceHolder Pod，节点上也会运行有其他的 Flink Pod。节点2也是同样道理，在这里可以看到节点1的资源水位比较高，节点2资源水位相对空闲一些，当节点1的进程资源消耗变化的时候，对应的 Placeholder 资源占用也会发生变化，并且触发重新部署。此时K8s 会根据设置的优先级保证高优先级的PlaceHolder Pod的资源得到满足，在资源不足的时候会把低优先级的 Flink Pod 驱逐调度给到其它节点。这里就会把节点1上的 Flink Pod 驱逐到相对比较空闲的节点2上，基于 Flink 内部的 Failover 机制完成作业的恢复。基于进程的 PlaceHolder 映射、K8s 优先级和抢占机制可以实现裸进程服务和Flink作业的混部。

为了更好的利用 K8s 的弹性伸缩能力，平台为 Flink 作业提供了自动扩缩容的功能。如上图所示，我们会把 K8s 还有 Flink 以及 Pulsar的指标通过 VM agent 去采集上报到Prometheus服务，同时还会在内部的 Monitor 指标管理平台设置一些清洗、关联的规则。比如说 Pulsar的 Lag 指标默认没有暴露到 Client 端，这里就可以对它进行清洗转换并和 Flink 指标关联；中间会有名为 Talos 的扩缩容服务去指标服务里面查询需要的指标，结合作业设置的扩缩容算法，比如说 DS 2 或者 Lag 算法；同时感知集群的状态感知，去为作业计算最新的资源配置和选择合适的集群。当作业的资源配置发生变化的时候，会向 Lambda 调度器发起扩缩容请求，并发送最新的资源配置。Lambda 调度器会把最新的资源配置提交给K8s的API Server 去完成 Flink 作业的资源伸缩。目前已经支持了 Task Manager 水平和垂直的伸缩，同时也支持了 JobManager 的垂直伸缩。Talos 扩缩容服务还会对作业扩缩容之后的状态进行跟踪，保证它在扩缩容之后最少能够完成一次checkpoint。如果中间出现了问题，会通过内部的 BlackHole 告警平台去通知到用户或者管理员进行处理。

从这张图可以看出，作业开启了自动扩缩容之后，作业资源是会跟着流量变化而变化的。在流量上涨的时候，会自动去扩容申请更多的资源提高消费能力。在流量下来的时候也会自动的释放一些资源，这样给业务带来的好处是可以减少因为数据延迟而需要人力介入的情况。这部分的人力成本大概可以减少30%。

同时，作业在开启了自动扩缩容之后，它的资源利用率也会有所提高，计算成本也会发生下降，这部分成本大概降低20%。

刚才说到会在 K8s上面混部一些批作业，其中有一部分批作业是来自于数据集成服务的共享算子模块，这些算子都是运行在 Flink 作业里，它既能以流模式来运行，也能以批模式来运行。共享算子主要是应用在数据集成上，所以它整体的流程和设计体现了ETL中数据抽取、清洗转换和输出的过程。

如图所示，Source算子可以通过 connector 去连接上游的Kafka、Pulsar或者Iceberg数据源；之后数据会经过 Filter 算子根据自定义的规则进行过滤；然后基于 Parser 和 Extender算子可以对数据进行结构化以及扩展；随后经过 Transform算子的时候会调用业务编写的UDF/UDTF进行数据处理，最后经过 Loader 算子的时候会输出到下游的各个系统。基于这样一个流程，业务的实时流作业和历史批作业就可以共用一套代码和规则，业务的开发和管理的成本明显降低的，对用户来说也不太容易出错。而业务的自定义逻辑，可以使用 UDF/UDTF 来表达，自由度比较高。同时结合自动扩缩容可以在低峰的时候会拉起一部分批作业去运行。

在K8s集群中，Flink Operator 服务的稳定性、可用性对于 Flink 作业起着重要的作用。在实际生产中我们对此也做了一些优化，首先看一下内存方面的优化。

当时的背景是在集群作业规模不变的情况下， Operator 的内存持续的增长，最后出现 OOM Kill的情况。从图中可以看到内存增长到了35G。但是当时集群作业规模比较小，只有1000个左右，这么大的内存使用明显是不合理的。

经过进一步Heap dump 和 GC 分析，发现堆内存使用正常。

怀疑此前内存增长应该是来自堆外内存，尝试对这部分内存进行分析。从当时监控可以看到，服务的总内存在4小时里增长了95MB左右。

后面我们使用pmap做了内存分析，发现内存增长主要来源于前面栈内存地址段，从74MB到165MB，增长了大概91MB，说明前面95MB的增长大部分都来自于这一块区域。而且它有很多64MB的内存段，这部分很符合线程Arena内存块的大小，后续结合CPU profile 的分析，发现内存增长确实是发生在线程创建的内存分配阶段。而Operator 默认使用 glibc2.3 做内存分配，基本上可以确认是glibc在多线程场景下的内存碎片问题，所以后续升级为 Jemalloc 去优化这一块内存的使用。

对 Operator 除了内存方面的优化，还做了一些其他优化，比如Kerberos keytab 文件的同步优化。之前一般是把keytab文件同步到宿主机再挂载到 Operator 容器里，在作业启动的时候使用。但在混合云场景下要把文件同步到外部宿主存在很多限制，也有安全风险，所以基于 HDFS + S3 Proxy + Sidecar Container的方式做了优化。

具体的做法会先把 keytab 文件同步到内部的HDFS，通过 S3 proxy 可以被Operator的 Sidecar Container 以 S3A 协议的方式进行下载。另外一个优化是Operator在为Flink作业设置Ingress路由规则的时候，发现它只能路由到ClusterIP 类型的Service上。而在私有云场景下，其实是优先使用 HeadLess Service 类型。所以在这一块也进行了优化，让Operator 设置Ingress规则的时候，支持路由到HeadLess/ClusterIP Service类型。

04总结和展望

最后是我们的总结和展望。

先总结下目前生产的现状以及上云的情况，现在生产上面大概是有1万+的作业，资源规模大概是十万+的CPU core，目前已经完成了50%作业的上云的迁移，每天处理的数据量，大概是十万亿。

目前 Flink 上云带来的收益，主要分为以下三个方面。第一个是基于服务混部和自动扩缩容可以大幅提高集群的资源利用率，使得机器资源大概减少30%；另外基于 K8s 的细粒度资源管理，很多小作业在上云之后就可以设置比较小的CU，所以大部分的小作业计算成本是40%的下降，平均也有20%的下降；而 Flink 上云之后带来了更多廉价的弹性资源，满足了业务增长对计算资源的需求。

未来工作主要分为以下三个方面来开展，第一是性能和规模的提升，继续提升上云的规模，满足大规模算力场景下对调度性能和吞吐能力的要求；第二个是 Flink 流批一体能力的推广；第三个是结合机器学习为用户提供智能运维的能力，减少人力介入的成本。