Flink 数据集成服务在小红书的降本增效实践

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

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

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

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

1.1 小红书 Flink 使用场景特点

1

小红书的 Flink 特点包含以下三条:

  • 第一,云原生,复杂的多云、海内外架构。小红书从成立之初就将所有的技术体系全部搭建在公有云上,是真正意义上云的原住民。

    我们与多家云厂商都有合作,比如 AWS,腾讯云,华为云,阿里云等等。经过多年的发展,业务数据也分布到了不同的云厂商下。云原生本身就会带来天然的好处,比如资源隔离和扩展都非常容易。

  • 第二,数据集成链路较长,作业存在高峰期资源互相抢占的现象。以数据集成为例,在多云体系架构下,数据要经常进行跨云的传输,所以数据集成任务是重要且不可或缺的。我们在过去搭建了 Flink 的数据集成的独占集群,但随着数据集成任务的增多,出现了越来越多资源抢占的现象。

    因为 Flink 集成任务都是批任务,大部分会在凌晨同时集中运行,就会出现一部分任务因抢占不到资源而失败的情况。同时整个资源池的整体利用率也比较低,因为白天批任务运行的比较少,这个时候资源是空闲的。

  • 第三,数据集成的高优、低优作业均以 Flink 流模式引擎运行。有一些历史原因,一个是因为早期 Flink 版本的批模式引擎还不成熟,另外一个是流模式比较简单,它速度快,且不用考虑中间数据落盘的问题。在资源比较宽裕的情况下,它是更优的选择。

1.2 小红书 Flink 数据集成服务

2

小红书典型的数据集成类型有很多种,比如 Hive to Clickhouse、Hive to Doris、Hive to MySQL、Mongo to Hive 等等。

上图右侧是是一张 Top 图,一个数据源进行了一次 Mongo 的 Lookup Join,分为两个流写入到下游,这就是一个典型的 Flink 数据集成任务。

1.3 降本增效的大环境要求

3

随着小红书的发展,基础设施越来越完善,资源的使用也更加规范化。过去那种资源野蛮申请的时代已经结束,现在逐渐重视集群的 CPU 利用率。

在这样的背景下,我们来看 Flink 的资源集群。一方面我们现在的 Flink 资源集群主要采用独占模式,部分小资源池任务比较少,容易产生资源碎片,存在资源浪费。另外一方面 Flink 集成任务的集群,在晚上存在资源抢占的现象,而在白天又因为资源空闲而利用不起来,会造成整体的资源利用率不高。

4

针对以上两个问题,有什么解决方法,来提升整体资源的利用率呢?可以分为如下两点:

  • 第一,如何规避小规模集群。我们可以将小规模集群进行合并,然后配合 K8s 的 Resource Quota 进行资源隔离。除此之外,我们还有一个更好的解决方案,即使用容器团队提供的在离线混部集群。将小规模集群的任务迁移到在离线混部集群中,然后将小规模集群的资源释放掉。
  • 第二,如何减少高峰时期的资源抢占。从平台的角度来考虑,我们可以优化资源的调度,细化任务的优先级。从 Flink 引擎的角度来考虑,我们可以推广 Flink 的批模式引擎,因为批模式引擎对资源的要求更低。但我们的切入点不太一样,我们是从资源角度来考虑的。

1.4 降本增效视角下的 Flink 流模式/批模式对比

5

接下来我们从资源角度对比一下 Flink 的流模式和批模式。

Flink 的流模式引擎运行的时候没有阶段的概念,数据以 pipeline 的方式进行流转。这就要求所有的算子和并发的资源都要实时准备就绪,程序才能正常运行。而对于批模式引擎来说,任务被划分到几个阶段,上一个阶段运行结束后才能运行下一个阶段,且只需要部分算子和并发获取到资源就可以运行了。

从另外一个角度来看,部分聚合类型的批任务,在流模式运行的时候,会不可避免地引入 State 和 Watermark,这就需要更多的CPU和内存资源。而在批模式引擎下不需要 State 和 Watermark,仅需要 Shuffle 中间数据,这对磁盘的要求也很高,但磁盘相对于 CPU 和内存来说更加便宜。

这就是资源视角下流模式和批模式的对比,也是我们将批任务从流模式切换到批模式来运行的一些考虑。

二、Flink 与在离线混部实践

2.1 在离线混部的 K8s 集群

6

