Spark Shuffle 服务：外置 Shuffle 为什么比内置的更稳

一、一个凌晨三点的故障

先讲故事。上个月凌晨三点，我们的大盘看板突然挂了。查了半天发现是 Spark 任务跑到一个 Join 算子时，4 个 Executor 先后 OOM 退出。没等 YARN 重新分配好资源，整个 Stage 就失败了。日志里全是 FetchFailedException: Failed to connect to xxx。

根因就一句话：那台节点挂了，上面 Executor 存的 Shuffle 中间数据全丢了。

这引出一个灵魂拷问：Shuffle 数据到底该放哪儿？ Spark 社区给出的答案就是 External Shuffle Service（外部 Shuffle 服务，简称 ESS）。

flowchart TB
    subgraph 内置Shuffle["内置 Shuffle（Executor管理）"]
        A1[Task 写完 Shuffle 文件] --> A2[文件存在 Executor 本地]
        A3[Executor 因 OOM/GC/NodeLost 挂掉] -.-> A4["Shuffle 数据全丢<br/>Stage 重算!"]
    end
    
    subgraph 外置Shuffle["外置 Shuffle Service（ESS）"]
        B1[Task 写完 Shuffle 文件] --> B2[文件注册到 ESS 守护进程]
        B3[Executor 挂掉] -.-> B4["ESS 继续提供数据<br/>仅重算丢失的 Task 分区"]
    end
    
    style A4 fill:#E74C3C,color:#fff
    style B4 fill:#27AE60,color:#fff

二、Shuffle 到底慢在哪儿

理解 ESS 之前，得先搞清楚 Shuffle 的本质。

Shuffle 是一个让分布式计算头疼了二十年的操作。它的过程比你想象的要复杂得多：

1. Map 端：每个 Task 把数据按 key 分区，写入本地磁盘。注意，这里每个 Task 要产生 N 个文件（N = 下游 Reduce Task 数量）。所以如果有 100 个 Map Task 和 50 个 Reduce Task，就会产生 5000 个小文件。

2. Reduce 端：每个 Task 从所有 Map Task 的对应分区拉取数据。这就涉及大量的网络连接。

3. 合并排序：拉过来的数据要在内存中排序或聚合。

可以看出 Shuffle 的痛点非常集中：磁盘 I/O、网络带宽、小文件数量。这三个随便哪个拉胯，整个 Stage 就卡住了。

内置 Shuffle 模式下，数据生命周期和 Executor 绑定。Executor 一挂，所有本地 Shuffle 数据都消失，只能重算整个 Stage。这就是开头那个故障的根因。

三、ESS 的架构与工作原理

External Shuffle Service 的核心设计理念就一句话：让计算和管理解耦。

看看 ESS 的架构怎么运转的：

# 伪代码：ESS 的注册与读取流程
# 这不是可运行的代码，而是帮你理解 ESS 做了什么

class ExternalShuffleService:
    """外部 Shuffle 服务的核心逻辑示意"""
    
    def __init__(self):
        # 维护 Executor 到 Shuffle 文件的映射关系
        # 即使 Executor 挂了，映射还在
        self.registered_executors = {}  # executor_id -> {block_map}
    
    def register_executor(self, executor_id: str, app_id: str):
        """Executor 启动时向 ESS 注册
        
        这一步是 ESS 和内置模式的最大区别：
        内置模式：Shuffle数据 = Executor的私有财产
        外置模式：Shuffle数据 = 托管给ESS的公共资产
        """
        self.registered_executors[executor_id] = {
            "app_id": app_id,
            "blocks": {},  # block_id -> 本地文件路径
            "status": "alive"
        }
    
    def report_blocks(self, executor_id: str, block_locations: dict):
        """Task 写完 Shuffle 数据后，把文件路径告诉 ESS
        
        比如 Task 1 写完了 partition_0 的数据，路径是 /tmp/spark/shuffle_1_0.data
        把这个信息注册到 ESS，其他节点才能找到
        """
        self.registered_executors[executor_id]["blocks"].update(block_locations)
    
    def fetch_shuffle_block(self, shuffle_id: int, partition_id: int) -> bytes:
        """下游 Task 通过 ESS 读取 Shuffle 数据
        
        即使源 Executor 挂了，ESS 作为独立进程仍然可以读取本地文件
        """
        # ESS 遍历所有注册的 Executor，查找目标 partition 的数据
        for executor_id, info in self.registered_executors.items():
            block_key = f"shuffle_{shuffle_id}_{partition_id}"
            if block_key in info["blocks"]:
                file_path = info["blocks"][block_key]
                # ESS 直接读取本地磁盘文件，不经过已挂的 Executor
                with open(file_path, 'rb') as f:
                    return f.read()
        raise Exception(f"Shuffle block {shuffle_id}_{partition_id} not found")

ESS 带来的效果非常直观：

对比维度	内置 Shuffle	ESS 外置 Shuffle
Executor 退出	数据丢失，重算整个 Stage	数据保留，仅重算退出的 Task
动态资源分配	不支持（释放Executor会丢数据）	支持（Executor可以安全释放）
进程稳定性	Executor和Shuffle同进程，互相影响	ESS独立守护进程，隔离故障
磁盘管理	Executor 自己管	ESS 统一管理，可清理

动态资源分配是 ESS 带来的一个意外收获。没有 ESS 时，Spark 不敢随便释放 Executor，因为一旦释放，上面的 Shuffle 数据就没了。有了 ESS，Executor 用完就可以释放，云原生环境下能省不少钱。

