1. 项目概述：当Serverless Spark遇上Shuffle，一场性能与成本的硬仗

如果你在云上跑过Spark，尤其是EMR Serverless Spark这种按需付费的弹性服务，那你一定对Shuffle这个环节又爱又恨。爱的是，它是实现复杂数据处理逻辑（如Join、GroupBy、Aggregation）的基石；恨的是，它往往是作业性能的“阿喀琉斯之踵”，更是成本失控的潜在元凶。在传统的、有固定计算集群的环境里，Shuffle的问题虽然棘手，但好歹我们可以通过堆资源、调优参数、甚至“人肉运维”来勉强应对。然而，当场景切换到Serverless Spark，游戏规则彻底变了。

Serverless的核心魅力在于“按需使用，用完即走”，你无需关心底层服务器的采购、部署和维护。但这也意味着，你失去了对计算节点生命周期的直接控制。在EMR Serverless Spark中，执行任务的Executor是动态申请和释放的。想象一下这个场景：一个Stage的Executor完成了自己的计算任务，生成了Shuffle数据，然后就被回收了。此时，下游Stage的Executor启动，需要去读取这些Shuffle数据，却发现数据来源的“房东”已经“退租消失”了。这就是Serverless Spark面临的经典Shuffle挑战——计算与存储的强耦合被打破，数据可靠性（Reliability）和可用性（Availability）成了大问题。

传统的Spark Shuffle（如SortShuffleManager）将中间数据写在每个Executor的本地磁盘上。这在本机环境下高效，但在Serverless场景下就是一场灾难。节点一旦释放，其本地磁盘上的数据也随之灰飞烟灭，导致下游任务无法获取数据而失败。为了解决这个问题，社区和云厂商早期普遍采用“推”的方式，比如将Shuffle数据写到远端稳定的存储系统，如HDFS或S3。这确实解决了数据持久化的问题，但引入了新的性能瓶颈：大量的随机小IO写入对象存储，延迟高、成本也不菲，而且S3这类存储的“最终一致性”模型也可能带来数据可见性的问题。

正是在这样的背景下，Celeborn（原名Remote Shuffle Service， RSS）走进了我们的视野。它并非为Serverless而生，但其架构理念恰好完美地契合了Serverless Spark的需求。简单来说，Celeborn是一个独立的、专为Shuffle设计的服务。它将Shuffle数据的存储和计算解耦：Executor不再自己保存Shuffle数据，而是将数据“推送”到独立的Celeborn集群节点上；下游的Executor则从Celeborn集群“拉取”所需的数据。对于EMR Serverless Spark而言，Celeborn就像是一个高可用、高性能的“Shuffle数据交换中心”，让动态的、临时的计算节点可以安心地“生”数据，也可以放心地“取”数据。

所以，当我们谈论“Celeborn如何让EMR Serverless Spark的Shuffle舒心、放心、安心”时，我们实际上是在探讨一套系统工程：如何通过一个外部服务，解决Serverless弹性模式下的数据可靠性顽疾，同时还要兼顾性能、成本和运维的复杂度。这不仅仅是换一个Shuffle Manager那么简单，它涉及到架构选型、部署集成、参数调优和故障诊断的全链路。接下来，我将结合实践，为你层层拆解。

2. Celeborn架构精解：为什么它是Serverless Spark的“解药”

要理解Celeborn为何有效，必须深入其架构设计。它的核心思想是“存算分离”和“服务化”，但这并非简单的远程存储，而是一套精心设计的、针对Shuffle工作负载特性的系统。

2.1 核心组件与数据流

Celeborn集群主要由两类角色构成：Master和Worker。Master负责集群管理和元数据协调，Worker则是实际存储和提供Shuffle数据服务的节点。在EMR Serverless场景下，Celeborn集群通常是独立于Spark计算集群部署的、长期运行的服务。

一个典型的Shuffle数据流如下：

1. 注册与分配：Spark Driver在作业启动时，会向Celeborn Master注册该应用。当Executor启动并需要输出Shuffle数据时（即Shuffle Map Task），它会向Master请求分配一些Worker节点作为数据推送的目标。
2. 数据推送（Push）：Executor（作为Push Client）将产生的Shuffle数据分区（Partition），通过Netty连接直接推送到分配给它的一个或多个Celeborn Worker上。这里有一个关键优化：并行推送与副本机制。Celeborn支持将同一个分区的数据同时推送到多个Worker（默认副本数为2），这极大地提高了数据的可靠性，即使某个Worker节点宕机，数据依然可从副本读取。
3. 数据存储：Worker接收到数据后，会将其写入本地存储（如SSD、HDD或内存），并管理这些数据的元信息。与直接写S3不同，这是顺序写入本地盘，性能有数量级的提升。
4. 数据读取（Fetch）：当下游的Stage启动，其Executor（作为Fetch Client）需要读取Shuffle数据时，它会向Master查询所需数据分区的存放位置（即位于哪些Worker上），然后直接连接对应的Worker拉取数据。

