EMR Serverless Spark 借助 Celeborn 实现存算分离与 10 倍 Shuffle 性能优化

为什么需要 Celeborn？Spark 原生 Shuffle 的四大瓶颈

在 Spark 中，Shuffle 主要用于解决宽依赖问题。当某些 RDD 操作（如 groupBy、reduceByKey、repartition 等）需要依赖上游每个 partition 的数据时，就必须通过 Shuffle 来重新分布数据。

传统的 Spark Sort Shuffle 流程中，每个 Map Task 首先对数据进行排序操作，将有序数据写入本地磁盘，然后每个 Reducer 从各个 Map 输出的文件中拉取对应分区的数据。这种架构存在几个显著的缺陷：

1. 缺乏存算分离能力
   在传统 Spark 架构中，如果某个作业需要写入大量 Shuffle 数据，就必须为集群配置相应规模的存储资源。但如果只有少数作业有大数据量需求，其他作业实际上并不需要如此多的存储空间，这就造成了资源的严重浪费。在 Serverless 场景中，这个问题尤为突出——无法根据实际需求灵活调配存储和计算资源。

2. M×N 的高网络连接数
   每个 Reducer 都需要与所有 Mapper 建立连接来获取对应 partition 的数据。假设有 M 个 Mapper 和 N 个 Reducer，总连接数将达到 M×N，网络连接数呈指数级增长。在大规模作业中，这会导致网络连接成为严重的瓶颈。

3. 大量随机 IO 操作
   由于每个 Reducer 需要从多个 Map 输出文件中读取数据，每次读取都是一次独立的 IO 操作，导致大量的随机 IO，严重影响读取性能。

4. 单副本带来的稳定性风险
   原生 Spark 采用单副本机制，一旦某个节点出现故障，整个作业就会失败，需要重新执行，稳定性和可靠性都难以保证。

综合来看，传统 Shuffle 架构在灵活性、稳定性和可扩展性方面都存在明显不足。

Celeborn 是什么？开源社区验证的大数据 Shuffle 新范式

Celeborn 正是为了解决上述问题而诞生的大数据中间数据服务。从诞生至今，Celeborn 已经建立了活跃的开源社区，吸引了众多企业开发者参与共建，成为了 Apache 的顶级项目。当前社区版本已迭代至 0.6 版本。

Celeborn 支持 MapReduce、Flink、Spark（包括 Gluten、Blaze 等加速引擎）等多种大数据计算引擎，真正实现了存算分离的架构理念。在生产环境中，Celeborn 已经得到充分验证：单个集群可支持超过 2500 个节点，每日处理的 Shuffle 数据量超过 10PB，展现出卓越的性能和稳定性。

1. Celeborn 的架构设计

Celeborn 采用 Master-Worker 的经典分布式架构。Master 层基于 Raft 协议实现高可用，支持单副本和多副本部署模式，确保元数据管理的可靠性。Worker 集群负责实际的数据存储，每个 Worker 定期向 Master 发送心跳，汇报健康状态和资源使用情况。

在计算集群侧，Celeborn 通过 LifecycleManager 组件与 Spark Driver、Flink JobManager 或 YARN ApplicationMaster 集成。LifecycleManager 负责向 Celeborn Master 申请资源，建立起计算引擎与 Celeborn 集群之间的桥梁。一旦资源申请完成，计算节点的 Executor 或 Task 就可以直接与 Celeborn Worker 进行数据读写交互。

整个架构的数据流向清晰明确：Driver/JobManager 通过 LifecycleManager 向 Master 申请资源 → Master 分配 Worker 资源 → Executor/Task 向 Worker 推送数据 → Reducer 从 Worker 拉取数据。这种设计彻底实现了计算与存储的解耦。

2. Celeborn 的核心技术特性

Celeborn 的两大核心技术特性是 Push Shuffle 和 Partition Split，几乎所有的性能优化都基于这两个基础机制。

Push Shuffle 机制：

与传统 Spark 的 Pull 模式不同，Celeborn 采用 Push Shuffle 模式。每个 Map Task 会根据不同的 Partition 将数据主动推送到预先申请的 Celeborn Worker 资源中，数据按 Partition 分别写入不同的文件。这样，Reducer 在读取数据时，只需要从对应的 Partition 文件中顺序读取，无需与所有 Mapper 建立连接。

这带来了两个显著优势：

• 网络连接数从 M×N 降低到 N 级别，大幅减少网络开销。
• 将随机 IO 转变为顺序 IO，显著提升读取性能。

Partition Split 机制：

