PB级Spark Shuffle优化实践：基于Celeborn的定位、部署与调优

1. 为什么在PB级任务里，Spark的Shuffle成了头号瓶颈

我们做数据平台的人，最怕看到两类现象：一类是任务明明都跑完了，可最后那几个Reduce任务拖在后面，整个调度链一直挂着；另一类是磁盘看着看着就满了，第二天准报容量告警。最近我们所在的某头部终端厂商数据团队，日处理数据量稳定在PB级之后，这两件事同时爆发了。以前那套优化手段，优化Spark调度参数、拆任务、加资源、改数据倾斜，能用的都用了，性价比越来越低。最终我们把注意力放回了一个看似基础、实际上坑最深的环节：Shuffle。

Shuffle在Spark任务里最常见，也最容易被低估。它的本质就是一次数据重排：Map阶段把数据按照Key分成一份份，Reduce阶段再跨节点拉取、合并。数据少时，这无非就是多几次磁盘IO；数据到了PB级，Shuffle中间文件的生成、持久化、传输、清理都会被指数级放大，会放大到让人吃惊的程度。这是我们决定研究Celeborn的起点。下面要写的内容，全是我们围绕“PB级Shuffle优化处理”这个场景折腾出来的实践：选型比对、部署架构、参数配置、压测数据，还有生产上踩过的坑，尽量把当时的判断依据和最终结果都讲清楚，给后面的人少走点弯路。

1.1 数据规模一起来，先顶不住的不是计算

在很多业务团队看来，PB级数据意味着要加机器、加并发。但实际上到了PB级，CPU和内存相对好办，真正难的是那些需要跨节点交换数据的逻辑，比如Join、GroupBy、Distinct、窗口函数。这些操作底层都会触发Shuffle，一个任务只要Stages一多，Shuffle就变成整条DAG里最容易出问题的链路。

Shuffle有个特点，就是看起来简单：只要能完成分发的任务都能用。缺点同样明显：数据落盘次数多、小文件数量大、磁盘占用高、Reduce拉数据慢、失败以后重算成本高。这些缺点在PB级场景会被数字放大。举一个我们日常的真实例子：一张日活表Join一张计费表，其中一张表一天产生约5PB的中间数据，经过Shuffle放大，按2.5倍估算就有十几PB的临时数据，反复写读加清理，等于把磁盘和网络的吞吐逼到极限。在这种量级下面，优化某一个算子的执行计划效果很有限，上游Shuffle一慢，下游无论是什么引擎都得等。

还有一点容易被忽视：Shuffle不只是“计算环节”，它其实是一个非常典型的分布式存储问题。Map端写出来的数据要被Reduce端读取，那它就得具备存储系统的三个能力：持久化、可寻址、可恢复。原生Spark把这三件事都放在Executor的本地磁盘上，等于每一个计算节点同时扮演存储节点。数据量小的时候没事，数据量一大，计算和存储互相抢资源，必然先出问题。

1.2 我们现场看到的三个典型“症状”

第一个症状是元数据服务吃紧。当时我们的HDFS集群里，一个小任务平均产生几十万个Shuffle临时文件，全集群几千个任务一起跑，小文件数量直接冲到千万级甚至亿级，NameNode的RPC处理能力成了瓶颈。这不是单个任务的问题，是整个集群层面的系统性风险。

第二个症状是长尾Task。几百个Reduce任务里，总有那么几个要跑数倍时间。现场分析起来原因极多，可能是网络拥塞，可能是Block拉取排队，也可能不同盘性能差异大。总之，Reducer比Map更敏感，少部分机器的异常会被长尾无限放大，最终拖垮整个DAG的完成时间。

第三个症状是失败重试的成本被放大。一个Executor挂掉，Spark默认需要把对应Shuffle的Map端数据全部重算，而重算会产生新的落盘、新的网络传输。集群一旦出现波动，这种雪球效应极其明显。我们统计过，高峰期集群里相当比例的计算资源其实是在做“重算已经算过的东西”。

这三个症状叠加在一起，才是我们决定系统性解决Shuffle问题的直接原因。以前那种“哪个SQL慢就救哪个”的思路，在PB级场景下已经失效了，必须换架构。

