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[内置 ShuffleExecutor管理] A1[Task 写完 Shuffle 文件] -- A2[文件存在 Executor 本地] A3[Executor 因 OOM/GC/NodeLost 挂掉] -.- A4[Shuffle 数据全丢br/Stage 重算!] end subgraph 外置Shuffle[外置 Shuffle ServiceESS] 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 是一个让分布式计算头疼了二十年的操作。它的过程比你想象的要复杂得多Map 端每个 Task 把数据按 key 分区写入本地磁盘。注意这里每个 Task 要产生 N 个文件N 下游 Reduce Task 数量。所以如果有 100 个 Map Task 和 50 个 Reduce Task就会产生 5000 个小文件。Reduce 端每个 Task 从所有 Map Task 的对应分区拉取数据。这就涉及大量的网络连接。合并排序拉过来的数据要在内存中排序或聚合。可以看出 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 fshuffle_{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(fShuffle block {shuffle_id}_{partition_id} not found)ESS 带来的效果非常直观对比维度内置 ShuffleESS 外置 ShuffleExecutor 退出数据丢失重算整个 Stage数据保留仅重算退出的 Task动态资源分配不支持释放Executor会丢数据支持Executor可以安全释放进程稳定性Executor和Shuffle同进程互相影响ESS独立守护进程隔离故障磁盘管理Executor 自己管ESS 统一管理可清理动态资源分配是 ESS 带来的一个意外收获。没有 ESS 时Spark 不敢随便释放 Executor因为一旦释放上面的 Shuffle 数据就没了。有了 ESSExecutor 用完就可以释放云原生环境下能省不少钱。为什么 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 做磁盘级别的负载均衡。 踩坑提醒ESS 端口7337多集群共享时的冲突— 如果一个物理节点上跑了两个 Hadoop 集群比如生产测试两个 ESS 抢同一个端口后启动的会绑定失败。必须给不同集群分配不同端口spark.shuffle.service.port 7337生产、7338测试并在防火墙规则里区分。spark.shuffle.service.removeShuffle.timeout设太短导致任务失败— 这个超时是 ESS 清理 Shuffle 文件的 TTL。如果你的 ETL 任务有 10 分钟的 stage 间隔上一个 stage 结束到下一个 stage 开始拉 Shuffle 数据之间空等 10 分钟而 timeout 只有 120 秒ESS 会在你还没拉数据之前就把文件删了——下一个 stage 会收到FetchFailedException。正确做法设成任务最长 stage 间隔的 2 倍。K8s 环境不要用 ESS——那是给 YARN 设计的— 如果你在 Kubernetes 上跑 SparkESS 无法部署K8s 的 Node 上没有 NodeManager 守护进程。Spark 3.2 提供了 push-based shuffle效果跟 ESS 类似但架构不同Map Task 写完后主动把数据 push 给远程 Shuffle 服务而不是等 Reduce Task 来拉。五、总结ESS 解决的问题其实很简单——把 Shuffle 数据的生命周期从 Executor 的铁链上解下来。但就是这一个变化让你的 Spark 集群稳定性提升了一个数量级。划几个重点ESS 是生产环境标配除非你的 Spark 任务从不因 Executor 挂掉而重算——但这不现实。最大的收益是动态资源分配云原生环境能省不少钱。磁盘管理是关键别把 Shuffle 目录放在系统盘记得加监控和清理策略。如果你在用 Kubernetes 运行 SparkESS 不适用——K8s 环境用 Spark 3.2 的 push-based shuffle 更合适。一句话总结ESS 就是给 Shuffle 数据买了个独立保险Executor 可以挂数据不能丢。我是朱大喜Spark 调优这条路上踩过的坑比工资条还长。你的 Shuffle 也翻过车吗评论区聊聊