当单个 Partition 文件的大小超过配置的阈值时，LifecycleManager 会自动向 Celeborn 集群申请新的资源，将同一个 Partition 的后续数据写入新的文件。虽然物理上是不同的文件，但逻辑上它们属于同一个 Partition。

这个看似简单的机制，为 Celeborn 的动态扩容、负载均衡和故障自愈提供了坚实的技术基础。当集群新增节点时，正在运行的作业可以通过 Partition Split 将新数据写入新节点，实现动态资源利用；当某个 Worker 压力过大时，可以通过 Split 将数据分流到其他健康节点，避免单点过载。

Serverless Spark 实践：Celeborn 如何让 Shuffle 更可靠高效？

在 EMR Serverless Spark 的生产环境中，Celeborn 展现出了多方面的实用优势。从完善的监控体系、无感知的升级扩容，到智能的故障自愈和对 Spark AQE 的原生支持，Celeborn 为企业级大数据处理提供了坚实保障。

1. 舒心：完整的端到端监控能力

Celeborn 自带完整的指标体系，并原生集成 Prometheus，提供端到端的监控能力。使用者无需进行复杂的配置，就能建立起完善的监控系统。

在实际运维中，Celeborn 提供了丰富的监控维度，主要包括：

• 资源消耗监控（Resource Consumption）：清晰展示用户或租户在特定时间段内的 Shuffle 数据写入量、集群节点状态（总节点数 vs. 使用中节点数），帮助判断负载与扩容状态。
• 关键性能指标：
  • Direct Memory 使用率：反映 Worker 的内存压力，是判断是否需要触发反压机制的关键指标。
  • CPU 使用率：直观展示 Worker 的计算压力。
  • 带宽使用率：监控网络 IO 是否成为瓶颈。
  • 可用磁盘空间：预警整个集群的存储压力。

2. 放心：智能流控 + 动态扩缩容 + 故障自愈

（1）反压与流控机制

面对高压力场景，Celeborn 设计了两层保护机制：反压（Backpressure）和流控（Flow Control）。

反压机制：
当某个 Worker 的内存使用率达到高水位阈值时，会暂停客户端向该 Worker 继续写入数据，转而全力将内存中的数据刷写到磁盘。等待该 Worker 的状态恢复健康后，再恢复正常的数据写入。这种机制有效防止了内存溢出导致的 Worker 崩溃。

流控机制：
Celeborn 的流控机制类似于 TCP 的拥塞控制。在短时间内面对超大流量时，Worker 可能无法立即承载，流控机制会暂缓作业的数据写入速率，避免整个集群因瞬时流量过大而崩溃。
通过 CIP-1（拥塞控制）和 CIP-11（扩展拥塞控制）提案，Celeborn 还实现了多租户级别的流控能力，可以针对不同租户设置不同的流控策略，保证资源的公平分配。

（2）无感知的升级与扩缩容

缩容场景：Decommission 机制
当某个 Worker 进入 Decommission 状态时，意味着该 Worker 不再接受新的作业分配，但会继续完成已分配的任务，类似于“不接新活，但干完手头活再下班”。任务完成后，Worker 可安全下线进行升级或缩容移除。

扩容场景与动态资源能力

在 Serverless Spark 环境中，主要有两种扩容场景：

• 带宽压力扩容：作业并发大、写入速度快，导致带宽打满、CPU 和内存压力高。
• 磁盘容量扩容：作业数据量大，磁盘空间即将用尽。

Celeborn 通过 Partition Split 机制优雅地解决了“新扩容节点如何分担老作业压力”的问题。正在运行的作业，当数据达到 Partition 的 Split 阈值后，会自动通过 LifecycleManager 向新加入的 Worker 申请资源，将后续数据写入新 Worker。新老 Worker 并行工作，共同完成数据写入。

（3）故障自愈 & Stage Rerun

故障自愈：
当某个 Worker 的磁盘使用率达到较高水位（如 64%，可配置）时，它仍然可用。但如果继续写入导致使用率接近危险阈值（如 98%），Celeborn 会自动停止向该 Worker 写入数据，主动触发 Split，将数据分流到其他健康的 Worker。通过预留 Disk Reserve Size 空间，可以防止 Worker 被“写爆”，并将压力均匀分摊，这本身就是故障自愈能力的体现。

Stage Rerun 机制（CIP-4）：
早期版本中，若 Worker 磁盘损坏导致数据丢失，通常需要依赖成本翻倍的双副本机制。Celeborn 引入了 Stage Rerun 能力，大幅降低了容错成本。当 Reducer 读取数据时发现某部分数据丢失，Celeborn 会向 Spark 返回 ChunkFetchFailureException 错误，触发重跑数据丢失对应的 Stage。
整个作业不会失败，只需重跑丢失数据涉及的 Stage，恢复速度远快于作业整体重跑，且无需双副本，节省了存储和写入成本。通过 Master-Worker 间的心跳监控，能够快速发现异常并触发该流程。