1.3 选型思考：原生Shuffle、External Shuffle Service与Celeborn的区别

在最终方案讨论时，我们对三种方案做了对比。

原生Spark Shuffle：Executor自己管理临时文件。优势是没有额外组件、部署简单；劣势是Partition过多时小文件问题无法避免，而且Executor挂了要重算。这个方案我们跑了很久，已经很熟悉它的上限在哪里。

External Shuffle Service（ESS）：典型的部署是Spark on YARN中随NodeManager一起启动的常驻服务，解决的是Executor被删除后Shuffle数据可以保留的问题。它确实让Shuffle数据不再跟随Executor生命周期一起消失，但并没有从根上解决文件数量、写放大和长尾问题。相当于是给原来的方案打了个补丁，而不是换一套机制。

Celeborn：本质上是一个远程Shuffle Service集群。Map端把Shuffle数据推给Celeborn Worker，由Worker在内存中接收、合并后刷盘，最后生成数量可控的大文件；Reduce端再从Worker拉取结果。它把原来插在计算节点上的临时文件剥离出来，形成一个独立的存储服务。中间文件的生命周期、数量、恢复策略全部换了一套。

选型时我们的判断很直接：如果任务量在TB级，没必要折腾；但到了PB级，Shuffle的“文件数量”这个核心矛盾，必须有一个东西从架构层面来解决。Celeborn不是单纯换个实现，而是把临时数据和计算彻底分离，这直接决定了我们后续部署它的方式。

2. 部署形态与资源规划：把Celeborn当成一个分布式存储系统来设计

2.1 Celeborn组件到底在干什么

先讲一下组成部分，不然后面说不清。Celeborn有Master和Worker两种角色。Master负责集群元数据、Worker负载分配和容错协调，可以用Raft协议做选主，保证高可用。Worker负责实际的数据接收和存储：Executor的Map端会把Shuffle数据推送到Worker，Worker在内存里对多个推送做合并，再按照配置刷盘成文件。

为什么要说它是存储系统而不是缓存系统？因为它保证数据一旦被Worker接收，就有持久化能力，Map端的任务可以认为自己写完了就能结束，不用等下游Reduce确认。这个特征非常关键，意味着Shuffle数据不再依赖计算节点存活，计算节点随便重启都不影响已经完成的Shuffle结果。

Shuffle文件的读取也很有特点。Reduce端不是按Partition直接拉取文件，而是按照Chunk的粒度去读，数据的加载做了连续化处理。实际效果是随机IO很大程度上变成了顺序IO，对机械盘和SSD混合环境都比较友好，这一点后面压测时感受特别明显。

还有一个细节是LifecycleManager。它运行在Spark Driver内部，负责协调整个Shuffle的注册、Worker分配和清理。这个设计的作用是让“一个Application的Shuffle中间数据”有一个明确的管理者，整体生命周期都由它决定。简单说，Driver负责告诉Executor往哪写、从哪读，Worker负责真正把数据落盘。

2.2 我们采用的部署拓扑

我先把拓扑说清楚。核心思路是“存储节点与计算节点分离”。

计算资源是现有的Spark集群，几千个节点，不需要改动调度模型。Celeborn集群单独部署，初期采用了一批旧存储服务器改造，每台机器配备多块NVMe SSD，单机容量约20TB左右。这样改的一个好处是磁盘成本能压下来：旧存储节点计算能力弱，但磁盘密度高，非常适合做这种吞吐型存储服务。

Master至少三节点，组成一个小集群，用Raft保证高可用。为什么一定要三节点？因为Master并不参与数据传输，压力没那么大，单节点也能跑，但生产环境我们不敢赌单点。还有一点要特别提醒：Celeborn的Worker不直接暴露数据目录给客户端，而是走RPC通道，所以网络配置必须允许Spark Executor和Worker之间的端口互通，不然接入后第一个现象就是Push失败。

正式部署时，我们给每个Worker的JVM堆给了大约16GB，再加上较充足的堆外内存。为什么要强调堆外？因为Celeborn的数据接收大部分发生在堆外，Netty使用的DirectMemory也占堆外。堆给得太大，数据接收并不会因此变快；堆外如果不足，反而会频繁出现OOM。这个比例的拿捏，我们后面调整过好几轮才稳定下来。

