查看原文
其他

知乎基于Celeborn优化Spark Shuffle的实践

李福龙 DataFunTalk
2024-09-10

1 背景

知乎Hadoop集群上每天运行着大量的Spark作业,包括调度平台提交的例行作业、Kyuubi提交的Spark SQL作业,每天Spark作业的Shuffle量达到3PB以上,单个Spark作业Shuffle量最大接近100TB,单个Stage 50TB。同时Hadoop集群上每天有1PB左右的MR作业Shuffle,以及DataNode的磁盘IO。
Spark ESS Shuffle在大作业稳定性上更有优势,在Executor意外退出或者GC严重时,已经完成的Map端的Shuffle数据,可以继续被下游读取,不受影响,所以知乎使用的是ESS(External Shuffle Service)作为Spark的Shuffle服务。
但是ESS也有自己的局限性,ESS Shuffle过程中,每个Reducer Task需要去每个上游Mapper Task的输出文件中读取属于自己的Block,从而产生大量的网络连接以及随机IO,大量的随机IO会导致容易达到磁盘的IOPS瓶颈,作业性能和稳定性都会明显下降 [1][2]。在知乎,经常遇到IO负载高的节点导致个别Spark作业Shuffle Read耗时不稳定甚至超时导致的作业执行耗时不稳定、失败等问题。下图可以清楚的描述ESS Shuffle过程中的磁盘读取及网络连接情况:

Spark ESS Shuffle流程图

为了更好的表述ESS Shuffle存在的问题,借用LinkedIn论文中的一张Shuffle数据统计图,图中很清楚的描述了使用ESS Shuffle时,5000个样本作业不同Shuffle Stage的平均Shuffle Read Block大小分布,以及作业平均Shuffle Read Block大小与作业Task Shuffle Read耗时关联关系。从图中不难看出,存在大量的KB级别的Shuffle Read Block(每个Block至少会一次磁盘IO),而且Shuffle Read Block越小,整体Shuffle Read耗时更长。

作业ShuffleBlock大小均值、Task Shuffle Read耗时均值的分布

ESS生产环境个别Task ShuffleRead 10M数据10min+
针对ESS 的问题,业界提出了Push Based Shuffle方案,核心思路是Mapper Task Shufle数据不写本地磁盘,而是写入一个远程Shuffle服务(RSS),同一个Reducer Task的数据写入到同一个远程节点,远程Shuffle节点对同一个Reducer Task的数据进行合并,当Reducer Task读取数据时只需从一个节点的连续的磁盘空间读取,下图清楚的描述了RSS在Shuffle过程中磁盘读取时与ESS的不同:

Spark RSS Shuffle流程图

2 Spark RSS实现调研

RSS有不少开源实现,我们主要关注了两个国内大厂的开源实现,腾讯开源的Apache Uniffle以及阿里开源的Apache Celeborn。Uniffle在知乎之前的在离线混部项目中有使用,当时的Celeborn版本还不支持MR作业,而我们需要同时把符合条件的MR、Spark作业调度到混部Hadoop集群,该项目就选择了Uniffle,很好地支持了知乎的在离线混部项目,参考 知乎k8s在离线混部-离线篇。
在本次大数据作业迁移RSS项目中,我们跟Celeborn社区进行了一次交流,Celeborn社区给介绍了下Celeborn的特性,发现Celeborn具有一些吸引我们的特性,比如平滑升级、磁盘容错、基于磁盘负载的调度等,同时Celeborn也已经支持了MR作业,于是我们重新对ESS、Uniffle、Celeborn做了一次对比测试。
我们在测试环境中使用TPC-DS 3000sf数据集进行了ESS、Celeborn、Uniffle对比测试,在我们的测试场景Celeborn、Uniffle查询性能相比ESS都有明显优势,Celeborn与Uniffle差距不大,同时Celeborn的内存消耗更低,下图是在测试环境ESS与Celeborn的耗时比值,数值是ESS耗时相对Celeborn耗时的倍数,大于1代表ESS更慢:

ESS与Celeborn查询耗时比值(大于1表示ESS耗时高于Celeborn)

线上环境相对测试环境磁盘负载要高很多,我们线上环境目前确实存在严重的随机Shuffle问题,理论上线上环境RSS相对ESS的优势会更大一些,所以选择RSS势在必行;另外,考虑到我们在测试中Celeborn相对Uniffle的内存消耗优势,以及Celeborn社区介绍的平滑升级、磁盘容错、基于负载的调度的特性,最终我们选择了在线上环境部署Celeborn。