首先来看看什么是在离线混部。一般公司都会有两种类型的服务。一种是在线服务,它的特点就是运行时间长,服务流量和资源利用率具有潮汐性。也就是在白天使用人数多的时候,资源利用率就会高,流量也会高,而到了晚上使用人群数量降下来之后,资源利用率也会降下去。另外一种是离线作业。它只会运行一段时间,运行期间资源利用率非常高,一般也是时延不敏感的,只要在一个时间点之前运行结束之后资源就会空闲下来。

所谓在离线混部就是指将在线服务空闲的资源匀给离线作业使用,提升资源的整体利用率。对离线业务来说,能极大降低这资源的使用成本。在离线任务混跑期间,需要保护在线服务,可能会对离线业务的运行进行资源压制等操作。

7

上图是在离线混部集群的示意图。容器团队将各个在线服务集群的空闲资源收集起来,组成一个资源集群。从用户角度,只能看到一些虚拟节点,但实际上每个虚拟节点背后都对应着一到多个真正的资源节点。对用户来说,虚拟集群的使用和真正独占集群是一样的,唯一不一样的是,虚拟节点的资源可能在不断变化。容器团队提供了在离线混部集群,而我们正好有离线任务,且有资源利用率的压力,算是一拍即合。

2.2 适合在离线混部的离线任务特点

8

哪些任务适合迁移过去,主要的考虑的特点有以下三个:

  • 第一个是迁移过去的任务必须是非延时敏感的,因为在离线混部集群会压缩离线资源,离线任务运行的时间可能会更长。

  • 第二个是任务要具有潮汐的特性,需要选择刚好在资源空闲时大量运行的离线任务迁移过去。一般来说,在线服务在晚上资源比较空闲,而离线任务都是集中在晚上运行比较多,这一点比较契合。

  • 第三个是具有容错能力,因为在离线混部可能会压缩离线任务的资源,并对 Pod 进行驱逐,所以需要任务具有一定容错能力。

2.3 适合在离线混部的 Flink 任务

9

对批任务来说,由于 Pod 可能被驱逐掉,当被驱逐的时候,在其他节点上拉起就有可能重新消费数据,造成数据的重复,所以我们要选择 Sink 端支持幂等插入或不在意重复数据的批任务迁移。对批模式引擎,我们要尽可能让所有算子 chain 到一起,选择这一部分的任务迁移。因为算子如果不 chain 到一起,就会进行中间数据的落盘,这样就会对资源节点的要求更高。尽量选择在夜间大量运行的批任务迁移,因为在离线混部集群在晚上资源比较空闲。一般在离线混部集群不适合上流任务,但因为它在白天会有一些空闲资源能够支持一部分的流任务运行,所以我们也选择迁移一部分低优的流任务,且这部分流任务需要能够容忍 Fail Over,允许一段时间的延迟。

2.4 Flink 与在离线共建

10

首先我们会部署一个 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 避免宿主机上临时数据文件的残存

11

第一个问题,如何避免宿主机上临时数据文件的残存。使用过 K8s 容器技术的人都会遇到这样的问题,默认情况下启动一个容器,容器中的临时数据文件都存在 docker 盘中。如果临时数据文件过大就会影响 docker 的运行稳定性,这个时候我们可以在容器中挂载另外一块数据盘,让临时数据文件写到这块数据盘中,这样就不会影响 docker 的运行稳定性了。

在 K8s 里挂载数据盘一般都通过 hostPath volume 的挂载方式,这种方式的好处是可以指定一个宿主机的挂载目录,挂载方式简单,但 hostPath 挂载方式依赖程序本身临时文件的清理逻辑。如果 Pod 异常退出,比如遇到了 OOM 被 K8s Kill 掉了,此时临时数据文件的清理逻辑还没来得及执行 Pod 已经结束掉了,那么这个临时数据文件就会残存在宿主机上。当残存的文件越来越多,占满了整个数据盘,就会影响任务运行的稳定性。那么我们是如何解决的呢?

K8s 有一种挂载方式叫 emptyDir,它与 Pod 同生命周期。所以无论 Pod 是正常结束还是异常结束,只要结束之后 emptyDir 挂载目录中的临时数据文件都会被清理掉,这就降低了对程序清理逻辑的依赖。

这里有一点需要要注意,emptyDir 不能指定挂载目录,默认使用 kubelet 工作目录存储。一般这个目录在系统盘里,如果不做任何处理,临时文件写入系统盘就有可能会影响系统运行的稳定性,所以一般我们要在开机的时候,更换 kubelet 的工作目录到另外一块数据盘。