2.3 内存、磁盘与网络的预算分配

我把我们一版比较稳定的Worker资源比例列出来作参考。读者可以结合自己机器的情况设计，不必照抄。

磁盘不组RAID，这也是一个值得展开说的点。Celeborn对单盘故障有恢复策略，不依赖RAID保证数据安全。我们的做法是在单个Worker上挂12块SSD，每块盘是一个独立存储目录，让数据分布到多盘。一块盘坏了，只要整体余量充分，影响能控制在很小的范围，恢复机制会自动补齐副本。

内存分配最大的教训，是不要为了“缓存数据”去买内存。远程Shuffle服务的本质是让数据快速过境，尽快刷盘，不是真在内存里长期保留。内存给得再多，如果刷盘速度跟不上，最终瓶颈依然是磁盘。我们第一版配置就给了大量内存，磁盘却用机械盘，线上高峰一来，刷盘排队严重，延迟直接拉高。后来把磁盘换成NVMe SSD，把内存调回合理水位，才真正稳定下来。

3. 接入Spark与关键参数调优

3.1 Spark侧的一次性配置

接入方法比较成熟，核心是把Spark的ShuffleManager换成Celeborn的。下面是配置文件最核心的几行，可以直接照着试。

spark.shuffle.manager org.apache.spark.shuffle.celeborn.RssShuffleManager spark.celeborn.master.endpoints master1:9097,master2:9097,master3:9097 spark.serializer org.apache.spark.serializer.KryoSerializer spark.celeborn.push.replicate.enabled true

RssShuffleManager会把Map端输出路径指到Celeborn集群，不再写本地临时目录。这里注意：一旦使用RssShuffleManager，传统Spark参数中与本地磁盘临时文件相关的配置就会失效一部分，不要还想着“同时保留本地和远程写两份”，没有意义，还会白白浪费磁盘。

关于复制参数，我们的生产取向是开启。开启后每个Shuffle分片至少复制到两个Worker上。如果某个Worker挂了，数据还能从另一份副本读取。对于PB级任务来说，多一份存储，换的是重算概率大幅下降，这个账很划算。计算节点的Executor是廉价的，随时可能被调度器杀掉，而Worker上的数据是经过一次完整写入的，保护住它就能避免整条DAG回滚。

3.2 Celeborn服务端关键参数

服务端配置里，几个参数需要重点关注。

celeborn.worker.flush.buffer.size 128m-256m celeborn.worker.flush.queue.capacity 16 celeborn.worker.disk.buffer.size 32k-64k celeborn.worker.partition.sort.enabled true

先说flush buffer size。Map端推送过来的数据会先进内存队列，达到一定量后整块刷盘。这个值设得过小，刷盘频繁，盘IO会被打满；设得过大，单块内存占用高，突发流量下容易把Worker内存打满。建议根据单Worker的并发推送流数量来算，比如你一个应用在高峰时可能有50到80个Executor同时推数据，乘以预留的并发上限，就能得出一个相对安全的buffer值。

再说partition.sort.enabled。开启后，Worker在刷盘前会按Partition维度对数据做一次排序和合并，生成的文件是连续的大块。这样做会让读取更高效，代价是CPU小幅增加。在PB级场景下，这个开关基本都要打开。如果不开，数据在盘上乱散，Reduce端拉取时随机IO又会回来不少，得不偿失。

3.3 参数之间的联动关系与调优经验

有几个参数我们在调的时候特别花心思，拿出来供参考。

第一，Push流程的并发与带宽的关系。彻底走Celeborn之后，Map端不再写本地盘，而是直接写到网络上，这会让集群的北向流量涨得很猛。不能只盯着Spark任务本身，还要关注应用的任务并发度。比如spark.sql.shuffle.partitions只有几百，但Executor数量高达两千，可能会导致每个Executor往同一个Worker并发推送多个Partition的数据，反而让Worker端排队变长。我们的做法是按照Executor数量和Partition数量，重新调整客户端Push并发参数，把并发控制在一个合理范围内。