这个流程看似简单，但每个环节都针对Shuffle的痛点进行了优化：

• 解耦计算与存储：Executor释放不影响已推送到Celeborn的数据，解决了Serverless的核心痛点。
• 数据高可用：多副本机制避免了单点故障导致作业失败。
• 高效I/O：从计算节点的随机写+下游读，转变为向Celeborn的顺序写+从Celeborn的顺序读，充分利用了本地磁盘的性能。

2.2 针对Serverless场景的关键设计

Celeborn的许多特性，仿佛是为EMR Serverless Spark量身定做：

1. 动态资源适配：Celeborn Worker节点是常驻的，而Spark Executor是动态的。这种模式天然匹配。Celeborn集群的容量可以根据历史Shuffle数据量进行规划并保持稳定，无需随Spark作业的伸缩而频繁变动。
2. 处理慢节点（Straggler）与数据倾斜：Celeborn有一个重要的“延迟推送”机制。当某个Executor（Push Client）推送数据特别慢（可能因为负载高、GC或网络问题），Worker可以主动向Master汇报。Master可以协调其他正常的Executor，帮助这个慢节点推送剩余的数据，避免单个慢任务拖垮整个Stage。这对于Serverless环境中可能出现的性能波动节点是一个有效的容错手段。
3. 高效的垃圾回收（GC）：Shuffle数据是中间数据，作业结束后必须清理。Celeborn与Spark Driver紧密集成，当Spark作业结束时，Driver会通知Celeborn Master，由Master协调所有Worker清理该作业相关的所有Shuffle数据。这避免了孤儿数据占用存储空间，对于需要为存储付费的云环境至关重要。
4. 与云存储的互补：虽然Celeborn使用本地存储，但它的定位是高性能缓存层，而非最终存储。对于超大规模Shuffle或需要长期保留Shuffle数据的场景（如容错重试），可以结合云存储使用。Celeborn社区也在探索分层存储等更高级的特性。

注意：Celeborn并非银弹。它引入了新的组件，意味着你需要维护一个额外的、高可用的Celeborn集群。在EMR这类托管服务中，这个集群通常由云服务商负责运维和保障，对用户透明，这是选择托管服务的一大优势。如果是自建，则需要考虑Master的HA、Worker的监控扩缩容等运维成本。

3. 在EMR Serverless Spark中集成Celeborn：实操指南

理论很美好，但让Celeborn在EMR Serverless Spark中真正跑起来，需要正确的配置和调优。下面我以一个实际作业的配置过程为例，详细说明关键步骤。

3.1 环境准备与基础配置

首先，确保你的EMR Serverless应用配置了正确的Celeborn客户端。通常，EMR运行环境会预置Celeborn相关的JAR包。你需要通过Spark配置项来启用它。

核心的Spark配置如下（可以在创建Application时通过spark-defaults.conf或直接作为配置参数传入）：

# 启用Celeborn作为Shuffle管理器 spark.shuffle.manager org.apache.spark.shuffle.celeborn.RssShuffleManager # 指定Celeborn Master的地址，EMR Serverless通常提供内部服务发现，这里可能是预定义的 spark.celeborn.master.endpoints <master-host>:<port> # 对于EMR，这个地址可能是类似“celeborn-master.${cluster-id}.internal:9097”的形式 # 声明使用RSS的Shuffle方式 spark.shuffle.service.enabled false spark.sql.adaptive.skewJoin.enabled true # 结合Celeborn处理数据倾斜更有效

在EMR Serverless控制台创建应用时，你可以在“配置”部分以JSON格式指定这些参数：

{ "classification": "spark-defaults", "properties": { "spark.shuffle.manager": "org.apache.spark.shuffle.celeborn.RssShuffleManager", "spark.celeborn.master.endpoints": "ip-xx-xx-xx-xx.ec2.internal:9097", "spark.serializer": "org.apache.spark.serializer.KryoSerializer" } }

实操心得：在首次配置时，最常遇到的坑就是spark.celeborn.master.endpoints地址不对。在EMR托管环境中，最佳实践是查阅当前EMR版本和Serverless环境的官方文档，找到Celeborn服务的内网端点。直接使用IP可能因为集群伸缩而变化，使用内部DNS名称更可靠。如果作业启动失败，日志中出现“Cannot connect to Celeborn Master”的错误，首先排查这个地址。

3.2 关键参数调优详解

