Apache Celeborn落地实践：解决PB级Spark Shuffle瓶颈

做大数据平台的男人，最怕半夜被手机震醒。那天凌晨两点我醒来看群，某张离线宽表任务报错刷屏，点进去一看，整个 Spark Stage 挂在 shuffle 阶段，磁盘 IO 打满，几个节点的数据目录直接被写爆。这不是第一次了。在日处理 PB 级数据的规模下，shuffle 已经把集群稳定性反复按在地上摩擦。

今天想聊的，是我们引入 Apache Celeborn 做 Remote Shuffle Service 的完整实践。从为什么一定要动 shuffle，到 Celeborn 的核心机制、部署接入、参数调优，再到上线过程中踩过的坑，一次说清楚。如果你也在维护大规模 Spark 集群，或者被 shuffle write 失败、磁盘倾斜、Executor 被 Kill 折腾到头秃，这篇文章应该能给你一套可落地的解法。

需要先说清楚适用人群：大数据平台研发、数仓工程师、以及所有要把 Spark/Flink 跑稳的运维同学。下面内容不需要你多资深，但最好对 Spark 的 shuffle 机制有基本认识。

1. 为什么 PB 级场景下 Shuffle 成了最大瓶颈

1.1 Shuffle 到底慢在哪里

Shuffle 的本质是数据重分布。Map 端把 key 按照分区规则拆开，写到本地磁盘；Reduce 端再从所有 Map 任务所在节点拉数据。这个过程看起来简单，但在 PB 级规模下它有三大痛点。

首先是写放大。每个 Map Task 都要把自己处理完的中间结果落盘，如果 Map Task 有上万个，数据总量摆在那里，本地磁盘的写压力是巨大的。更麻烦的是，一个 Executor 上有多个并发 Task 同时写，磁盘 IO 瞬间打满，出现"大作业拖垮节点"的情况非常常见。

其次是 fetch 重试成本高昂。Reduce Task 去拉数据的时候，如果对方节点正在 Full GC，或磁盘出问题，fetch 失败后 Spark 会重新调度重试。重试逻辑本身不复杂，但在节点故障频繁的大集群里，fetch 重试带来的网络风暴和调度压力会被成倍放大。

最后是数据倾斜被本地磁盘放大。一个热门 key 对应的大分区，可能被分配到某一个 Executor 上，本地磁盘直接变成热点。做过 PB 级 join 的同学都有经验：一百个节点好好的，偏偏有一个节点磁盘满了，整个 job 被拖死。

1.2 传统优化手段的边际效应越来越低

刚开始我们也是常规操作。调大 spark.shuffle.file.buffer，换 ZStandard 压缩，增大 spark.shuffle.compress 相关参数，调 spark.sql.adaptive.shuffle.targetPostShuffleInputSize 让 AQE 合并小分区。这些手段在小规模集群有效果，但到 PB 级就力不从心了。

先说压缩。Shuffle 数据落盘和网络传输都有压缩，确实能降低 IO，但 CPU 开销同步上来了。对 CPU 密集型 SQL 作业来说，压缩省下的 IO 时间往往还没压缩消耗的 CPU 时间多。再说 AQE，它能把 reducer 数量从几万个合并到几千个，这是个巨大进步，但 map 端写入本地磁盘然后再拉取的本质没有变化，只是把问题从"太多小文件"变成了"中等数量的大文件"。

更关键的是，早期我们依赖 YARN 的 NodeManager 上自带的 External Shuffle Service（ESS）。ESS 解决了 Executor 退出后 Reduce 端还能取数的问题，但它的数据仍然是落在计算节点本地，仍然是同一个磁盘。物理机磁盘容量和 IO 能力就是天花板，数据量大了照样装不下。而且 ESS 无法横向扩展，也不适合 K8s 等动态调度场景，这基本就是走到了死胡同。

1.3 存算分离是这条路必须走的方向

回归根本，我们把 shuffle 看作一个"中间存储系统"。它的生命周期很短，但访问模式很固定：map 端写一次，reduce 端读一次。那为什么不把它从计算节点里抽出来，变成一个独立的集群服务？这就是 Remote Shuffle Service（RSS）的核心思想。

RSS 不是新概念，市面上也有几个开源实现。我们当时做选型，核心看四点：第一，是否真的做到 shuffle data 的存算分离；第二，是否具备双副本、多副本能力，避免 Worker 单点导致重新计算；第三，是否对 Spark 版本足够兼容；第四，社区活跃度和生产案例。最终选择了 Apache Celeborn。