3.2 批模式在云原生场景下的 OOM 问题

12

第二个问题,批模式在云原生场景下的 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。

13

顺着这个思路,我们登录到机器节点上去查看机器日志。如上图所示,我们找到了一个调用栈,可以看到是由于申请 Page Cache 造成的 OOM。实际上就是云盘的性能不足,在 Shuffle 数据时瞬间大量写 Page Cache,不能及时将数据刷到磁盘,导致内存超用,触发 OOM Kill。

我们有一个临时的解决方案。增加 Pod 数量,减少单个 Pod 处理的数据量,然后尽量让 Pod 分布到不同的机器节点上,降低机器节点的压力。或者升级机器内核,通过调整内核参数进行限流。除此之外,我们还可以从 Flink 引擎本身着手,在 Shuffle 数据阶段直接进行限流。

四、未来展望

14

未来小红书将要探索的方向,主要包含以下三部分。

  • 第一,批模式应用深入挖掘。我们希望能够深入用户,挖掘更多的批模式引擎的使用场景,真正推广 Flink 的流批一体。
  • 第二,配合使用 K8s 的 Resource Quota 功能,将业务方的多个小集群进行合并,减少机器的资源碎片问题。
  • 第三,Serverless 是批模式引擎在云原生环境下部署的一个重要目标,但是强行部署为 serverless 意味着如果 pod 被 Kill 掉中间数据就会被清理,会影响任务的故障恢复,这个时候 remote Shuffle Service 的价值就体现出来了,使用 Remote Shuffle Service 可以有效减少对本地磁盘的部分依赖,提升资源利用率,助力云原生架构。

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

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/45892.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Spring Clould 负载均衡 - Ribbon

视频地址:微服务(SpringCloudRabbitMQDockerRedis搜索分布式) Ribbon-负载均衡原理(P14) 具体实现时通过LoaBalanced注解实现,表示RestTemplate要被Ribbon拦截处理 orderservice调用user时候&#xff0c…

FPGA原理与结构——RAM IP核的使用与测试

目录 一、前言 二、RAM IP核定制 1、RAM IP核 step1 打开vivado工程,点击左侧栏中的IP Catalog step2 在搜索栏搜索RAM,找到Block Memory Generator IP核: 2、IP核定制 step3 Baisc界面定制 step4 端口定制 step5 Other Options st…

八种架构演进

日升时奋斗,日落时自省 目录 1、单机架构 2、应用数据分离架构 3、应用服务集群架构 4、读写分离/主从分离架构 5、冷热分离架构 6、垂直分库架构 7、微服务架构 8、容器编排架构 9、小结 1、单机架构 特征:应用服务和数据库服务器公用一台服务…

【vim 学习系列文章 5 - cscope 过滤掉某些目录】