启用只是第一步，调优才能发挥最大效力。Celeborn提供了大量参数，以下是与性能和稳定性最相关的几个：

1. celeborn.push.replicate.enabled(默认: true)

   • 作用：是否启用数据多副本。对于生产环境的Serverless作业，强烈建议保持开启（true）。这是数据高可用的基石。虽然会增加网络开销和存储开销，但相比因节点丢失导致整个作业失败的成本，这个开销是值得的。
   • 调优建议：除非是在做性能极限压测且对成本极度敏感的非关键测试，否则不要关闭它。

2. celeborn.push.buffer.size(默认: 64k) 和celeborn.push.queue.capacity(默认: 512)

   • 作用：这两个参数共同决定了推送数据的缓冲机制。buffer.size是每次推送的数据块大小，queue.capacity是内存中缓冲队列的容量。
   • 调优建议：如果你的作业Shuffle数据量巨大（TB级别），且Executor内存充足，可以适当增大buffer.size（例如128k或256k）和queue.capacity（例如1024）。这可以减少网络发送次数，提升吞吐量。但要注意，过大的buffer会占用更多内存，可能加剧GC压力。一个平衡的方法是观察Celeborn Worker的日志和网络监控，如果网络利用率不高且CPU有空闲，可以尝试调大。

3. celeborn.fetch.chunk.size(默认: 8m)

   • 作用：下游Executor从Celeborn Worker拉取数据时的块大小。
   • 调优建议：增大此值可以减少拉取请求的次数，对于Shuffle Read量很大的作业有积极影响。可以尝试设置为16m或32m。但同样需要权衡：过大的块可能使单个请求耗时变长，在存在数据倾斜时，容易造成单个Task处理时间过长。建议根据作业的Shuffle Read Size/Total Task的平均值来调整。

4. celeborn.worker.storage.dirs和celeborn.worker.memory.reserved

   • 作用：这两个是CelebornWorker侧的参数，通常由EMR运维团队在部署Celeborn集群时设置。但作为用户，了解它们有助于你判断集群容量。
   • storage.dirs：Worker用于存储数据的本地磁盘目录。更多的磁盘通常意味着更好的I/O并行性。
   • memory.reserved：Worker预留的内存，用于缓存热数据或元数据。在Shuffle极度频繁的作业中，适当增加此值可以提升读取性能。

一个针对大规模ETL作业的调优配置示例：

spark.shuffle.manager org.apache.spark.shuffle.celeborn.RssShuffleManager spark.celeborn.master.endpoints celeborn-master.emr.internal:9097 spark.celeborn.push.replicate.enabled true spark.celeborn.push.buffer.size 128k spark.celeborn.push.queue.capacity 1024 spark.celeborn.fetch.chunk.size 16m # 启用Kryo序列化以减少Shuffle数据大小 spark.serializer org.apache.spark.serializer.KryoSerializer spark.kryoserializer.buffer.max 256m

3.3 部署模式与资源规划

在EMR Serverless中，Celeborn集群的部署模式通常有两种：

• 共享集群模式：一个Celeborn集群为多个EMR Serverless应用（甚至多个租户）提供Shuffle服务。优点是资源利用率高，运维成本低。缺点是可能存在“吵闹邻居”问题，一个重度Shuffle作业可能影响其他作业。
• 专属集群模式：为重要的生产作业或部门单独部署Celeborn集群。优点是性能隔离性好，资源有保障。缺点是成本较高。

作为用户，你可能无法直接选择部署模式，但可以通过观察Shuffle性能指标来判断。如果发现Shuffle Fetch延迟不稳定，时高时低，且与其他作业的启动时间有相关性，可能就是共享集群模式下的资源争抢。此时，可以向运维团队反馈，考虑对关键作业进行资源隔离或使用专属集群。

资源规划经验：

• Celeborn Worker磁盘：规划容量时，一个粗略的估计是预计日均Shuffle数据总量 * 副本数 * 数据保留天数（通常为作业最长运行时间+缓冲）。例如，每天产生1TB Shuffle数据，副本为2，作业最长运行1天，则至少需要2TB的可用存储。建议预留20%-30%的缓冲空间。
• 网络带宽：Celeborn集群的网络是瓶颈之一。确保Worker节点有足够的网络带宽（例如10Gbps或更高），特别是当Executor数量众多，同时进行Push和Fetch操作时。

4. 性能对比与效果验证：数据说话

配置好了，参数也调了，效果到底如何？我们不能只凭感觉，必须用数据来验证。我设计了一个对比实验，使用同一个复杂的Spark SQL作业（涉及多张大表的Join和Aggregation，Shuffle数据量约500GB），分别在EMR Serverless Spark上使用默认Shuffle（写S3）和启用Celeborn两种模式下运行。

结果分析：

