摘要：本文整理自实时引擎研发工程师袁奎，在 Flink Forward Asia 2022 数据集成专场的分享。本篇内容主要分为四个部分：
- 小红书实时服务降本增效背景
- Flink 与在离线混部实践
- 实践过程中遇到的问题及解决方案
- 未来展望
一、小红书实时服务降本增效背景
1.1 小红书 Flink 使用场景特点
小红书的 Flink 特点包含以下三条：
- 第一，云原生，复杂的多云、海内外架构。小红书从成立之初就将所有的技术体系全部搭建在公有云上，是真正意义上云的原住民。
我们与多家云厂商都有合作，比如 AWS，腾讯云，华为云，阿里云等等。经过多年的发展，业务数据也分布到了不同的云厂商下。云原生本身就会带来天然的好处，比如资源隔离和扩展都非常容易。 - 第二，数据集成链路较长，作业存在高峰期资源互相抢占的现象。以数据集成为例，在多云体系架构下，数据要经常进行跨云的传输，所以数据集成任务是重要且不可或缺的。我们在过去搭建了 Flink 的数据集成的独占集群，但随着数据集成任务的增多，出现了越来越多资源抢占的现象。
因为 Flink 集成任务都是批任务，大部分会在凌晨同时集中运行，就会出现一部分任务因抢占不到资源而失败的情况。同时整个资源池的整体利用率也比较低，因为白天批任务运行的比较少，这个时候资源是空闲的。 - 第三，数据集成的高优、低优作业均以 Flink 流模式引擎运行。有一些历史原因，一个是因为早期 Flink 版本的批模式引擎还不成熟，另外一个是流模式比较简单，它速度快，且不用考虑中间数据落盘的问题。在资源比较宽裕的情况下，它是更优的选择。
1.2 小红书 Flink 数据集成服务
小红书典型的数据集成类型有很多种，比如 Hive to Clickhouse、Hive to Doris、Hive to MySQL、Mongo to Hive 等等。
上图右侧是是一张 Top 图，一个数据源进行了一次 Mongo 的 Lookup Join，分为两个流写入到下游，这就是一个典型的 Flink 数据集成任务。
1.3 降本增效的大环境要求
随着小红书的发展，基础设施越来越完善，资源的使用也更加规范化。过去那种资源野蛮申请的时代已经结束，现在逐渐重视集群的 CPU 利用率。
在这样的背景下，我们来看 Flink 的资源集群。一方面我们现在的 Flink 资源集群主要采用独占模式，部分小资源池任务比较少，容易产生资源碎片，存在资源浪费。另外一方面 Flink 集成任务的集群，在晚上存在资源抢占的现象，而在白天又因为资源空闲而利用不起来，会造成整体的资源利用率不高。
针对以上两个问题，有什么解决方法，来提升整体资源的利用率呢？可以分为如下两点：
- 第一，如何规避小规模集群。我们可以将小规模集群进行合并，然后配合 K8s 的 Resource Quota 进行资源隔离。除此之外，我们还有一个更好的解决方案，即使用容器团队提供的在离线混部集群。将小规模集群的任务迁移到在离线混部集群中，然后将小规模集群的资源释放掉。
- 第二，如何减少高峰时期的资源抢占。从平台的角度来考虑，我们可以优化资源的调度，细化任务的优先级。从 Flink 引擎的角度来考虑，我们可以推广 Flink 的批模式引擎，因为批模式引擎对资源的要求更低。但我们的切入点不太一样，我们是从资源角度来考虑的。
1.4 降本增效视角下的 Flink 流模式/批模式对比
接下来我们从资源角度对比一下 Flink 的流模式和批模式。
Flink 的流模式引擎运行的时候没有阶段的概念，数据以 pipeline 的方式进行流转。这就要求所有的算子和并发的资源都要实时准备就绪，程序才能正常运行。而对于批模式引擎来说，任务被划分到几个阶段，上一个阶段运行结束后才能运行下一个阶段，且只需要部分算子和并发获取到资源就可以运行了。
从另外一个角度来看，部分聚合类型的批任务，在流模式运行的时候，会不可避免地引入 State 和 Watermark，这就需要更多的CPU和内存资源。而在批模式引擎下不需要 State 和 Watermark，仅需要 Shuffle 中间数据，这对磁盘的要求也很高，但磁盘相对于 CPU 和内存来说更加便宜。
这就是资源视角下流模式和批模式的对比，也是我们将批任务从流模式切换到批模式来运行的一些考虑。
二、Flink 与在离线混部实践
2.1 在离线混部的 K8s 集群
首先来看看什么是在离线混部。一般公司都会有两种类型的服务。一种是在线服务，它的特点就是运行时间长，服务流量和资源利用率具有潮汐性。也就是在白天使用人数多的时候，资源利用率就会高，流量也会高，而到了晚上使用人群数量降下来之后，资源利用率也会降下去。另外一种是离线作业。它只会运行一段时间，运行期间资源利用率非常高，一般也是时延不敏感的，只要在一个时间点之前运行结束之后资源就会空闲下来。
所谓在离线混部就是指将在线服务空闲的资源匀给离线作业使用，提升资源的整体利用率。对离线业务来说，能极大降低这资源的使用成本。在离线任务混跑期间，需要保护在线服务，可能会对离线业务的运行进行资源压制等操作。
上图是在离线混部集群的示意图。容器团队将各个在线服务集群的空闲资源收集起来，组成一个资源集群。从用户角度，只能看到一些虚拟节点，但实际上每个虚拟节点背后都对应着一到多个真正的资源节点。对用户来说，虚拟集群的使用和真正独占集群是一样的，唯一不一样的是，虚拟节点的资源可能在不断变化。容器团队提供了在离线混部集群，而我们正好有离线任务，且有资源利用率的压力，算是一拍即合。
2.2 适合在离线混部的离线任务特点
哪些任务适合迁移过去，主要的考虑的特点有以下三个：
- 第一个是迁移过去的任务必须是非延时敏感的，因为在离线混部集群会压缩离线资源，离线任务运行的时间可能会更长。
- 第二个是任务要具有潮汐的特性，需要选择刚好在资源空闲时大量运行的离线任务迁移过去。一般来说，在线服务在晚上资源比较空闲，而离线任务都是集中在晚上运行比较多，这一点比较契合。
- 第三个是具有容错能力，因为在离线混部可能会压缩离线任务的资源，并对 Pod 进行驱逐，所以需要任务具有一定容错能力。
2.3 适合在离线混部的 Flink 任务
对批任务来说，由于 Pod 可能被驱逐掉，当被驱逐的时候，在其他节点上拉起就有可能重新消费数据，造成数据的重复，所以我们要选择 Sink 端支持幂等插入或不在意重复数据的批任务迁移。对批模式引擎，我们要尽可能让所有算子 chain 到一起，选择这一部分的任务迁移。因为算子如果不 chain 到一起，就会进行中间数据的落盘，这样就会对资源节点的要求更高。尽量选择在夜间大量运行的批任务迁移，因为在离线混部集群在晚上资源比较空闲。一般在离线混部集群不适合上流任务，但因为它在白天会有一些空闲资源能够支持一部分的流任务运行，所以我们也选择迁移一部分低优的流任务，且这部分流任务需要能够容忍 Fail Over，允许一段时间的延迟。
2.4 Flink 与在离线共建
首先我们会部署一个 Flink 的独占集群，它上面没有独占的节点，然后容器团队将虚拟节点部署到我们的独占集群中。虚拟节点背后对应着一个 controller 和真正的资源节点，当我们提交任务时，只需要将任务提交给虚拟节点，deployment 就会在虚拟节点上拉起 JobManager 的 Pod。最后这个创建过程会被虚拟节点的 controller 下发到背后真正的资源节点上执行。
我们采用的是 Flink Native K8s 的方式，所以 TaskManager 由 JobManager 拉起。这个创建过程和 deployment 的创建过程一样，也会被虚拟节点下发到真正的资源节点去执行。也就是说最终 JobManager 和 TaskManager 的 Pod 都运行在背后的资源节点上，在虚拟节点上只有 Pod 的一份镜像。对于 Configmaps、Service、Ingress 等 K8s 资源，它的源数据都存在 ETCD 中，只需要同步一部分过去就可以了。
通过这种方式，我们可以在 Flink 独占集群正常提交任务，且能正常通过 kubectl 命令操作 Pod，对我们来说使用在离线的虚拟集群就和使用一个普通的 flink 独占集群是一样的。当然实现过程中有一些问题，比如 JobManager 和 TaskManager 分属于两个集群，他们之间如何进行通信，日志和监控指标如何采集等等，这些都是一些工程实现上的问题，这里就不再赘述了。
三、实践过程中遇到的问题及解决方案
最后一部分就是我们在实践过程中遇到的一些问题，作为云的原住民，这里问题也聚焦于我们在云原生上遇到的一些问题和解决方案。
3.1 避免宿主机上临时数据文件的残存
第一个问题，如何避免宿主机上临时数据文件的残存。使用过 K8s 容器技术的人都会遇到这样的问题，默认情况下启动一个容器，容器中的临时数据文件都存在 docker 盘中。如果临时数据文件过大就会影响 docker 的运行稳定性，这个时候我们可以在容器中挂载另外一块数据盘，让临时数据文件写到这块数据盘中，这样就不会影响 docker 的运行稳定性了。
在 K8s 里挂载数据盘一般都通过 hostPath volume 的挂载方式，这种方式的好处是可以指定一个宿主机的挂载目录，挂载方式简单，但 hostPath 挂载方式依赖程序本身临时文件的清理逻辑。如果 Pod 异常退出，比如遇到了 OOM 被 K8s Kill 掉了，此时临时数据文件的清理逻辑还没来得及执行 Pod 已经结束掉了，那么这个临时数据文件就会残存在宿主机上。当残存的文件越来越多，占满了整个数据盘，就会影响任务运行的稳定性。那么我们是如何解决的呢？
K8s 有一种挂载方式叫 emptyDir，它与 Pod 同生命周期。所以无论 Pod 是正常结束还是异常结束，只要结束之后 emptyDir 挂载目录中的临时数据文件都会被清理掉，这就降低了对程序清理逻辑的依赖。
这里有一点需要要注意，emptyDir 不能指定挂载目录，默认使用 kubelet 工作目录存储。一般这个目录在系统盘里，如果不做任何处理，临时文件写入系统盘就有可能会影响系统运行的稳定性，所以一般我们要在开机的时候，更换 kubelet 的工作目录到另外一块数据盘。
3.2 批模式在云原生场景下的 OOM 问题
第二个问题，批模式在云原生场景下的 OOM 问题。这个任务在流模式引擎运行的非常顺畅，但转换到批模式引擎运行之后就会频繁出现 OOM 问题。
这个任务在 chain 之后依然有两个算子，也就是说中间会进行一次数据的 Shuffle，OOM 就发生在写 Shuffle 数据的这个阶段。从上图右上角的监控图，可以明显看到两个阶段，第一个阶段是写 Shuffle 数据的阶段，有一些 work-set 飙升的情况，一旦超过容器限制就会触发 OOM Kill。
出现这种情况，首先我们首先从 Flink 的 webui 上观察堆内存使用情况，目前看堆内存的使用是正常的，从 GC 监控界面也可以看到 GC 情况是正常的。那么我们怀疑可能是堆外内存出现了泄漏，于是我们进入 Pod 里面通过 pmap 命令查看 RSS 的使用情况。也就是右下角的这张图，可以看到 RSS 也是正常的，且 RSS 只有 7G 左右，没有达到 20G 的限制，也就可以说明不是堆外内存泄露导致的。
到这里答案其实已经呼之欲出了。work-set 指标可以简单理解为 RSS+Page Cache，RSS 是正常的，work-set 又出现飙升的情况，所以我们就可以怀疑是 Page Cache 造成的 OOM。
顺着这个思路，我们登录到机器节点上去查看机器日志。如上图所示，我们找到了一个调用栈，可以看到是由于申请 Page Cache 造成的 OOM。实际上就是云盘的性能不足，在 Shuffle 数据时瞬间大量写 Page Cache，不能及时将数据刷到磁盘，导致内存超用，触发 OOM Kill。
我们有一个临时的解决方案。增加 Pod 数量，减少单个 Pod 处理的数据量，然后尽量让 Pod 分布到不同的机器节点上，降低机器节点的压力。或者升级机器内核，通过调整内核参数进行限流。除此之外，我们还可以从 Flink 引擎本身着手，在 Shuffle 数据阶段直接进行限流。
四、未来展望
未来小红书将要探索的方向，主要包含以下三部分。
- 第一，批模式应用深入挖掘。我们希望能够深入用户，挖掘更多的批模式引擎的使用场景，真正推广 Flink 的流批一体。
- 第二，配合使用 K8s 的 Resource Quota 功能，将业务方的多个小集群进行合并，减少机器的资源碎片问题。
- 第三，Serverless 是批模式引擎在云原生环境下部署的一个重要目标，但是强行部署为 serverless 意味着如果 pod 被 Kill 掉中间数据就会被清理，会影响任务的故障恢复，这个时候 remote Shuffle Service 的价值就体现出来了，使用 Remote Shuffle Service 可以有效减少对本地磁盘的部分依赖，提升资源利用率，助力云原生架构。