文章目录 cscope 过滤目录介绍 cscope 过滤目录介绍 第一步创建自己的cscope脚本~/.local/bin/cscope.sh,如下: function my_cscope() {CODE_PATHpwdecho "$CODE_PATH"echo "start cscope...."if [ ! -f "$CODE_PATH/cscope.…

图片速览 FlashAttention+Softmax的安全计算形式(暂记)

FlashAttention最基础的方案来自使用高速的share memory来加速Softmax操作,实现Softmax的tiling方案。(Q,K,V之间的乘法可由gemm实现。) 左侧为GPU各部分的访问速度比较 FlashAttention使用平铺来防止大型实体化𝑁 &#x1d44…

docker搭建es+kibana

docker搭建eskibana 0 安装docker 如果是mac或者windows,可以直接安装Docker Desktop更加便捷。 前提条件: Docker可以运行在Windows、Mac、CentOS、Ubuntu等操作系统上 Docker支持以下的CentOS版本: CentOS 7 (64-bit)CentOS 6.5 (64-bit…

k8s集群监控方案--node-exporter+prometheus+grafana

目录 前置条件 一、下载yaml文件 二、部署yaml各个组件 2.1 node-exporter.yaml 2.2 Prometheus 2.3 grafana 2.4访问测试 三、grafana初始化 3.1加载数据源 3.2导入模板 四、helm方式部署 前置条件 安装好k8s集群(几个节点都可以,本人为了方便实验k8s集…

【自动化测试】接口自动化01

文章目录 一、熟悉若requests库以及底层方法的调用逻辑二、接口自动化以及正则和Jsonpath提取器的应用6. 高频面试题:9. 示例:接口关联13. 文件上传示例14. cookie关联的接口 努力经营当下 直至未来明朗 一、熟悉若requests库以及底层方法的调用逻辑 接…

系统架构师---软件重用、基于架构的软件设计、软件模型

目录 软件重用 构件技术 基于架构的软件设计 ABSD方法与生命周期 抽象功能需求 用例 抽象的质量和业务需求 架构选项 质量场景 约束 基于架构的软件开发模型 架构需求 需求获取 标识构件 需求评审 架构设计 架构文档 架构复审 架构实现 架构演化 前言&…

Linux:iptables SNAT与DNAT

目录 一、SNAT 1.1 SNAT原理与应用 1.2 SNAT转换前提条件 1.3 SNAT工作原理 1.4 SNAT实例 二、DNAT 2.1DNAT原理与应用 2.2 DNAT转换前提条件 2.2实例 一、SNAT 1.1 SNAT原理与应用 SNAT 应用环境:局域网主机共享单个公网IP地址接入Internet (私有IP不能在Internet中正…

性能比较 - Spring Boot 应用程序中的线程池与虚拟线程 (Project Loom)

本文比较了 Spring Boot 应用程序中的不同请求处理方法:ThreadPool、WebFlux、协程和虚拟线程 (Project Loom)。 在本文中,我们将简要描述并粗略比较可在 Spring Boot 应用程序中使用的各种请求处理方法的性能。 高效的请求处理在开发高性能后端…

嵌入式不如CS?信了你就上当了!

在现在的互联网上,有人疯狂劝退新人进入嵌入式行业。他们认为嵌入式的工资低于计算机科学,学习难度较大,前景也不好。 首先是工资方面,你要知道,计算机科学专业也有月薪四五千的人,也有月薪六位数的人。相…

Jmeter 连接 MySQL 数据库脚本

1、创建线程组 2、创建 JDBC Connection Configuration 3、创建 JDBC Request 4、最终创建的目录 5、重点来了 5.1 在百度中下载个 MySQL-connector-Java-8.0.28.jar,放在 jmeter 的 bin 目录下 5.2 在测试计划中,将 jar 包添加到脚本中 5.3 输入参…

面试之ReentrantLock

一,ReentrantLock 1.ReentrantLock是什么? ReentrantLock实现了Lock接口,是一个可重入且独占式的锁,和Synchronized关键字类似,不过ReentrantLock更灵活,更强大,增加了轮询、超时、中断、公平锁…

不同企业如何选择合适的CRM系统?

市场上的CRM系统千差万别,如何选到适合的CRM系统?很多企业凭借感觉盲目选型,结果上线后发现CRM系统功能不符合需求。这就好比买衣服,不试穿就买回家,结果发现尺码不合适,还不能退换。下面说说企业如何进行C…

series的数据对齐功能

Series 是一种类似于 Numpy 中一维数组的对象,它由一组任意类型的数据以及一组与之相关的数据标签(即索引)组成。举个最简单的例子: 上面的代码将打印出如下内容: 左边的是数据的标签,默认从 0 开始依次递增…

Git 目录详解

一、Git目录详解 在使用Git时,有几个目录和文件在Git项目中扮演着重要的角色,下面详细介绍一下这些目录和文件的作用 1、.git目录 .git目录是Git项目的核心,包含了Git的版本库和元数据等重要信息。在该目录中,有一些重要的子目录和…

嵌入式:ARM Day6

作业:完成cortex-A7核UART总线实验 目的:1.输入a,显示b,将输入的字符的ASCII码下一位字符输出 2.原样输出输入的字符串 源码: uart4.h #ifndef __UART4_H__ #define __UART4_H__#include "stm32mp1xx_rcc.h" #incl…

excel条件格式:不同组对应位置对比标记

问题描述 下图中有两组数据,想要对比两个对应位置的数据并标记 条件格式 选中其中一个单元格,条件格式->新建规则 使用公式确定要设置格式的单元格,自定义需求 格式化剩余同样标准的单元格

QT中资源文件resourcefile的使用

QT中资源文件resourcefile的使用 之前添加图标的方法使用资源文件的方法创建资源文件资源文件添加前缀资源文件添加资源使用资源文件中的资源 在Qt中引入资源文件好处在于他能提高应用程序的部署效率并且减少一些错误的发生。 在程序编译过程中, 添加到资源文件中的…