第二，Reducer拉取的并发。读取的并发可以适当调高一些，因为Celeborn写盘是有序连续块，读取不再像原来本地小文件那样频繁随机。我们把读取并发从默认的3调到5，长尾任务明显变少。不过不建议无限增大，增大到一定值后网络队列延迟会上升，反而影响整体。

第三，HDFS的参数不能全盘照搬。有些配置在网上流传很广，但并不是所有场景都合适。比如三副本策略，Celeborn这里完全不需要，因为中间数据不是业务持久数据，三副本的成本太高。我们生产里做一份冗余副本，加上恢复机制，就已经足够。

3.4 倾斜场景与超大Partition的处理

前面讲的是常规调优，这里补充一个实际遇到的痛点：数据倾斜和超大Partition。

Shuffle天然就有倾斜问题，PB级任务里某些关键Key上的数据量特别大。原生Spark情况下，倾斜会导致某个Reducer压力奇高；用Celeborn之后，倾斜会表现为某一个Worker的某个分区文件特别大，读取时可能变慢。我们试过几个手段。

首先，确保Worker的Partition排序开启，同时给Spark侧聚合留够内存。如果内存不足，倾斜Key会触发大量Spill，反而增加Worker端压力。

其次，对于极端倾斜的Key，先做加盐处理。比如对Join Key拼接一个随机前缀，把大Key拆成多个小Key，等Reduce完成后再去掉前缀。这个方案其实和Celeborn无关，可以在上游Spark作业里直接做，但在Celeborn架构下，倾斜带来的表现会集中到某个Worker的特定分区，所以这个改造反而比原来更重要。

另外，超大Partition可以启用独立的读取通道配置，减少与其他小分区的争用。这个需要结合客户端版本能力来看，版本不同支持程度不一样，实施前先查一下当前版本的特性。

最后强调一句：无论配置多少，必须用真实业务场景做压测，不能只在小数据量环境里验证。小数据量下所有方案都好看，只有压到PB级，才能看出参数是不是真的合理。

4. 压测设计与全链路验证

4.1 压测用例怎么设计才代表PB级生产

完成基本调研和参数配置后，就进入压测阶段。压测不是找一个公开数据集跑一遍就行，必须贴近PB级生产。我们选了三类典型任务：

• 全量宽表构建，涉及多张大表的Join加GroupBy，Shuffle读数据量达到TB级别；
• 长周期窗口计算，涉及全局排序和窗口分区，Shuffle数据量大，对稳定性要求高；
• 高并发混跑场景，在集群里同时提交多个中型任务，模拟生产时的互相争抢。

测的时候还要把任务设置在接近生产数据规模的级别。我们会先用平时最大任务的十分之一数据量跑，确认功能和配置都没问题后，直接全量上线白名单任务。压测期间的风险点集中在网络和磁盘，所以要同时采集网络流量、磁盘吞吐、CPU、GC和Shuffle相关指标。

压测前记得把数据量、资源规格、基准耗时全部记录下来，不然后面对比无从谈起。我们当时专门整理了一张基准表，每个压测用例都带着原始数据规模和执行参数，跑完一轮就更新一次，效果一目了然。

4.2 实测数据与效果

把几个核心指标列出来。下面数值都是我们环境中实际跑出来的，不同集群设备性能不同，仅供参考。

间接影响同样重要。HDFS的NameNode压力明显下降，小文件产生的目录操作不再成为瓶颈。之前因为HDFS小文件导致的告警，在压测期间一次都没有出现。

还有一个数据我们当时很关注：Map端写路径的耗时。原生Shuffle下，Map端写完数据要等文件落盘，涉及文件关闭、元数据更新等操作，时间不稳定；走Celeborn后，Push完成就意味着数据已经被远端接收，Map端任务结束得更干脆，整个Stage之间的衔接流畅了很多。

4.3 为什么网络和磁盘没有成为新瓶颈

看到上面的结果，很多人第一反应是：这不就是把磁盘问题换成了网络问题吗，网络怎么会不炸？这里必须把原理讲清楚。

原生Spark Shuffle，Map端写本地盘是一次全量IO，Reduce端远程拉取又是一次全量IO；换成Celeborn，Map端走网络推到Worker，Worker刷盘一次，Reduce再从Worker拉过来。看起来网络流量多了，但实际上这条网络链路是连续的大块传输，不是原来那种成百万个小文件碎片在网络上乱飞的模式。同样的数据量，连续IO和碎IO的吞吐差距是数量级的。