3 Celeborn上线

3.1 Celeborn部署

在Celeborn之前,知乎使用ESS为Spark、MR作业提供Shuffle服务,ESS嵌入在NodeManager服务中,为了更好的ESS性能,缓存更多的Shuffle 分区元数据信息到内存,给NodeManager配置了比较高的内存。
我们上线RSS的目标除了解决Spark作业Shuffle稳定性和性能问题,同时希望不增加额外的机器成本,最终RSS可以完全平替ESS,所有作业迁移到RSS后,下线ESS,降低NodeManager的资源配置,整体不因为部署RSS增加资源开销。所以我们最终确定Celeborn跟Hadoop集群混合部署,每一台Hadoop节点部署一个Celeborn Worker服务。
为了保障Spark作业迁移到RSS过程的稳定性,作业分批进行灰度迁移,前期只在部分机器上部署Celeborn Worker服务,同时为了验证Spark作业迁移到RSS后是否确实有收益,在灰度上线阶段,避免ESS和RSS共用磁盘互相影响,我们对RSS使用的磁盘及ESS使用的磁盘进行了隔离,每台Hadoop机器有12块磁盘,DataNode依然同时使用所有磁盘,但是配置ESS只使用其中的10块磁盘,RSS使用另外2块磁盘,整体部署架构如下:

Celeborn部署磁盘分配图

3.2 Spark作业迁移Celeborn

迁移Spark作业到Celeborn Shuffle,需要解决两个问题:
在灰度迁移过程中,如何对迁移的作业自动添加Celeborn相关Spark参数,如设置CelebornShuffleManger、Celeborn Master地址等,异常时支持快速回滚,避免人工手动修改;
哪些作业更适合优先灰度迁移到Celeborn Shuffle。
针对自动修改Spark作业Celeborn参数问题,知乎之前做过一个Spark作业自动优化资源参数的项目,该项目支持通过规则自动修改Spark作业参数。在该项目中,通过改造Spark Launcher模块的逻辑,在Spark Launcher模块中采集用户Spark作业参数,使用采集的参数请求作业优化服务,作业优化服务基于不同规则计算新的Spark作业参数返回给客户端,客户端在收到返回的新参数后,修改用户原始参数,并使用新参数提交Spark作业。这个项目正好可以用到我们灰度迁移Spark 作业到Celeborn上,通过增加新的参数修改规则,以及黑白名单控制该规则对哪些队列、哪些作业生效,即可做到对我们预期的作业自动增删Celeborn参数。整体流程如下:

自动调整Spark作业Celeborn参数
在选择哪些作业优先灰度到RSS时,我们综合考虑了作业的优先级、作业Shuffle Read Block数、作业Shuffle Read Block平均大小(我们在Spark作业自动调参项目中采集了每个作业Shuffle Read大小、Shuffle Read Block数等Metric信息),为了降低RSS异常时对业务的影响以及尽快减少ESS磁盘IO压力,前期我们优先迁移的是作业优先级低的、Shuffle Read Block数多的、平均Shuffle Read Block Size小的作业,灰度几个批次后,符合条件的作业都已经迁移完成,后期就直接按照队列和作业优先级迁移了。

3.3 Celeborn升级

知乎最初部署Celeborn时,是基于社区Release 0.3.1之后的main分支部署的,截止到我们迁移完所有Spark jar包类型的作业,我们部署的Celeborn已经落后社区最新代码几百个Commit,有些急迫的特性我们通过patch的方式在我们公司用了起来,但我们也想追齐社区最新的代码,包括一些bug的修复以及新特性。中间穿插着patch过几个社区的补丁,继续patch剩余补丁会有大量冲突,我们最终选择了使用社区最新代码直接替换我们在用的代码。
这样要面临一次整体集群的升级,Celeborn 支持滚动升级,无论是升级还是下线,都能够在不影响当前正在运行的作业的情况下进行。这种优雅的升级和下线方式,保证了业务的连续性和稳定性,避免了长尾作业的等待时间,提高了运维效率。
而且在部署层面,运维的感知就是部署了一个程序,然后检测到程序启动即可,很多的状态等等恢复工作 Celeborn 都会自动完成,不需要运维干预,相较于常规部署与升级 Hadoop 要简单很多。
同时这次的升级我们全程都在正常的工作时间进行,搭配自身的 Ansible 运维平台,我们只需极低的运维开发就可以实现数千个 Celeborn Worker 节点的运维。

