导读Overview

本文根据 EMR Serverless Spark 团队帅镇涛的技术分享整理而成。帅镇涛负责 Celeborn 在 EMR Serverless Spark 中的研发工作，在本次分享中，他详细介绍了 Celeborn 这一大数据中间数据服务如何解决传统 Spark Shuffle 面临的存算分离、网络连接数、IO 性能和容错能力等核心问题。

为什么需要 Celeborn？

Spark 原生 Shuffle 的四大瓶颈

在 Spark 中，Shuffle 主要用于解决宽依赖问题。当某些 RDD 操作(如 group by、reduce by、repartition 等)需要依赖上游每个 partition 的数据时，就必须通过Shuffle来重新分布数据。传统的 Spark Sort Shuffle 流程中，每个 Map Task 首先对数据进行排序操作，将有序数据写入本地磁盘，然后每个 Reducer 从各个 Map 输出的文件中拉取对应分区的数据。

这种架构存在几个显著的缺陷：

1. 缺乏存算分离能力

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

2. M×N 的高网络连接数

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

3. 大量随机 IO 操作

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

4. 单副本带来的稳定性风险

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

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

Celeborn 是什么？

开源社区验证的大数据 Shuffle 新范式

Celeborn 正是为了解决上述问题而诞生的大数据中间数据服务。从诞生至今，Celeborn 已经建立了活跃的开源社区，拥有众多企业用户。并且经历了多个版本迭代，当前社区版本已迭代至 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 数据写入量。例如用户在几个小时内写入了多少数据，这个数据量是否异常，异常大数据量会给集群造成较大的压力，一些集群的压力指标可以帮助运维人员了解集群的实际负载情况。同时，监控也会显示 Celeborn 集群的节点状态——总节点数与正在使用的节点数的差异，反映出集群是否处于扩容、缩容或升级状态。

关键性能指标：

• Direct Memory(直接内存使用率)： 反映 Worker 的内存压力，是判断是否需要反压的重要指标

• CPU 使用率： 数值显示当前正在使用的 Worker CPU 核数，可以直观判断计算压力

• 带宽使用率： 监控网络 IO 是否成为瓶颈

• 可用磁盘空间： 显示整个集群当前可用的磁盘容量，预警存储压力

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

（1）反压与流控机制

在某个高并发作业场景中，30 分钟内写入了约 90TB 数据。监控显示，这段时间内带宽利用率接近饱和，CPU 使用率也达到了较高水平（在 16 核机器上，部分 Worker 的 CPU 已经用满）。由于 CPU 用满导致数据来不及刷盘，内存占用也随之升高。根据这些指标，系统自动触发了扩容机制，确保集群能够承载突发的高压力负载。面对高压力场景，Celeborn 设计了两层保护机制：反压（Backpressure）和流控（Flow Control）。

反压机制：

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

流控机制：

Celeborn 的流控机制类似于 TCP 的拥塞控制。在短时间内面对超大流量时，Worker 可能无法立即承载，流控机制会暂缓作业的数据写入速率，避免整个集群因瞬时流量过大而崩溃。

通过 CIP-1 (拥塞控制)和 CIP-11 (扩展拥塞控制)提案，Celeborn 还实现了多租户级别的流控能力，可以针对不同租户设置不同的流控策略，保证资源的公平分配。

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

缩容场景： Decommission 机制

当某个 Worker 进入 Decommission 状态时，意味着该 Worker 不再接受新的作业分配，但会继续完成已分配的任务。就像一个员工"上完这个班再下班"——不接新活，但把手头的活干完。完成所有任务后，Worker 可以安全地下线进行升级，或者直接进行缩容移除。

扩容场景：Dynamic Resource 动态资源+故障自愈+Stage Rerun

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

• 带宽压力扩容： 作业并发大、写入速度快，导致带宽打满、CPU 和内存压力高

• 磁盘容量扩容： 作业数据量大，磁盘空间即将用尽

假如某作业需要写入 500 多 TB 的 Shuffle 数据，而集群初始磁盘容量不足 500 TB。在扩容过程中，集群总容量勉强达到 600 TB。此时面临一个关键问题：新扩容的 Worker 能否分担正在运行作业的压力?

Dynamic Resource 动态资源能力：

Celeborn 通过 Partition Split 机制优雅地解决了这个问题。正在运行的作业，当数据达到 Partition 的 Split 阈值后，会自动通过 Lifecyclemanager 向新加入的 Worker 申请资源，将后续数据写入新的 Worker。这样，新节点可以立即分担正在运行作业的压力，而不是只能服务于新启动的作业。同时，老的 Worker 也不会立即停止写入，因为它们仍处于可用状态。新老 Worker 并行工作，共同完成作业的数据写入。

故障自愈

当某个 Worker 的磁盘使用率达到 64% 等较高水位时（可配置），该 Worker 仍然可用。但如果继续写入导致磁盘使用率接近 98%，Celeborn 会自动停止向该 Worker 写入数据，主动触发 Split，将数据分流到其他健康的 Worker。