选择 Celeborn 有几个现实原因。它对 Spark 2.4 / 3.x 都有官方支持，可以做到零业务代码改造；它天然支持双副本写入，这对于动辄跑几个小时的离线作业特别重要；它提供了一定的内存与磁盘分层，不像一些简化实现那样没数据就失控。另外，Celeborn 在社区里有不少一线团队的生产实践反馈，我们踩坑时能搜到真实案例，这在开源项目选型里是很大的加分项。

2. Celeborn 核心机制与原理

2.1 三个角色搞定一切

Celeborn 的架构很清晰，由三个角色组成：Master、Worker、Client。Master 负责集群管理，维护 shuffle 的元数据信息；Worker 负责实际存储 shuffle 数据；Client 以 Spark 插件的方式集成在 Executor 里，负责把 map 端的数据推给 Worker，以及让 reduce 端从 Worker 拉数据。

Master 可以多节点部署，通过 Raft 协议选主，解决单点问题。Worker 是真正干活的人，可以看成是专门为 shuffle 设计的"高并发存储节点"。每个 Worker 配置若干存储目录，把 shuffle 中间数据写成文件，并提供网络服务让客户端读取。

打个比方，传统 shuffle 就像每个家庭都自己囤菜做饭，厨房面积小、锅碗瓢盆不够用；Celeborn 就像一个中央厨房，所有家庭做好半成品送过去，要取菜的时候再去中央厨房拿。中央厨房的厨具和场地是按规模设计的，能扛住更大的吞吐。

2.2 一次 Shuffle 的完整旅行

我们来看看一次 shuffle 在 Celeborn 里是怎么流动的。Map 阶段，Executor 里的 CelebornShuffleManager 会向 Master 注册一个 shuffle，Master 为它分配 Worker 资源。每个 Map Task 输出时，会把不同 reducer 分区的数据分别推到对应的 Worker 上。

Worker 收到数据后，不是直接落盘，而是先在堆外内存里做缓冲聚合。缓冲是有最大限制的，超过阈值后把内存中的数据按分区刷成一块文件。注意这里刷盘不是每来一条就写一条，而是积攒到一定量才批量写，这样磁盘 IO 效率会高很多。

等所有 Map Task 完成，Worker 上就有了这个 shuffle 的完整 chunk 数据。Reduce 端开始拉数据时，只需要根据分区编号去对应的 Worker 读取，拿到文件偏移和长度，直接拉取相应数据块即可。整个过程中，计算节点本地不保留任何 shuffle 数据，Executor 挂掉也不影响后续任务。

Celeborn 还做了一件事：双副本写入。默认情况下，每个 shuffle partition 的数据会同时推到两个不同的 Worker 上，一个作主副本，一个作备副本。正常读取时用主副本，如果 Worker 故障或磁盘异常，客户端可以切到备副本继续读。对于 PB 级作业来说，这一点真的很重要，它避免了"重新计算整个 stage"这种最坏情况。

2.3 与 Spark 的集成机制

从 Spark 的角度看，Celeborn 做了一件极其聪明的封装。它实现了 Spark 的 ShuffleManager 接口，你只要把 spark.shuffle.manager 指到 CelebornShuffleManager，剩下的逻辑全部由客户端包接住。业务代码不需要改，Spark SQL 也不需要改，运维只需要把依赖放进 classpath。

更重要的是，Celeborn 的设计和 Spark AQE 的动态分区合并是兼容的。原本 AQE 在做 coalesce 时，reducer 数量发生变化，可能导致某些本地 shuffle 文件失效需要重新计算；但在 RSS 模式下，所有数据已经推到了 Worker 上，reducer 合并后仍然可以从之前的文件里读取，省掉了重复计算。这在 PB 级场景下收益非常明显。

另外，Celeborn 对动态资源申请也有天然优势。Executor 申请之后很快退出，shuffle 数据不在本地，新起来的 Executor 依然能从 Worker 拉数据。在 K8s 或大规模弹性集群上，这个特性几乎是刚需。

3. 落地部署与集群接入流程

3.1 集群规划与容量估算

部署 Celeborn 前，最该想清楚的是规模。很多团队第一步就栽在这里：Worker 节点太少，shuffle 高峰一来直接被压垮；或者机器过多，资源利用率上不去。科学的方法是按峰值 shuffle 数据量来算。