3.4 Celeborn兼容Spark1.6

前面介绍过,我们最终的目标是期望Celeborn可以平替ESS,所有Spark作业的Shuffle都迁移到Celeborn,但是Celeborn官方不支持Spark1.6(发现所有的RSS开源实现都没有支持Spark1.6版本)。跟Celeborn社区交流后得知Celeborn之所以没有支持Spark1.6,是因为目前使用Spark1.6的公司已经很少,并不是无法支持。
但是知乎还有不少历史作业是使用Spark1.6实现的,每天大概600个作业,当时评估推动业务方升级作业Spark版本到Spark2、Spark3很困难,为了把所有Spark作业都迁移到RSS,我们自己实现了Celeborn Spark1.6的客户端,支持Spark1.6作业。
Spark1客户端相对于Spark2、Spark3版本最主要的不同在于处理Stage重试相关逻辑。Celeborn Spark高版本客户端在发生Shuffle Fetch Failed异常触发Stage重试时,通过Spark的MapOutputTracker接口清理异常Stage上游所有Map的输出,依赖TaskContext中的stage重试次数信息,决定是否生成新的Celeborn内部ShuffleID。Spark1中MapOutputTracker接口、TaskContext字段信息跟高版本不同,针对Spark1的Stage重试,我们在Celeborn侧ShuffleWriter尝试获取ShuffleID时,是否生成新的ShuffleID不依赖Stage重试信息,而是依赖当前Stage在用的ShuffleID是否发生了ShuffleFetch异常,发生过异常,则在后面同一个Shuffle Stage的Write任务申请ShuffleID时,分配新的ShuffleID;Spark侧在开启Celeborn Shuffle的情况下,发生Stage重试时,内部直接清理Shuffle Stage Output信息,保障重试时上游Stage所有Task的重新提交。
目前已经在所有Spark1.x作业上线,运行稳定,并且遇到异常Celeborn Worker节点导致FetchFailed时,可以触发Stage重试并成功。

4 遇到的问题

4.1 Kyuubi Spark SQL上线Celeborn后无法创建线程

知乎使用Kyuubi管理支持Spark SQL Adhoc查询的不同租户的Spark引擎,在两台高配服务器上部署了Kyuubi服务,Kyuubi服务在两台服务器上启动不同租户的多个Spark Yarn Client模式的实例用于支持公司所有用户的Spark SQL Adhoc查询。我们将Kyuubi启动的Spark作业Shuffle也迁移到了Celeborn。
在一次Celeborn服务从500节点扩容到700节点后,每当用户查询高峰时间段,所有Spark Driver日志中都会有
OutOfMemoryError: unable to create new native thread线程创建异常的报错,导致大量用户查询失败,此时机器及Jvm内存都远没有达到上限。
通过跟Celeborn社区的交流,以及统计分析Spark Driver的线程栈,发现Celeborn针对每个Shuffle Stage都会启动新的线程池用于向分配给当前 Stage Shuffle资源的节点创建连接、预留Slot、提交Commit数据,而在我们的Kyuubi Spark Adhoc场景,每个Spark作业中可能同时运行大量SQL,大量的Shuffle Stage,单个Driver进程线程数高峰可能达到接近1w+个,其中Celeborn InitWorker、ReserveSlot、CommitFiles相关的线程累计8k+,当Worker节点是500时,因为需要连接的Worker相对少,扩容到700后,需要创建更多线程连接更多的Worker节点,怀疑线程数达到了上限。
在该问题发生时,我们检查了机器操作系统配置的线程数、进程数上限,分别是400w+、30w+,机器上实际启动的线程数并没有达到这个上限。在知乎,Kyuubi服务是通过Systemd启动的,Spark Driver进程都是Kyuubi的子进程,我们最终发现达到的上限是Systemd的DefaultTasksMax限制的,该配置会限制Systemd启动的服务以及该服务所有子进程可以使用的线程数上限(所有子进程累加),因为问题发生时,机器内存、CPU等资源还很富裕,所以我们增加了Systemd DefaultTasksMax的配置暂时解决了这个问题。
后续Celeborn社区对线程池创建相关逻辑进行了优化,不同Shuffle复用同一个InitWorker、ReserveSlot、CommitFiles线程池,目前该优化也已在知乎上线,大幅优化了创建的线程数。