我们测下来，PB级任务下Celeborn的南向流量确实很大，但网络利用率很高。同时，交换机层面提前做了容量和带宽规划，没有因为网络拥塞恶化。磁盘方面，NVMe SSD是关键。如果继续用机械盘，刷盘排队会抵消内存合并带来的优势；换成NVMe SSD后，刷盘速度很快，多线程Push的堆积窗口都很短。这是后面长期稳定运行的重要基础。

5. 生产环境稳定性问题与故障排查实录

5.1 Worker磁盘写满导致的任务卡死

上线第一个月，我们遇到一个典型问题：某个Worker磁盘被写满后，该Worker上所有Shuffle数据推送集中失败，直接导致一批应用重启。

排查下来，当时只做了单盘余量监控，没有设置“剩余空间比例低于阈值时Worker主动停止注册并迁移数据”的保护机制。后来在Worker上配置了磁盘水位线，低于阈值时该Worker不再接收新的Shuffle任务，并让Master把新任务分配给其他Worker。

这里有个细节：磁盘写满不等于任务立刻失败。部分应用在Push失败后会重试，重试会再次把数据发到同一个Worker，导致盘上出现大量垃圾数据。恢复的办法是尽快清理失效分片，并把涉及的相关Executor重启。

这个经验总结起来就一句话：磁盘容量监控和水位线配置必须提前做，别等爆炸再救火。远程Shuffle服务的存储地域就是它的磁盘，存储地域的健康度决定了整个服务的地基。

5.2 Master切换与Push超时

还有一次是在Master发版升级时，因为Raft选主切换，出现了短暂几秒的Push超时。

Master在Celeborn链路里不承担数据传输，所以多数Push请求不经过Master。但客户端要注册、要拿Worker分配方案，这些都属于控制面操作，切换时有一小段连接抖动。我们的应对是两件事：一是把客户端重试次数调高，保证偶发的控制面超时不影响业务；二是把故障切换时的backoff设置得短一些，别让任务一遇到切换就宣告失败。

如果在生产环境里看到大量“Push timeout”但随后又恢复，多数是控制面瞬断，看看是不是处于Master切换或网络波动窗口，而不是盲目重启集群。盲目重启会让所有正在推送的任务一起断线，反而扩大影响面。

5.3 快速排查对照表

另外入门时有个常见坑：Celeborn的客户端Jar必须加到Spark的classpath里，否则Spark会回退到默认ShuffleManager，你以为在用Celeborn，实际上还在用原生Shuffle，压测数据自然好看不了。判断方法也简单，看Map端临时目录是否还有大量文件产生，或者看Executor日志里的ShuffleManager实现类名。

6. 什么时候该上Celeborn，什么时候不要用

我们磨合了差不多一个季度，最后想给一个客观的结论：远程Shuffle服务不是银弹。

如果你正在处理的任务量还在TB级别，或者只在少数几个SQL上遇到了Shuffle性能问题，我建议先用常规手段：调整分区数、优化Join顺序、开启AQE，效果更直接，成本也最低。没必要为一个可以绕过去的问题引入一套分布式存储体系。

如果你的集群确实已经到PB级，并且你面对的是上千并发任务、巨大Shuffle量、频繁小文件、重算风险高，那Celeborn值得认真调研。现在它已经是开源项目，自己跑Demo验证完全没问题，不需要额外采购。但部署并不轻量，需要额外推一套服务，运维也要多一份看护负担。它不是装上去就万事大吉的东西，需要长期的容量规划、版本升级、故障演练。

我个人的体会是，一旦把Celeborn当成一个真正的存储系统而不是一个插件去看待，运维心态就会不一样。你会开始思考它的容量、水位、冗余、升级节奏、监控告警，这和平时调Spark参数完全不是一个层面的工作。但它带来的收益也是结构性的，不是优化某个SQL，而是整个Shuffle层换了一套更适应PB级规模的地基。

如果让我给下一个要接手这套系统的朋友一句话，我会说：先花时间把磁盘预算和监控做好，再谈性能调优。地基稳了，PB级也就那样。