一个可用的预估公式是：Worker 磁盘总量 >= 高峰期同时运行的 shuffle 数据量 × 副本数 ×（1 + 其他开销比例）。这里的高峰期同时运行的 shuffle 数据量，可以通过 YARN 或 Spark UI 的历史指标统计，取最近一个月的 p99 值比较稳妥。我们当时按最小化配置，先部署了 12 个 Worker，单节点配置 8 块 4T SATA 盘，每节点 64GB 堆外内存。原因是 shuffle 数据主要是顺序写，没必要上昂贵的 NVMe，SATA 配合批量刷盘足够支撑业务。

容量之外，网络是很容易被忽视的一环。shuffle 数据从 Executor 推到 Worker 是跨网络传输，节点多且并发大时，万兆网卡几乎是必须的。如果跑在云上，还要注意虚拟化环境对带宽的限速，很多看起来是"io 问题"的故障，实际是网络先到天花板了。

3.2 Master 和 Worker 部署细节

Master 至少部署三台，通过 Raft 保证选主。配置相对简单，关键是保证节点间网络连通，以及机器时间尽量同步。Worker 部署则在每个节点上配置好存储目录，生产环境建议多目录，并且避免目录存在系统盘。

启动顺序没什么玄学：先起 Master，确认 Master 页面能打开，再起 Worker。每个 Worker 启动后会主动向 Master 注册，所以 Master 页面里能看到所有 Worker 的在线状态、磁盘与内存信息。这地方有一个容易踩的坑：Worker 的 hostname 必须在 Master 和 Client 之间都能解析，很多 shuffer 连接超时问题都源于 /etc/hosts 没配全。

启动后记得看一下 Master 的 Web UI，所有 Worker 的可用内存、活跃 shuffle 数一目了然。上线初期我建议每天盯几次 UI 上的 shuffle 文件总量，方便评估容量预估是否准确。

3.3 Spark 作业接入步骤

接入分三步。第一步，把对应版本的 celeborn-client-spark jar 放到 Spark 的 jars 目录，或者通过 spark.jars 动态配置。第二步，在 spark-defaults.conf 里增加以下核心配置：

bash复制spark.shuffle.manager=org.apache.spark.shuffle.celeborn.CelebornShuffleManager
spark.celeborn.master.endpoints=master-01:9097,master-02:9097,master-03:9097
spark.celeborn.client.push.buffer.max.size=256k
spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.celeborn.client.registration.shuffle.enabled=false

第三步，提交一个典型作业做小流量验证。不急着切全量，先挑两三个跑得慢但又不会影响核心链路的作业，观察 Celeborn 页面上的 push/fetch 数据量和耗时。如果一切正常，再逐步把新作业接入。

这里必须强调版本一致性。Clent jar 的版本和 Celeborn Server 的版本要严格对应，否则可能出现协议不兼容或莫名其妙的反序列化错误。我们早期因为随手拿了一个 master 分支的 jar 包，结果和线上 0.3.2 的 server 对不上，排查了很久才定位到。

4. 调优实践：从能跑到跑得好

4.1 关键参数与推荐值

我们上线初期用的是 Celeborn 默认参数，能跑通，但离好用还有距离。经过几轮压测和生产调优，锁定了一组比较可靠的参数，这里分享出来仅供参考，具体还需要结合业务调整。

参数调整要遵循先算后调的原则。比如 push.buffer.max.size，它和 Executor 端并发 push 线程数强相关。如果你一个 Executor 上并发度是 4，每个 buffer 512k，单个 Executor 峰值就有 2MB 左右缓冲，这个量在绝大多数场景是安全的。如果并发度是 16，那 512k 的 buffer 加在一起就有 8MB，在 2G 堆内存的 Executor 里就可能吃紧。

4.2 性能表现与收益

经过两三轮调优后，我们对线上跑批作业做了一个对比。下面表格来自我们生产集群中几类典型作业的实测数据，注意这是特定场景下的结果，不代表所有作业都能达到同样水平，但趋势很能说明问题。

为什么总耗时能下降？关键原因是 IO 和 CPU 的平衡发生变化。传统 shuffle 的 map 端本地写和 reduce 端的拉取，都在争抢同一批物理机的磁盘和网络；Celeborn 把这部分流量转移到了专门的 Worker 集群，计算节点本地 IO 压力骤减，GC 也少了，单 Task 执行速度明显提升。另外，因为双副本机制，节点故障导致的 stage 重算也大量减少，整体作业稳定性自然就上去了。

4.3 搭配 AQE 和动态资源使用

如果你的 Spark 已经在用 AQE，记得把 spark.sql.adaptive.coalescePartitions.enabled 保持开启。Celeborn 对 AQE 的适配做得不错，reducer 合并后，Worker 上的数据文件不需要重新 shuffle，只需要在读取索引上做一次合并，这一下能省掉大量重复计算。