4.2 Spark作业设置Celeborn参数后不生效

对于迁移到Celeborn的作业,我们关闭了ESS,灰度迁移过程中,我们发现有个别作业执行更不稳定了,排查日志发现作业实际Shuffle的时候,并没有使用Celeborn,而是用了Spark原生Shuffle,Executor多次被驱逐导致作业不稳定。之所以开启了Celeborn Shuffle,最终却没有使用Celeborn Shuffle的原因是我们通过--jars参数配置的Celeborn客户端Jar,同时用户作业设置了参数spark.executor.userClassPathFirst=true。同时设置该参数以及开启Celeborn Shuffle后,完整的Shuffle处理流程是:
Spark Executor启动的时候,检查到spark.executor.userClassPathFirst配置为True,则初始化ChildFirstURLClassLoader并在创建Task执行线程的时候设置线程ContextClassLoader为ChildFirstURLClassLoader。
Executor使用ContextClassLoader中的ChildFirstURLClassLoader反序列化接收到的Spark Task信息,其中就包括Task相关的RDD、ShuffleDependency等,ShuffleDependency中包含了ShuffleHandle实例
ChildFirstURLClassLoader的实现截断了Java ClassLoader的双亲委托模型,优先使用ChildFirstURLClassLoader自身去加载类,加载不到的情况下,才通过父加载器去加载,所以ChildFirstURLClassLoader反序列化Task信息的时候,加载了CelebornShuffleHandle,并反序列化了ShuffleHandle实例。
Spark Executor默认的类加载器还是java默认的AppClassLoader,所以当直接引用CelebornShuffleHandle类时(instanceof检查时),会使用AppClassLoader加载CelebornShuffleHandle类。
Celeborn SparkShuffleManager在创建ShuffleWriter、ShuffleReader时,会检查当前ShuffleHandle实例是否是CelebornShuffleHandle类型,然而因为当前ShuffleHandle实例是通过ChildFirstURLClassLoader加载创建的,而SparkShufflleManager内部的CelebornShuffleHandle类是使用AppClassLoader加载的,两边CelebornShuffleHandle是使用的不同类加载器加载的,检查结果是False,最终使用了Spark原生的Shuffle。
跟业务沟通后,发现业务作业可以去掉spark.executor.userClassPathFirst=true参数配置,最终该问题通过修改作业参数解决。

4.3 Spark作业中包含GlobalLimit算子时使用RSS问题

Spark Adhoc查询Shuffle切换到RSS后,业务反馈有个Spark SQL查询执行多次,总是失败。通过查看作业执行计划以及作业执行日志,发现因SQL语句中包含了Insert Hdfs Directory以及Limit逻辑,生成的作业执行计划中包含了GlobalLimit算子,而GlobalLimit算子会触发Shuffle,并且Shuffle下游并行度是1,同时该SQL作业结果在Limit之前的数据规模很大,TB级别。
在切换到RSS之前,GlobalLimit算子在进行Shuffle的时候,每个上游Task数据都是写本地,下游Task在Shuffle Read到Limit限制的数据后就结束了;切换到RSS之后,上游Task数据是写到RSS节点,并且属于同一个Reduce Task的数据会写到同一个RSS节点,也就是TB级别的数据会写同一个RSS节点,直接导致该RSS节点因数据写入太快,内存使用过高被Exclude,SQL执行失败。
GlobalLimit算子问题在我们这边是普遍存在的,在知乎Spark是Adhoc的底层查询引擎之一,每天会有大量的Spark SQL Adhoc查询,大量的Spark SQL查询复用一批Spark执行环境,节约启动耗时。为了减少复用的Spark环境的Driver内存压力,我们会自动修改用户的SQL,添加查询结果Insert Hdfs Directory的逻辑,查询平台再从Hdfs上读取结果给用户展示,同时在用户没有主动添加Limit的情况下,我们会自动给用户添加Limit限制。
这个问题不能通过修改单个SQL进行解决,我们调研发现Spark中有一个LocalLimit算子,该算子可以限制每个Spark RDD Partition的结果数据行数,于是我们的解决思路,就是给Spark添加一个执行计划优化规则,当执行计划中包含GlobalLimit算子,并且GlobalLimit算子的上游是一个Project算子(即Select)的情况下,自动在GlobalLimit算子前添加一个LocalLimit算子。

增加LocalLimit之前GlobalLimit算子Shuffle数据量