3. 安心：原生支持 Spark AQE 优化，应对真实业务复杂性

Spark 的自适应查询执行（AQE, Adaptive Query Execution）是重要的性能优化特性，主要包括 Partition 合并、Join 策略切换和 Skew Join 优化。要支持 AQE，Shuffle 服务必须具备两个核心能力：

• Partition Range Read：一个 Reducer 读取多个 Partition 的数据。
• Mapper Range Read：一个 Reducer 读取多个 Mapper 的数据。

（1）Partition Range Read：

这对 Celeborn 来说是天然支持的。无论一个 Partition 被 Split 成几个文件，逻辑上仍然是一个 Partition。Reducer 读取一个 Partition 和读取多个 Partition，在实现上没有本质区别。

（2）Mapper Range Read 的挑战：

Mapper Range Read 则更具挑战性。原本一个 Mapper 的数据在本地是有序的，但 Celeborn 的 Push Shuffle 将不同 Mapper 的数据分散到了不同的 Worker。要重新合并，需要对 Partition 中的数据按 Mapper 维度重新排序，这个排序操作 IO 开销大，存在超时风险。

在真实生产环境中，Mapper Range Read 场景很常见：

• 场景一：Spark Skew Join

  

  当某个 partition 数据特别大时，让单个 Reducer 读取会非常慢。Spark AQE 会将这个大 partition 拆分成多个 Sub Reduce Task 并行读取，这就需要按 Mapper 维度进行数据划分。

• 场景二：LocalShuffleReader（执行计划动态切换）

  

  当 Sort Merge Join 的某个表经过 Filter 后变得很小，Spark AQE 会动态将执行计划切换为 Broadcast Hash Join。这个过程中会产生一个 Broadcast Exchange 阶段，需要进行 Mapper Range Read。

这些场景往往伴随着严重的数据倾斜。

（3）Celeborn 的创新解决方案：Chunk Read

Celeborn 在 0.5 版本中将 Sort Read 优化为 Chunk Read，实现了性能的飞跃。

• Sort Read（旧）：像让 N 个人排队自己切蛋糕，每个人切多吃得久，整体等待时间长。
• Chunk Read（新）：直接预先切成 N 块均匀的蛋糕，每人拿一块直接吃，无需等待，效率显著提高。

具体来说，Chunk Read 将 Mapper Range 读取转换为 Partition Chunk 读取。数据被分成均匀的 Chunk，每个 Sub Reduce Task 读取相应的 Chunk Set，避免了排序过程，消除了排序超时风险，实现了更均匀的负载分配。在测试中，Chunk Read 在特定场景下可以达到 10 倍的性能提升。

（4）应对数据倾斜的并发 Shuffle Writer

Partition Split 机制面临数据倾斜挑战：当某个 Partition 特别大、需要不断 Split 时，串行阻塞的 Split 会产生长尾效应——写入快的 Partition 需要等待写入慢的 Partition。

Celeborn 社区对此提出了积极的优化方案，例如动态并行度优化：根据 Split 速度动态计算需要分配的资源数量，将单线 Split 变为多线 Split，实现并发的 Shuffle Writer。该方案在生产验证中，在特定场景下同样带来了 10 倍的性能提升。

未来规划：更强大的 Celeborn 生态

1. 增强拥塞控制与配额管理

Celeborn 未来将实现多层次的精细化管理：

• 多层次拥塞控制：集群层面、租户层面、用户层面、Worker 层面的写入速度控制。
• 多维度配额控制：针对不同用户、租户设置资源配额（如 Shuffle 数据总量、并发任务数）。当超过配额时，可自动中断大作业或进行拥塞控制，确保集群资源公平分配。

2. 版本演进与新功能开发

EMR Serverless Spark 正快速推进 Celeborn 版本与社区（0.6）保持同步。同时，将持续进行 0.7 版本新特性开发：

• Auto Scale（自动扩缩容）：根据负载自动调整集群规模。
• App Priority（应用优先级）：支持不同优先级的作业调度。
• Parallel Shuffle Writer（并发 Shuffle Writer）：将高效的并发 Split 能力正式合入主线。

这些新特性将进一步提升 Celeborn 的自动化运维能力和资源利用效率，为云原生时代的企业级大数据处理提供更加智能、高效的 Shuffle 服务。如果你对 Celeborn 的技术实现或在大数据生态中的应用有更多疑问，欢迎到云栈社区的相应板块进行交流探讨。