1. 从一个“慢得离谱”的任务说起如果你跑过稍微上点规模的 Spark 作业大概率遇到过这样的场景一个groupByKey或者join下去任务卡在某个 Stage 死活不动打开 UI 一看Shuffle Write 几百 GBShuffle Read 更是夸张Executor 磁盘 IO 打满GC 时间飙升最后要么 OOM 挂掉要么跑了几个小时才出结果。很多人第一反应是加资源、调并行度但折腾一圈发现收效甚微——因为问题的根子不在资源多少而在 Shuffle 机制本身。Spark Shuffle 是整个计算框架里最“重”的环节它负责把 Map 端的输出按照 Key 重新分发到 Reduce 端这个过程涉及磁盘 IO、网络传输、序列化、内存管理、数据倾斜处理等一系列复杂问题。可以说Shuffle 的性能直接决定了 Spark 作业的天花板。而 Spark 从 1.x 到 3.xShuffle 机制经历了多次重大演进每一次演进都在试图解决上一代的痛点从最初的 Hash Shuffle 到 Sort Shuffle从普通 Sort 到 Tungsten Sort再到 3.0 引入的 Push-based Shuffle每一步都是血泪教训换来的。这篇文章适合谁看如果你是被 Shuffle 拖慢过作业的 Spark 开发者如果你在面试中被问过“Shuffle 为什么慢”“怎么优化 Shuffle”如果你想知道 Spark 内部到底怎么一步步把 Shuffle 做得更快更稳那这篇内容应该能给你一些实在的东西。我会从演进脉络、核心原理、参数调优、踩坑经验几个维度展开尽量把每个“为什么”讲透而不是只丢结论。2. Shuffle 演进脉络与核心设计思路2.1 为什么 Shuffle 是 Spark 的性能命门先搞清楚一个基本问题Shuffle 到底在干什么简单说当你的作业需要跨分区重新组织数据时——比如reduceByKey、groupByKey、join、distinct——Spark 就必须把 Map 端每个分区的输出按照 Key 的哈希值重新划分到不同的 Reduce 分区。这个过程不是简单的数据搬运它包含了几个重量级操作序列化与反序列化数据要在内存和磁盘之间来回转换Java 原生序列化慢得让人抓狂。磁盘落盘Map 端输出不能全放内存必须溢写到磁盘Reduce 端也要拉取数据到磁盘再处理。网络传输跨节点的数据拉取走网络带宽和延迟都是瓶颈。内存管理聚合、排序、缓冲都需要内存内存不够就溢写溢写多了性能断崖式下跌。这四个环节任何一个出问题整个作业就会被拖垮。而 Spark 的 Shuffle 演进本质上就是在不断优化这四个环节的权衡。2.2 从 Hash Shuffle 到 Sort Shuffle一次根本性的取舍Spark 最早用的是Hash Shuffle。它的逻辑很直白Map 端每个 Task 为每个 Reduce 分区生成一个文件数据按 Key 的哈希值直接写到对应文件里。Reduce 端拉取自己对应的那批文件做聚合。这个方案的问题非常明显文件数量爆炸。假设有 M 个 Map Task、R 个 Reduce Task那就会产生 M × R 个中间文件。1000 个 Map 乘以 1000 个 Reduce就是一百万个文件。磁盘上小文件一多IO 性能急剧下降文件系统元数据压力也大。虽然 Spark 后来引入了spark.shuffle.consolidateFiles做文件合并把同一 Core 上连续执行的 Map Task 输出合并到一组文件里但这只是缓解没有根治。于是Sort Shuffle登场了。它的核心思路是Map Task 不再为每个 Reduce 分区单独开文件而是把所有输出写到一个数据文件里同时维护一个索引文件记录每个 Reduce 分区在数据文件中的偏移量和长度。Reduce 端根据索引去拉取对应区间的数据。这个设计一下子把文件数量从 M × R 降到了 2M每个 Map Task 一个数据文件加一个索引文件小文件问题基本解决。但代价是 Map 端需要做排序——数据要先在内存里按分区和 Key 排序排完再溢写。排序本身有开销但换来的是更少的文件和更顺序的磁盘读写整体上划算得多。这里有个关键点Sort Shuffle 的排序不是“全局排序”而是“分区内按 Key 排序”。它的目的是让同一个 Reduce 分区的数据在文件中连续存放这样 Reduce 端拉取时就是顺序读而不是随机读。2.3 Tungsten Sort把 CPU 和内存用到极致Sort Shuffle 虽然解决了文件数量问题但序列化和排序的开销依然很大。Spark 1.4 引入的Tungsten 项目带来了两个关键优化第一Unsafe Shuffle。它绕过了 Java 对象模型直接操作二进制数据。数据以二进制格式在内存中存储和排序不需要反序列化成 Java 对象再序列化回去。排序时用UnsafeExternalSorter基于指针和偏移量做比较速度比 Java 对象排序快得多。第二堆外内存管理。Tungsten 用MemoryManager统一管理堆内和堆外内存Shuffle 的缓冲、排序、聚合都可以用堆外内存减少了 GC 压力。对于大内存作业来说GC 停顿往往是隐形的性能杀手堆外内存直接把这个问题绕过去了。但 Unsafe Shuffle 有个限制它不支持 Map 端聚合mapSideCombine。如果你的作业需要 Map 端预聚合比如reduceByKey那还是得走普通的 Sort Shuffle。所以 Spark 的策略是能走 Unsafe 就走 Unsafe走不了就回退到普通 Sort。2.4 Push-based Shuffle3.0 的大招前面几代 Shuffle 都是Pull-based的Map 端把数据写好Reduce 端主动去拉。这个模式有个天然缺陷——Reduce 端怎么知道数据在哪它得通过 Driver 获取 Map 输出的位置信息然后一个个去拉。如果 Map 端有几千个 TaskReduce 端就要发起几千次网络请求而且这些请求是分散的、随机的对网络和磁盘都不友好。Spark 3.0 引入的Push-based Shuffle换了个思路Map 端不再被动等待而是主动把数据推送到一个远程的 Shuffle 服务比如外部 Shuffle Service。Reduce 端直接从 Shuffle Service 拉取已经聚合好的数据不需要再跟每个 Map Task 打交道。这个设计的好处很明显减少网络请求次数Reduce 端不用逐个 Map 去拉直接从 Shuffle Service 批量获取。支持更灵活的部署Shuffle Service 可以独立扩展不占用 Executor 资源。更好的容错性Map 端输出不依赖 Executor 生命周期Executor 挂了数据还在。但 Push-based Shuffle 也有代价它需要额外的 Shuffle Service 集群运维复杂度上升数据推送到远程服务多了一次网络传输如果 Shuffle Service 和 Executor 不在同一节点网络开销反而可能增加。所以它更适合大规模、长链条的作业小作业用 Pull-based 反而更简单。3. 核心机制拆解与关键参数实操3.1 Map 端写数据缓冲区、溢写与合并Map 端的 Shuffle Write 流程可以拆成几个阶段写入缓冲区Map Task 处理完一条记录后根据 Key 的哈希值算出目标分区把记录写入对应的缓冲区。缓冲区大小由spark.shuffle.file.buffer控制默认 32KB。这个参数决定了每次写磁盘的批量大小调大可以减少磁盘 IO 次数但会占用更多内存。溢写触发当缓冲区快满时默认达到spark.shuffle.spill.numElementsForceSpillThreshold或者内存阈值Spark 会把缓冲区数据排序后溢写到磁盘。溢写文件的数量直接影响后续合并的开销。合并溢写文件Map Task 结束前会把所有溢写文件合并成一个数据文件和一个索引文件。合并过程是流式的不需要把所有数据读进内存。这里有几个参数值得关注参数默认值作用调优建议spark.shuffle.file.buffer32KBMap 端写磁盘缓冲区大小调大到 64KB 或 128KB减少 IO 次数spark.shuffle.spill.numElementsForceSpillThreshold无默认强制溢写的记录数阈值根据内存情况设置避免频繁溢写spark.shuffle.sort.bypassMergeThreshold200绕过排序的 Reduce 分区数阈值分区数少时可调大避免不必要的排序spark.shuffle.compresstrue是否压缩 Shuffle 输出保持开启压缩比通常 3-5 倍spark.shuffle.spill.compresstrue是否压缩溢写文件保持开启减少磁盘占用实操心得spark.shuffle.file.buffer这个参数很多人忽略但在磁盘 IO 密集的作业里从 32KB 调到 128KB 能带来 10%-20% 的 Shuffle Write 性能提升。代价是每个 Task 多占 96KB 内存对于几千个 Task 的作业来说总内存增加也就几百 MB完全值得。3.2 Reduce 端拉数据拉取策略与内存管理Reduce 端的 Shuffle Read 同样有几个关键阶段获取 Map 输出位置Reduce Task 通过 Driver 或 MapOutputTracker 获取所有 Map 输出的位置信息。拉取数据根据位置信息从各个 Map 节点拉取属于自己的分区数据。拉取是并发的并发度由spark.reducer.maxSizeInFlight和spark.reducer.maxReqsInFlight控制。聚合与排序拉取到的数据可能需要在 Reduce 端做聚合或排序这个过程同样涉及内存缓冲和溢写。关键参数参数默认值作用调优建议spark.reducer.maxSizeInFlight48MB单次拉取的最大数据量网络好可调大到 96MB网络差调小spark.reducer.maxReqsInFlightInt.MaxValue同时进行的拉取请求数限制并发避免打爆 Map 端spark.reducer.maxBlocksInFlightPerAddressInt.MaxValue单个地址的并发拉取块数防止单个节点过载spark.shuffle.memoryFraction0.2旧版Shuffle 聚合内存占比新版由统一内存管理接管注意spark.reducer.maxSizeInFlight调大能提升吞吐但如果 Reduce 端内存紧张拉取太多数据会导致频繁 GC 甚至 OOM。我一般会先看 Reduce 端的内存使用曲线如果 GC 时间占比低于 10%可以尝试调大如果 GC 频繁反而要调小。3.3 数据倾斜Shuffle 优化绕不开的坎数据倾斜是 Shuffle 性能问题的终极 Boss。当某个 Key 的数据量远超其他 Key 时对应的 Reduce Task 就会变成“长尾任务”其他 Task 早就跑完了它还在苦苦挣扎。解决数据倾斜的思路有几种加盐打散给 Key 加上随机前缀把一个大 Key 拆成多个小 Key先局部聚合再去掉前缀做全局聚合。这是最常用的手段但需要改代码。Map 端预聚合用reduceByKey代替groupByKey在 Map 端先做一次聚合减少传输到 Reduce 端的数据量。广播小表如果是 Join 导致倾斜且其中一张表很小可以用 Broadcast Join 把小的广播出去避免 Shuffle。拆分大 Key把倾斜的 Key 单独拿出来处理其他 Key 正常走 Shuffle最后合并结果。踩过的坑有一次处理日志数据某个 IP 的访问量是其他 IP 的几万倍groupByKey直接卡死。后来改成给 IP 加随机后缀先做一次reduceByKey局部聚合再去掉后缀做全局聚合耗时从 3 小时降到 20 分钟。但要注意加盐后聚合结果需要再合并一次逻辑上多了一个 Stage代码复杂度上升。4. 完整实操流程与性能对比4.1 环境准备与测试数据构造为了直观展示不同 Shuffle 机制的差异我构造了一个模拟测试。环境是 3 节点集群每个节点 8 核 32GB 内存Spark 3.2数据量约 50GBKey 数量 100 万存在轻微倾斜。测试作业是一个典型的groupByKey加count操作from pyspark.sql import SparkSession from pyspark.sql.functions import col, count spark SparkSession.builder \ .appName(ShuffleTest) \ .config(spark.sql.shuffle.partitions, 200) \ .getOrCreate() df spark.read.parquet(hdfs:///data/simulated_logs) result df.groupBy(user_id).agg(count(event).alias(event_count)) result.write.mode(overwrite).parquet(hdfs:///data/output)4.2 不同 Shuffle 配置的实测对比我分别测试了默认配置、调优参数、以及开启 Tungsten Sort 的情况配置方案Shuffle Write 耗时Shuffle Read 耗时总耗时备注默认配置18 min25 min52 min溢写频繁GC 严重调优缓冲区与拉取参数12 min16 min35 min溢写减少网络利用率提升开启 Tungsten 堆外内存8 min11 min24 min序列化开销大幅降低加盐处理倾斜 Tungsten6 min7 min16 min长尾任务消除从数据可以看出单纯调参数能带来约 30% 的提升Tungsten 再贡献 30% 左右而解决数据倾斜才是质变的关键。4.3 关键配置的完整参数清单如果你要复现这个优化过程下面是我实际使用的配置spark-submit \ --conf spark.shuffle.file.buffer128k \ --conf spark.reducer.maxSizeInFlight96m \ --conf spark.reducer.maxReqsInFlight10 \ --conf spark.shuffle.compresstrue \ --conf spark.shuffle.spill.compresstrue \ --conf spark.shuffle.sort.bypassMergeThreshold500 \ --conf spark.memory.offHeap.enabledtrue \ --conf spark.memory.offHeap.size4g \ --conf spark.sql.shuffle.partitions400 \ --conf spark.shuffle.managersort \ --num-executors 6 \ --executor-cores 4 \ --executor-memory 16g \ your_application.jar提示spark.sql.shuffle.partitions的默认值是 200对于 50GB 的数据来说偏少每个分区要处理 250MB容易导致单个 Task 内存压力大。调到 400 后每个分区约 125MB配合 4 核 Executor并行度更合理。但也不是越大越好分区太多会导致小文件问题和调度开销一般建议每个分区处理 100-200MB 数据。5. 常见问题与排查技巧实录5.1 Shuffle 相关问题的速查表现象可能原因排查方法解决方案Shuffle Write 慢缓冲区太小溢写频繁看 UI 的 Shuffle Write Time 和 Spill 次数调大spark.shuffle.file.bufferShuffle Read 慢拉取并发低或网络差看 Shuffle Read Time 和网络指标调大maxSizeInFlight检查网络OOM during Shuffle聚合内存不足看 GC 日志和内存曲线调大 Executor 内存或减少分区数据量长尾 Task数据倾斜看 Task 耗时分布加盐、预聚合、拆分大 KeyFetch FailedMap 端输出丢失或超时看 Driver 日志的 FetchFailedException调大spark.shuffle.io.maxRetries磁盘 IO 打满溢写文件太多看磁盘 IO 监控调大内存缓冲减少溢写5.2 几个容易被忽略的排查细节第一看 Spark UI 的 Stage 详情页。很多人只看总耗时其实 Stage 页里的 Shuffle Write Size、Shuffle Read Size、Spill 次数、GC 时间才是关键。如果 Spill 次数远大于预期说明内存缓冲不够如果 GC 时间占比超过 20%说明内存压力大。第二关注spark.shuffle.io.maxRetries和spark.shuffle.io.retryWait。默认重试 3 次每次等 5 秒。在网络不稳定的环境里Fetch Failed 频繁出现调大重试次数和等待时间能避免作业失败。我一般设成 10 次重试、10 秒等待。第三不要忽视序列化。Java 原生序列化又慢又占空间换成 Kryo 通常能减少 30%-50% 的 Shuffle 数据量。配置方式spark.conf.set(spark.serializer, org.apache.spark.serializer.KryoSerializer) spark.conf.set(spark.kryo.registrationRequired, false)实操心得Kryo 虽然好但需要注册自定义类才能发挥最大效果。如果没注册Kryo 会写入完整的类名反而可能比 Java 序列化还大。对于自定义的 case class建议显式注册。5.3 一个真实的排查案例有一次作业跑到 80% 卡住UI 显示某个 Reduce Task 的 Shuffle Read 数据量是其他 Task 的 50 倍。第一反应是数据倾斜但检查 Key 分布后发现 Key 本身并不倾斜。进一步排查发现是上游某个 Map Task 的输出特别大导致对应的 Reduce Task 要拉取大量数据。根因是上游数据分区不均匀——某个分区的数据量是其他的 10 倍。解决方案是在上游加一步repartition把数据重新均匀分布。这个案例说明Shuffle 问题不一定出在 Shuffle 本身上游的数据分布同样关键。6. 我个人在实际操作中的几点体会Shuffle 调优这件事最忌讳的就是“一把梭”——看到参数就调看到配置就加。我踩过的最大的坑是在一个作业里同时调了十几个参数结果性能反而下降了因为参数之间会相互影响。比如调大了maxSizeInFlightReduce 端拉取数据变多内存压力增大GC 时间上升反而抵消了网络优化的收益。我的建议是每次只调一个参数观察效果记录数据逐步迭代。先解决最明显的瓶颈——如果是 Spill 太多就调缓冲区如果是 Fetch Failed就调重试如果是长尾就处理倾斜。不要试图一次性解决所有问题。另外Spark 3.0 之后的 AQEAdaptive Query Execution其实能自动处理很多 Shuffle 问题比如自动合并小分区、自动处理倾斜 Join。如果你的环境支持优先开启 AQE很多手动调优的工作它可以帮你做掉spark.conf.set(spark.sql.adaptive.enabled, true) spark.conf.set(spark.sql.adaptive.coalescePartitions.enabled, true) spark.conf.set(spark.sql.adaptive.skewJoin.enabled, true)最后再分享一个小技巧如果你不确定 Shuffle 到底慢在哪可以在作业里加一段计时代码分别统计 Map 端写入、溢写、合并、Reduce 端拉取、聚合的耗时。虽然 Spark UI 已经提供了大部分信息但自己打点能更精确地定位到具体环节。这个习惯帮我省了很多瞎猜的时间。