为什么 ESS 作为一个"独立守护进程"就能解决所有问题？ 答案在于进程生命周期的独立性。内置模式下，Shuffle 数据和 Executor 共享同一个 JVM 进程——OOM 时杀的不是 Executor 的 Java 线程，而是整个 JVM 进程，磁盘上所有本地文件跟着变得不可达。ESS 把 Shuffle 数据的读写路径从 Executor 进程剥离到独立的 NodeManager 守护进程中（端口 7337），这个独立进程不跑 Task、不受用户 Spark 代码的 GC 和 OOM 影响，即使 4 个 Executor 全挂了，ESS 仍然可以读取本地磁盘上的 Shuffle 文件、通过网络发给下游 Task。本质上，ESS 做的事情跟 Redis 做缓存、HDFS 做存储一样——把"不稳定计算"和"稳态数据"的耦合解开，让故障的爆炸半径限定在计算层，不蔓延到数据层。

四、配置与调优实战

ESS 的配置不算复杂，但有几个参数值得认真调：

# spark-defaults.conf — ESS 关键配置项

# 1. 开启 External Shuffle Service（默认是关闭的）
# 为什么默认关闭？因为ESS需要单独部署守护进程，
# 对简单的批处理任务来说，额外运维成本不划算
spark.shuffle.service.enabled true

# 2. ESS 的端口（所有 NodeManager 需要统一）
spark.shuffle.service.port 7337

# 3. ESS 的磁盘目录 — 这是最容易踩的坑
# 默认在 /tmp 下，系统重启就没了！
# 生产环境必须指向大容量数据盘
spark.local.dir /data1/spark,/data2/spark,/data3/spark

# 4. Shuffle 文件保留超时时间（毫秒）
# 为什么要有超时？防止僵尸Executor的Shuffle文件永远不清理
# 默认120s，如果ETL任务量大可以适当增加
spark.shuffle.service.removeShuffle.timeout 600

# 5. ESS 的 IO 线程数
# 如果你的节点上有大量Executor同时读写，增加这个值
spark.shuffle.io.serverThreads 16

再补充两个实践中的细节：

关于磁盘监控。 ESS 会在 spark.local.dir 下不断写入 Shuffle 文件，如果不监控磁盘使用率，迟早会把数据盘写满。我们的做法是配置一个 cron 脚本，每 30 分钟检查一次，超过 80% 的节点自动发告警。

关于 ESS 的优雅停止。 如果你要重启 ESS 守护进程，直接用 kill -9 会导致正在运行的 Spark 任务失败。正确的做法是：

# 优雅重启 ESS 的正确姿势
# 1. 先把 ESS 标记为 draining（不接受新注册）
# 2. 等待当前所有 Shuffle 连接自然结束
# 3. 再执行重启
yarn nodemanager stop  # ESS 随 NodeManager 一起管理

为什么 spark.local.dir 必须指向大容量数据盘而非系统盘？ 这个问题如果不认真对待，某一天你会发现整个集群的根分区被写满了。Shuffle 文件的总量 = Map Task 数量 × Reduce 分区数 × 平均分区大小。假设一个 Stage 有 200 个 Map Task、100 个 Reduce 分区，每个分区 10MB，总 Shuffle 数据量 = 200 × 100 × 10MB = 200GB / 节点。这些文件全部写入 spark.local.dir，如果指向 /tmp（操作系统分区，通常只有 50-100GB），磁盘很快被塞满——后果不只是 Shuffle 失败，而是整个节点的操作系统因为磁盘满而失联（sshd 写不了日志、sudo 执行不了命令）。所以 spark.local.dir 必须配置多块数据盘（/data1/spark,/data2/spark,/data3/spark），让 ESS 做磁盘级别的负载均衡。

🚨 踩坑提醒

1. ESS 端口（7337）多集群共享时的冲突 — 如果一个物理节点上跑了两个 Hadoop 集群（比如生产+测试），两个 ESS 抢同一个端口，后启动的会绑定失败。必须给不同集群分配不同端口：spark.shuffle.service.port 7337（生产）、7338（测试），并在防火墙规则里区分。
2. spark.shuffle.service.removeShuffle.timeout 设太短导致任务失败 — 这个超时是 ESS 清理 Shuffle 文件的 TTL。如果你的 ETL 任务有 10 分钟的 stage 间隔（上一个 stage 结束到下一个 stage 开始拉 Shuffle 数据之间空等 10 分钟），而 timeout 只有 120 秒，ESS 会在你还没拉数据之前就把文件删了——下一个 stage 会收到 FetchFailedException。正确做法：设成任务最长 stage 间隔的 2 倍。
3. K8s 环境不要用 ESS——那是给 YARN 设计的 — 如果你在 Kubernetes 上跑 Spark，ESS 无法部署（K8s 的 Node 上没有 NodeManager 守护进程）。Spark 3.2+ 提供了 push-based shuffle，效果跟 ESS 类似但架构不同：Map Task 写完后主动把数据 push 给远程 Shuffle 服务，而不是等 Reduce Task 来拉。

五、总结

ESS 解决的问题其实很简单——把 Shuffle 数据的生命周期从 Executor 的铁链上解下来。但就是这一个变化，让你的 Spark 集群稳定性提升了一个数量级。

划几个重点：

1. ESS 是生产环境标配，除非你的 Spark 任务从不因 Executor 挂掉而重算——但这不现实。
2. 最大的收益是动态资源分配，云原生环境能省不少钱。
3. 磁盘管理是关键，别把 Shuffle 目录放在系统盘，记得加监控和清理策略。
4. 如果你在用 Kubernetes 运行 Spark，ESS 不适用——K8s 环境用 Spark 3.2+ 的 push-based shuffle 更合适。

一句话总结：ESS 就是给 Shuffle 数据买了个独立保险，Executor 可以挂，数据不能丢。

---

我是朱大喜，Spark 调优这条路上踩过的坑比工资条还长。你的 Shuffle 也翻过车吗？评论区聊聊？