这个配置项叫做 Disk Reserve Size，通过预留一定的磁盘空间，可以防止 Worker 被「写爆」，同时将压力均匀分摊到新上线的 Worker 上。这也是 Celeborn 故障自愈能力的一个重要体现——当某个 Worker 压力过大、健康状态不佳时，会自动临时下线，等待恢复正常后再重新加入。

Stage Rerun

在早期版本中，如果某个 Celeborn Worker 的磁盘损坏导致数据丢失，唯一的解决方案是增加 Worker 节点，并使用双副本机制——同一份数据写两次到不同的 Worker。虽然这能保证可靠性，但成本翻倍，对于企业来说并不经济。

Stage Rerun 机制（Cip-4）：

Celeborn 在近期版本中引入了 Stage Rerun 能力，大幅降低了容错成本。当 Reducer 读取数据时发现某部分数据丢失，Celeborn 会向 Spark 返回 Chunkfetchfailureexception 错误，告知 Spark 需要重跑数据丢失对应的 Stage。

整个作业不会失败，只需要重新执行丢失数据涉及的 Stage，然后重新写入和读取这部分数据即可。这种局部重跑机制，恢复速度比整个作业失败后重跑要快得多，而且不需要双副本，节省了存储成本和写入开销。

通过 Master-Worker 之间的心跳监控和故障检测，Celeborn 能够快速发现 Worker 异常或丢失，触发 Stage Rerun 流程，实现真正的故障自愈。

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 开销大，存在超时风险。在 Serverless Spark 生产环境中，Mapper Range Read 场景很常见：

场景一：Spark Skew Join

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

场景二：LocalShuffleReader（Sort Merge Join转Broadcast Hash Join）

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

在实际生产案例中，这两种场景不仅常见，而且往往伴随着数据倾斜——不仅 Partition 之间有倾斜，即使按 Task 划分后，各个 Task 的数据量仍然不均匀。

（3）Celeborn 的创新解决方案

Trunk Read

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

用一个生动的比喻来说明：

Sort Read：像是让 6 个人排队自己切蛋糕，每个人想吃多少切多少。切得多的人吃得久，切得少的人吃得快，快的人还要等慢的人，整体时间很长。

Chunk Read：直接按固定方式切成 6 块均匀的蛋糕，每个人拿一块直接吃。大家吃的时间一样长，而且不用排队，显著提高了效率。

具体来说，Chunk Read 将 Mapper Range 读取转换为 Partition Chunk 读取。数据被分成均匀的 Chunk，每个 Sub Reduce Task 读取相应的 Chunk Set，避免了排序过程，消除了排序超时风险，实现了更均匀的负载分配。

在测试中，Chunk Read 在特定场景下可以达到 10 倍的性能提升，特别是在 Task 级别也存在数据倾斜的情况下，效果更加显著。

应对数据倾斜的并发 Shuffle Writer

Partition Split 机制虽然强大，但也会面临数据倾斜带来的挑战。当某个 Partition 特别大，需要不断 Split 时，会产生长尾效应——写入快的 Partition 需要等待写入慢的 Partition。

原因在于 Celeborn 一直以来的 Partition Split 是串行阻塞的，只有在需要时才进行 Split，无法一次性拆分多个。

社区的多种优化方案：

Celeborn 社区对这个问题展现出很高的活跃度，提出了多种解法：

方案一：指数级 Split 第一次 Split 分 1 个，第二次分 2 个，第三次分 4 个，随着 Split 次数增加，每次分配的资源数量呈指数增长。

方案二：动态并行度优化 根据 Split 速度动态计算需要分配的资源数量。如果 Split 很快，说明数据写入速度快，就多分配一些；如果 Split 总数不多且比较慢，就少分配，同时预留一些冗余空间继续写入。

这种方案将单线 Split 变成了多线 Split，实现了并发的 Shuffle Writer。虽然这个方案还在社区讨论中尚未合入主分支，但在生产环境中已经得到验证，在特定场景下同样可以带来 10 倍的性能提升。

未来规划

更强大的 Celeborn 生态

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

Celeborn 未来将实现多层次的拥塞控制和配额管理机制：

多层次拥塞控制：

• 集群层面： 控制整个集群的流量上限

• 租户层面： 控制单个租户的写入速度

• 用户层面： 控制单个用户的 Shuffle 速度

• Worker 层面： 控制单个 Worker 的负载压力

多维度配额控制： 

除了速度控制，还将支持多维度的配额限制，可以针对不同的用户、租户设置不同的资源配额。当超过配额时，可以自动中断大作业或对大作业进行拥塞控制，确保集群资源的公平分配。

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

EMR Serverless Spark 目前使用的是 Celeborn 0.5 版本，而社区已经迭代到 0.6 版本。

我们正快速推进 Serverless Spark 的 Celeborn 与社区版本保持同步，获得社区最新的稳定性和性能优化。

• 此外，我们将持续进行 0.7 版本新特性开发：

• Auto Scale（自动扩缩容）：根据负载自动调整集群规模

• App Priority（应用优先级）：支持不同优先级的作业调度

• Parallel Shuffle Writer（并发 Shuffle Writer）：正式将并发 Split 能力合入主线

这些新特性将进一步提升 Celeborn 的自动化运维能力和资源利用效率，为企业级大数据处理提供更加智能和高效的 Shuffle 服务。