1. 性能提升显著：总耗时减少超过三分之一，核心的Shuffle读写阶段耗时减半。这主要得益于Celeborn将远程S3的随机小IO高延迟访问，转换为了本地（对Celeborn Worker而言）的顺序IO访问。
2. 成本直接下降：因为作业运行时间缩短，消耗的Serverless vCPU小时数自然减少，直接降低了计算成本。同时，S3 API调用次数的锐减也降低了潜在的存储请求成本。
3. 稳定性增强：摆脱了对S3强一致性的依赖，Celeborn的多副本机制和内部重试逻辑，使得作业对底层存储的瞬时故障容忍度更高。

踩坑记录：在最初对比测试时，我曾发现启用Celeborn后作业速度反而变慢。经过排查，原因是Celeborn Worker节点所在的EC2实例类型网络带宽不足（当时用的是通用型实例）。当上百个Executor同时向少量Worker推送数据时，网络成为瓶颈。后来将Worker节点更换为网络优化型实例（如c5n.4xlarge），性能立即得到巨大改善。所以，Celeborn Worker本身的资源配置（特别是CPU、内存、网络和磁盘IO）是性能的关键前提。

5. 故障排查与运维实践

即使有了Celeborn，在复杂的生产环境中问题依然可能出现。这里分享几个典型的故障场景和排查思路。

5.1 常见问题速查表

5.2 监控与可观测性建设

要让Celeborn真正让人“放心”，完善的监控必不可少。除了基础的Spark UI（其中会显示Shuffle Read/Write的细节，但数据源变成了Celeborn），你还需要关注Celeborn集群本身的指标。

关键监控指标：

• Celeborn Master：活跃应用数、Worker注册数、请求QPS/延迟。
• Celeborn Worker：
  • 磁盘使用率：celeborn_worker_storage_used_bytes/celeborn_worker_storage_available_bytes。设置告警，超过80%需要预警。
  • 活跃连接数：celeborn_worker_active_connection_count。反映当前负载。
  • 推送/拉取吞吐量：celeborn_worker_push_data_bytes/celeborn_worker_fetch_data_bytes。观察流量是否均衡，是否存在热点Worker。
  • 推送/拉取延迟：celeborn_worker_push_data_time/celeborn_worker_fetch_data_time。延迟突增通常是性能问题的先兆。
• Spark侧集成：通过Spark的Metrics系统，可以将Celeborn客户端的指标（如推送失败次数、重试次数）导出到Prometheus等监控系统，与作业级别的指标关联分析。

实操心得：我们曾遇到一个周期性出现的作业变慢问题。通过监控发现，每天下午特定时间点，某个Celeborn Worker的磁盘IO延迟会飙升。进一步排查发现，该Worker节点上同时部署了另一个团队的日志收集服务，在下午定时进行日志压缩和上传，挤占了磁盘IO资源。通过协调资源调度，将Celeborn Worker节点独立出来，问题得以解决。这个故事告诉我们，即使在云上，对底层资源的“吵闹邻居”效应也要保持警惕。

6. 进阶思考：Celeborn与Spark 3.x的Dynamic Resource Allocation

EMR Serverless Spark本身就具备极致的弹性，而Spark自带的Dynamic Resource Allocation（DRA）功能也能在作业运行时动态调整Executor数量。Celeborn与DRA的结合，能产生更奇妙的化学反应。

在没有Celeborn时，启用DRA需要非常小心Shuffle数据。因为Executor可能在持有Shuffle数据时被移除，导致数据丢失。通常需要开启spark.shuffle.service.enabled（外部Shuffle服务），而该服务在Serverless环境中部署复杂。

有了Celeborn，情况大为改观：

• Celeborn本身就是一个更强大、更可靠的外部Shuffle服务。Executor不持有数据，因此可以被安全地移除。
• 你可以更激进地配置DRA参数，让Spark在Stage初期申请大量Executor快速处理数据，在Shuffle Write完成后立即释放多余资源，在Shuffle Read阶段再按需申请。Celeborn保证了数据在Executor释放后依然可用。
• 配置示例：

  spark.dynamicAllocation.enabled true spark.shuffle.manager org.apache.spark.shuffle.celeborn.RssShuffleManager # 可以设置较小的初始Executor数，让集群快速启动 spark.dynamicAllocation.initialExecutors 5 # 允许Executor在空闲较短时间内被移除，因为数据在Celeborn，不怕丢 spark.dynamicAllocation.executorIdleTimeout 60s

这种组合能进一步优化资源利用率，降低Serverless作业的整体成本。当然，这需要对作业的行为有深入了解，避免Executor频繁启停带来的开销反而抵消了收益。建议先在测试作业上验证不同DRA参数的效果。