动态资源方面，我们开启了 spark.dynamicAllocation.executorIdleTimeout 配合 RSS。Executor 空闲退出不再担心 shuffle 数据丢失，因为数据在 Worker 上，新的 Executor 随时去拉就行。这一点在日批高峰期特别关键，集群资源吃紧时能更激进地释放空闲 Executor，把资源让给真正在跑的任务。

5. 上线半年踩过的坑与排查实录

5.1 Partition Not Found 背后的元数据竞态问题

初期我们遇到过一类诡异报错：某些 reduce 任务在拉数据时报 Partition File Not Found，但重试几次又能成功。一开始怀疑是 Worker 文件遗失了，后来排查发现是 Master 之间的元数据同步有时序问题。

具体来说，shuffle 在 map 端刚 commit 完成的时候，reducer 端如果立刻来查元数据，有可能拿到一个"尚未完成发布"的状态，导致查不到对应的 partition 文件。这个问题在作业峰值并发高时更容易出现。我们当时的处理方式是，在客户端适当增大元数据查询的重试次数和间隔，同时建议社区在 Master 端增加 commit 的可见性控制。如果你也遇到类似的偶发 not found，先看是不是网络抖动，再看 Master 端日志的 commit 时间，不要盲目怀疑 Worker 丢数据。

5.2 Worker 内存被堆外占用拖垮

某次压力测试，我们看到某个 Worker 的 off-heap 内存直接打满，进程卡了十几秒，之后一大堆 shuffle push 超时。排查时发现是我们的 push 并发开得太大，每个 Executor 设置了大量 push 线程，所有数据同时拥向一个 Worker，堆外内存瞬间爆掉。

解决思路分两步：一是控制 Executor 端 push 并发，spark.celeborn.client.push.retry.threads 调整到合理值；二是给 Worker 端的内存和 IO 做限流，让 Worker 不至于被打垮。大数据系统里最容易出的问题不是单点挂掉，而是"雪崩"，上游不控制速率，下游就被压死。Celeborn 提供了权重和限流配置，一定要用起来。

5.3 从 ESS 切换时的兼容性陷阱

有些 Spark 作业里，用户显式指定了 spark.shuffle.service.enabled=true，或者环境变量里残留了 ESS 相关配置。切换到 Celeborn 后，这些配置虽然不直接冲突，但会造成某些 Executor 仍尝试找 LocalShuffleCache 的错觉，表现为偶发的 fetch 失败。

更好的做法是迁移前做一次配置体检，把 spark.shuffle.service.enabled 显式关闭，把 spark.shuffle.manager 显式指定为 Celeborn 的实现，并在 YARN 侧移除 ESS 相关的 jar 包，避免 classloader 冲突。我们用脚本批量把所有作业的 default conf 刷了一遍，才彻底消掉这轮脏数据。

5.4 小作业反而变慢了

上线后我们发现，一些只有几 GB 数据的小任务，换到 RSS 后耗时反而变长。原因很好理解：小作业 shuffle 的数据量本身不大，本地磁盘写加读取可能就几秒钟，但推到远端 Worker 后多了网络传输 RTT，还要参与 Master 的注册与元信息协商。

针对这类作业，我们的做法是保持默认 ESS 或直接加一个参数开关：在小任务提交时绕开 RSS。你可以通过 spark.celeborn.enabled 这类开关按作业粒度控制。如果你们集群中小任务占比很高，建议优先保证这一类不受影响，不要一刀切全部迁移。这算是我最想提醒的一点：RSS 不是银弹，它的收益在数据量大、稳定性要求高的场景才最明显。

5.5 常见问题速查表

最后再说一点自己的体会

Celeborn 的落地不是一个"装好配置完"就结束的项目。从我这边实际操作的经验看，最关键的是节奏：先小流量，再逐步放量，同时配套完整的监控大盘。有事没事看一眼 Master 页面上的 Worker 内存曲线、shuffle 文件增长速度，你会发现很多隐患在高负载来之前就藏在趋势里。

灰度期间建议专门挑一些跑批链路中非核心但数据量大的作业，拿来做演练。等稳定性得到验证，再往核心链路推进。不夸张地说，RSS 替换 ESS 之后，我们的生产环境从"每天总有几个作业被 shuffle 搞挂"变成了"个位数失败且基本都能自动恢复"。如果你也想动 shuffle，别犹豫，先从最痛苦的作业开始试，Celeborn 值得投入。