关于后端:Flink-数据集成服务在小红书的降本增效实践

摘要：本文整顿自实时引擎研发工程师袁奎，在 Flink Forward Asia 2022 数据集成专场的分享。本篇内容次要分为四个局部：

1. 小红书实时服务降本增效背景
2. Flink 与在离线混部实际
3. 实际过程中遇到的问题及解决方案
4. 将来瞻望

点击查看原文视频 & 演讲 PPT

一、小红书实时服务降本增效背景

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 能够无效缩小对本地磁盘的局部依赖，晋升资源利用率，助力云原生架构。

点击查看原文视频 & 演讲 PPT

更多内容

流动举荐

阿里云基于 Apache Flink 构建的企业级产品 - 实时计算 Flink 版现开启流动：
0 元试用 实时计算 Flink 版（5000CU* 小时，3 个月内）
理解流动详情：https://click.aliyun.com/m/1000372333/