增加LocalLimit之后GlobalLimit算子Shuffle数据量
上面两张图是使用测试查询验证规则上线前后的效果对比情况,该规则上线后,GLobalLimit算子触发的Shuffle中,每个上游Task只需要Shuffle Write Limit限制的条数数据到RSS节点,不会再导致RSS节点Exclude,Limit前TB级数据规模的查询也可以正常执行,这种Case查询性能和稳定性都明显优化。
4.4 节点负载高导致shuffle不稳定
知乎Celeborn是跟Hadoop部署到同一批机器上的,在作业灰度迁移的过程中,偶尔会发生个别机器负载很高,导致Commit超时或者Spark跟Celeborn通信超时,触发Shuffle异常,影响了作业执行稳定性,针对该问题,我们做了几个调整
增加Celeborn Shuffle RPC、Commit超时时间;
增加Worker Flush Buffer Size,从256K调整到1M,优化Commit耗时;
上线Celeborn基于磁盘负载的调度,所有磁盘分5组,每组之间Slot分配比例差距1.1倍,计算磁盘负载时Fetch、Flush耗时系数都是0.5;
打Celeborn社区Stage重试的补丁,当发生Shuffle 异常时,触发Stage重试,而不是App重试。

4.5 Celeborn Worker连接数过高

灰度上线过程中,我们发现每个Worker节点的连接数很高,高峰时有的Worker 连接数达到1w个左右,跟社区沟通后,我们限制了每个作业最大使用的Worker节点数为500个,在后续我们持续扩容Worker节点数、迁移作业过程中,连接数基本稳定,没有持续增长。
celeborn.master.slot.assign.maxWorkers 500

5 收益

目前RSS使用1/6的集群磁盘资源,接入集群1/3+的Shuffle流量,作业迁移到RSS Shuffle后,作业执行性能明显优化,每天所有作业99分位数耗时提升30%以上;在持续有新作业上线的情况下,作业资源占用呈现下降的趋势。

5.1 作业性能优化

迁移RSS过程中 Spark作业耗时均值变化

迁移RSS过程中 Spark作业耗时99分位数变化

迁移RSS过程中 Spark作业Shuffle Read耗时变化

从作业整体分位数耗时、Shuffle耗时曲线看优化明显,大作业提速明显,所有作业99分位数耗时优化30%以上。
从用户反馈看,已经接入RSS Shuffle的作业,在性能和稳定性上都有了明显优化,业务方主动反馈作业变快了,近期我们也已经基本没有再收到用户反馈的shuffle耗时不稳定、shuffle连接异常等问题。
因为1/3以上的Shuffle流量切到了RSS,但是给RSS的磁盘只有整个集群的1/6,所以对于没有接入RSS的作业,也有明显的优化,公司核心例行作业结束时间提前了一个小时以上。

5.2 资源使用优化

迁移RSS过程中 Spark作业每日作业数变化

迁移RSS过程中 Spark作业每日内存资源占用变化

迁移RSS过程中 Spark作业每日CPU资源占用变化
从作业资源占用的角度看,也是有一些优化的,从截图中,可以看出,在RSS上线迁移过程中,公司一直有新作业上线,及时新增作业的情况下,所有作业CPU、内存占用有明显的下降趋势。

6 总结与展望

目前所有Spark Jar包、Spark SQL Adhoc作业均已接入RSS,并且取得了不错的效果,后续我们想在RSS方向继续做的事情包括:
Spark SQL ETL作业接入Celeborn
MR作业接入Celeborn
下线ESS服务,降低NodeManager内存占用
Celeborn更好的兼容高负载节点

7 参考

Spark+Celeborn:更快,更稳,更弹性
Magnet: Push-based Shuffle Service for Large-scale Data Processing


往期推荐


阿里云 PAI 大语言模型微调训练实践

大模型推荐系统的打怪升级之路!

从电商场景,看抖音集团数据治理实践

当大数据分析遇上方程式赛车:捕捉极速赛场上的时间印记

AI Agent 在 1688 电商平台中的应用

小红书图数据库在分布式并行查询上的探索

领域大模型的挑战与机遇:从构建到应用

电商知识图谱建设及大模型应用探索

大模型在推荐系统中的应用及挑战

指标体系技术成熟度曲线概述

点个在看你最好看

SPRING HAS ARRIVED

继续滑动看下一个
DataFunTalk
向上滑动看下一个

您可能也对以下帖子感兴趣

文章有问题?点此查看未经处理的缓存