Spark shuffle 和 MapReduce shuffle 的区别在 Spark 和 MapReduce 之间做技术选型或排查作业性能时shuffle 永远是绕不开的那道坎。很多刚接触分布式计算的人会把 Spark shuffle 和 MapReduce shuffle 混在一起以为都是“map 端排序、reduce 端归并、中间落盘”那一套。就我自己观察这个理解只对了一半两者要解决的核心问题确实一样但在数据组织、内存利用、容错和调优手段上设计思路几乎完全不同。这篇文章就从机制层面把两者彻底拆开从完整链路、排序逻辑、文件组织、内存取舍到参数调优和实战排坑尽量讲清楚它们的分岔点在哪以及面试或实际调优时怎么区分它们。无论你是刚入行的大数据开发还是正在把 Hive 作业从 MapReduce 迁到 Spark SQL这篇文章都值得花十分钟认真看完。1. 为什么这两个 shuffle 总被放在一起比较1.1 它们解决的是同一个问题shuffle 这个词在 MapReduce 里指 map 端输出到 reduce 端输入之前的整段数据搬运过程在 Spark 里指 stage 与 stage 之间的数据重排过程。它们本质上都在做同一件事把上游任务产生的数据按照 key 的某种分区逻辑重新分发给下游任务。举个最直观的例子一个按 customer_id 分组的聚合任务两个 map 任务分别处理了一半的数据某个 customer_id 的记录可能同时出现在两个 map 里。reduce/下游任务必须保证同一个 customer_id 的所有记录都落到同一个 reducer/executor 上否则 group by 就没法做。这个按 key 划分归属的动作就是 shuffle 的核心。但 MapReduce 和 Spark 在这个核心动作之外做了完全不同的加法。MapReduce 把 shuffle 做成了显式的、几乎强制排序的流程reduce 端拿到的是已经按 key 排序好的连续块Spark 则把 shuffle 当成 stage 边界上的数据交换它关心的是哪个分区属于哪个下游任务而不是key 是否全局有序。1.2 设计目标的岔路从磁盘归并到内存迭代MapReduce 诞生年代面对的是海量数据、廉价磁盘和不可靠节点设计者优先考虑的是稳定性和可恢复性。所以它的中间结果倾向于落盘reduce 拿到数据之前map 端已经完成了一次完整的排序和归并reduce 端再做一次合并和排序。每一步都在为即使某个节点挂了也能重新拉取做准备。Spark 的设计目标则更多面向迭代计算和交互式分析。它希望把中间结果尽量留在内存里减少磁盘和网络往返。Spark 早期也真的用过多份小文件形式的 hash shuffle后面发现自己按 partition 连续写入、配索引文件的方式更高效于是演进到我们现在常用的 sort-based shuffle。这里我要专门强调一点Spark 的 shuffle 并不是简单地把 MapReduce 的 shuffle 搬到内存里而是重新设计了输出文件格式和拉取模型。如果只看网上那些Spark 快是因为数据在内存的段子你很难解释为什么 Spark 在大 shuffle 场景下同样会出现磁盘溢写和 OOM。2. 两种 shuffle 的完整链路与关键节点2.1 MapReduce shuffle 是如何一步步把数据送到 reducer 的经典 MapReduce 的 shuffle 分为 map 端和 reduce 端两大部分。先看 map 端。MapTask 每输出一条记录会先写入一个环形缓冲区。这个缓冲区默认大小是 100MB由mapreduce.task.io.sort.mb控制。写入过程中已经按 key 做了分区分区器决定这条记录最终去哪个 reducer。当缓冲区使用率达到阈值默认是 80%后台线程就会开始 spill。spill 时会将缓冲区里的数据按分区和 key 进行一次快速排序然后写入一个小文件。如果有 combiner且 spill 文件数量达到一定条件还会在写盘前做一次本地聚合。combiner 本质上是 map 端的 mini-reducer它能把同一个 key 在 map 端先合并一次减少落到磁盘和网络的数据量但前提是聚合操作满足交换律和结合律。map 任务结束后可能会留下多个 spill 文件。此时 map 端会把这些文件按分区合并成一个最终输出文件。这个 merge 过程不是简单拼接而是要对同一个分区内的数据进行归并排序保证发给同一个 reducer 的数据key 是连续且有序的。对 reducer 来说每个 map 任务都会生成一个对应的输出文件如果 job 有 1000 个 map那 reduce 端要处理的就是最多 1000 个 map 输出分片。reduce 端的第一步是 fetch。Reducer 会启动多个线程并行去各个 map 输出节点拉取属于自己的分区数据默认并行拷贝数是 5可以通过mapreduce.reduce.shuffle.parallelcopies调整。拉来的数据先放内存内存放不下就落盘。reduce 端会持续做内存内和磁盘间的合并一边拉一边合并直到把所有 map 输出中属于本 reducer 的分区数据合并成一个大集合。这里的关键是归并排序。reduce 端拿到的是多个有序片段它使用一个小顶堆不停取最小 key把多个有序流合并为一个整体有序流。最后交给 reduce 函数时相同 key 的记录已经排在一起reducer 可以非常自然地把连续出现的相同 key 聚合为一组。这种先排序、再分组的方式让 reduce 逻辑变得简单而确定。2.2 Spark shuffle 从 ShuffleMapTask 到下游拉取Spark 进入 2.x 之后shuffle 实现统一为 sort-based shuffle早期纷繁复杂的 hash shuffle 和 Tungsten-Sort 基本退出了历史舞台。每次 shuffle 都发生在一个宽依赖所对应的 stage 边界上上游 stage 的最后一个 RDD 分区由 ShuffleMapTask 处理它的输出被分成多个 partition分别对应下游 stage 的各个任务。ShuffleMapTask 写数据的过程和 MapReduce 有不少相似之处但输出文件组织方式完全不同。SortShuffleWriter 有两种路径当分区数小于spark.shuffle.sort.bypassMergeThreshold默认 200 时走 bypass 路径不做 key 排序只按 partitionId 把数据直接写进多个临时文件然后合并成一个数据文件分区数较多时走普通排序路径把记录按分区 key排序写入一段一段的数据块最后合并成一个 data 文件同时生成一个 index 文件。很多人一看到 Spark 也排序就以为它和 MapReduce 是一样的。其实不一样。Spark 排序的主要目的不是让 reducer 拿到全局有序数据而是让合并多个文件时可以高效地顺序读。每个 ShuffleMapTask 最终只产出一个 data 文件和 index 文件下游任务读取时可以通过 index 文件里的 offset 快速定位属于自己的 block 范围不需要像 MapReduce 那样把一个 partition 的数据散落在 N 个 map 输出文件里再跨文件归并。下游 stage 的任务通过 ShuffleBlockFetcherIterator 去上游节点拉数据。拉取过程也可以配置并行度、重试次数和超时时间。拉回来的数据如果要聚合会用 AppendOnlyMap 优先在内存中聚合内存不够时才会溢写排序。所以 Spark 的 reduce 端聚合并不强制排序因为聚合操作依赖的是 hash map 或者内存中的状态不是key 连续出现这个前提。2.3 一张表对比核心步骤对比维度MapReduce shuffleSpark shuffle任务边界map 任务和 reduce 任务之间上游 stage 和下游 stage 之间map 端排序强制按分区和 key 排序可按分区排序key 排序不是必须输出文件每个 map 给每个 reducer 一个分片文件多每个 ShuffleMapTask 一个 data 文件加一个 index 文件shuffle 数据组织reduce 端跨多个 map 输出归并下游按 block 范围连续读reduce 端分组依赖整体有序相同 key 连续出现内存聚合为主溢写时才走排序中间结果落盘几乎默认落盘优先内存内存不足才溢写容错恢复失败任务重跑shuffle 数据重新生成如果 shuffle 文件丢失需要重算上游 stage调优侧重点排序缓冲、spill 阈值、拷贝线程数分区数、内存聚合、拉取并行度、AQE这张表可以应付大多数概念题。但真正到线上还需要理解每一行背后的原因接下来逐项展开。3. 深层差异机制、内存与容错3.1 排序到底是不是必须的MapReduce 的 reducer 在执行reduce(key, values)时默认是逐个 key 把 value 列表喂进来。框架必须保证同一个 key 的所有值在 reducer 端连续出现而要做到这一点最稳妥的方式就是整体排序。所以 MapReduce 的 sort 不是可选项它直接支撑了分组语义。Spark 则不同。Spark 的 RDD 算子像groupByKey、reduceByKey在下游拿到的是一个分区迭代器不需要依靠排序来保证相同 key 连续。它可以在内存里做一个哈希表把相同 key 直接合并数据量大到内存放不下才用外部排序辅助溢写。因此Spark 的普通排序路径只是为了让同一分区的数据在物理文件中连续存储便于 merge 和 index 定位并不是为了满足分组逻辑。理解这一点对调优很有用。MapReduce 里你几乎不可能跳过排序只能增大缓冲区、减少溢写次数而 Spark 里你可以通过 map 端聚合、合理分区、甚至repartition来避免部分排序开销。Spark 3.x 的 adaptive query execution 还能在运行过程中动态合并小分区这些都是 MapReduce 时代没有的灵活性。3.2 数据输出与文件管理方式很多老工程师刚接触 Spark 时会觉得奇怪为什么 shuffle write 文件那么少MapReduce 一个 map 任务要为每个 reducer 生成一个 shuffle 分片假设 map 数是 1000、reduce 数是 100光 map 端输出分片就是 10 万个reduce 端每个任务都要去拉取 1000 个小文件文件系统的 open 和网络连接压力很大。Spark 的 sort-based shuffle 在这方面是碾压级的优势。每个 ShuffleMapTask 无论下游有多少个任务最终都只写一个 data 文件和一个 index 文件。下游任务需要某个 partition 的数据时用 index 文件里的 start offset 和 end offset 去定位 data 文件对应区间。这样做的好处有三点文件数量大幅减少避免大量小文件下游读取时可以按 offset 快速跳过不关心的数据shuffle service 在提供服务时也更轻量。不过也要公正地说一句文件少不代表网络传输少。shuffle 数据量本身不变网络链路压力还在只是元数据管理和文件寻址开销大幅降低。Spark 的 shuffle read 如果遇到数据倾斜某个 Executor 还是会被单个超大数据块压垮。文件组织方式的差异解决的是分布问题不是数据量问题。3.3 内存与磁盘的取舍逻辑MapReduce 的 map 端在 shuffle 阶段能用的内存主要就是环形缓冲区。到了 Spark 里Executor 内存被分成 reserved、storage 和 execution 三大块shuffle 可以使用 execution 内存。这个设计让 Spark 的性能提升非常明显如果内存充足Map端聚合和 Reduce 端聚合都可以在内存里完成根本不碰磁盘。但这里有一个容易被误解的点Spark 并不是一定会比 MapReduce 快。如果 shuffle 数据量非常大而 Executor 内存又很紧张Spark 一样会把内存中的聚合结果 spill 到磁盘甚至因为频繁的 GC 导致性能恶化。我见过不少同学以为 Spark 是万能加速器把 Hive 的 MapReduce 任务换成 Spark SQL 后反而跑得更慢最后看到的就是大面积的磁盘溢写。所以更准确的说法是MapReduce 默认按磁盘优先设计shuffle 的每个环节都在为极端稳定做备份Spark 默认按内存优先设计shuffle 的性能好坏在很大程度上取决于你给它的内存和合理的并行度。调优时不要只盯着框架宣传的内存优势还要看你集群的资源配比。3.4 容错与重算逻辑MapReduce 的容错相对简单直接map 或 reduce 失败就把对应任务重新调度执行shuffle 数据会随任务重跑重新生成。由于中间结果基本在磁盘上task 失败后其他任务不太受牵连代价是多跑一部分重复计算。Spark 的容错依赖 lineage 和 shuffle 文件的持久化shuffle write 生成的 data/index 文件会被上游 stage 引用如果下游 stage 要重算优先复用已有的 shuffle 文件。但如果某个 Executor 崩溃它上面保存的 shuffle 文件就丢失了Spark 会把对应的上游 stage 分区重新计算。这种机制比 MapReduce 更精细但也更依赖 shuffle service 的介入。实际生产环境里很多 Spark 集群会开启 external shuffle service让 shuffle 文件的提供独立于 Executor 生命周期。Executor 挂掉后其他节点上的 shuffle 文件仍然可以被拉取减少上游重算的开销。这一点在长尾 Spark Streaming 或混合负载场景下尤其重要。4. 参数调优与配置对比4.1 MapReduce 调优先看这几个参数先说清楚版本不同参数名可能带mapreduce.或mapred.前缀生产上可以以你的实际 Hadoop 版本文档为准。mapreduce.task.io.sort.mbmap 端排序缓冲区大小默认 100MB。调大它可以减少 spill 次数但要小心 JVM 内存和 GC。mapreduce.map.sort.spill.percentspill 触发阈值默认 0.8。阈值越大缓冲区利用率越高但 spill 时排序耗时也更集中。mapreduce.map.output.compressmap 输出是否压缩。开启后能减少网络传输但会增加 CPU。mapreduce.reduce.shuffle.parallelcopiesreduce 端并行拷贝 map 输出的线程数默认 5。增加它可以让 fetch 更快但会抬高节点连接压力。mapreduce.reduce.shuffle.input.buffer.percentreduce shuffle 阶段可用内存比例。太小会导致频繁落盘太大会挤压 reduce 函数本身的内存。mapreduce.reduce.merge.inmem.threshold内存中的 shuffle segment 达到多少条开始合并默认 1000。调这些参数时最核心的思路是减少 spill 文件数和合并轮数。MapReduce 的 shuffle 就像在流水线上反复打包拆包一次 spill 就是一次排序和写盘一次 merge 就是一次多路归并。盲目把缓冲区调得很大可能让单次排序耗时失控盲目调大拷贝线程数也可能把节点网络打死。4.2 Spark shuffle 参数并没有想象中那么多Spark 的 shuffle 参数在 2.x 后收敛了很多真正值得动的其实就几个spark.shuffle.file.bufferShuffleMapTask 写 data 文件时的缓冲区大小默认 32k。增大它可以减少写文件的磁盘 IO 次数。spark.reducer.maxSizeInFlight每个 reducer 同时拉取的数据上限默认 48MB。调大可以增加吞吐但可能造成内存压力。spark.reducer.maxReqsInFlight每个 reducer 同时发出的拉取请求数上限默认是 Int.MaxValue。限制它可以避免同时打开太多连接。spark.shuffle.sort.bypassMergeThreshold分区数小于该值时走 bypass 排序路径默认 200。分区少时可以降低排序开销。spark.shuffle.io.maxRetries和spark.shuffle.io.retryWait拉取失败重试次数和等待时间。网络抖动频繁时可以适当调高。spark.sql.shuffle.partitionsSpark SQL 执行 shuffle 时默认生成的分区数默认 200。很多人把它当成万能旋钮其实它需要和数据量、并行度一起考虑。还有一个容易被忽略的点从 Spark 1.6 开始Executor 内存采用了统一内存管理shuffle 使用的 execution memory 和缓存 RDD 的 storage memory 之间可以互相借用。所以调 shuffle 时不能只看单独参数还要关注spark.memory.fraction和spark.memory.storageFraction的整体配比。4.3 从参数配置看两者调优方向调优关注点MapReduceSpark内存瓶颈环形缓冲区和 reduce shuffle 内存比例Executor 统一内存中 execution 部分和 storage 的分配文件数量通过合并 spill 文件降低文件碎片通过 sort-based shuffle 天然减少 data/index 文件数量数据压缩开启 map 输出压缩减少网络 IO靠spark.shuffle.compress控制CPU 换带宽重试语义任务失败重新执行拉取失败可重试shuffle 文件丢失则重算上游 stage分区数reducer 数量通常固定并行度可动态调整AQE 可以合并小分区看这个表你会发现MapReduce 的调优更偏硬件资源控制因为你几乎不能改变它的 shuffle 模型Spark 的调优更偏自适应策略因为有 AQE、动态合并这些运行时优化。这也是为什么 Spark 在 SQL 场景下比 Hive-on-MR 更吃内存但也能把资源利用得更细。5. 真实场景里的问题排查与避坑5.1 怎么快速判断 shuffle 卡在哪判断 MapReduce 卡在 shuffle 的最直接方法是看 JobHistory 里 reduce 各个子阶段耗时。如果 reduce 长时间停留在shuffle而不是reduce大概率是 fetch 或 merge 瓶颈。观察小文件数和 map 输出压缩率也能辅助定位。Spark 就直观得多。打开 Spark UI 的某个 stage 页面看 Shuffle Read/Write 大小以及 spill 相关指标。一个行之有效的排查思路先看 Shuffle Write 是不是远大于最终结果量如果是很可能缺少 map 端聚合再看是否有大量内存 spill如果是需要通过增大并行度或调整spark.sql.shuffle.partitions来降低单任务数据量最后看某个任务 Shuffle Read 明显大于其他任务那就是数据倾斜。我在实际排查中很少一上来就调参数而是先做一次数据量和分布的估算。shuffle 的瓶颈多数不是某个参数不合理而是分区数、数据量和 Executor 内存三者不匹配。比如一个 10 亿行的 join 产生上千 GB shuffle 数据分区数还默认 200每个分区几十 GB再好的参数也救不回来。5.2 数据倾斜两种框架的处理思路不一样数据倾斜是 shuffle 场景的经典问题。MapReduce 里出现倾斜通常是某个 key 的数据量远超平均导致对应 reducer 成为长尾。常规做法是加一层随机前缀打散 key或者自定义 Partitioner让超高热 key 分散到多个 reducer最终再聚合一次。Spark 里处理倾斜的手段更多但也更强调提前规避而不是事后补救。你可以通过 join 前过滤热点 key或者用 Broadcast Hash Join 把小表广播出去直接从源头避免 shuffle。Spark 3.0 之后 AQE 还会自动检测 skewed join把倾斜分区拆分后再 join这在某些场景下连 SQL 都不用改。这里有一个容易踩的坑Spark 里用repartition增加分区数不一定能解决倾斜。因为如果倾斜是由单个 key 引起的repartition只改变分区编号相同 key 仍然会被哈希到同一个分区。要做的是给 key 加盐或者确保spark.sql.adaptive.skewJoin.enabled已经打开。很多人以为提高并行度等于治倾斜结果只是增加了任务调度开销。5.3 高频问题速查表现象可能原因排查方向常用手段Reduce 阶段长时间不结束某些 map 输出分片很大或 fetch 慢看 map 输出大小和节点 IO开启 map 输出压缩调整 parallelcopiesSpark 任务大量 spillshuffle 数据量大于可用内存看 task 的 spill 指标提高并行度调大 execution 内存减少单任务数据量Shuffle Write 远大于输入缺少 map 端聚合看上游 RDD 分区结构使用 reduceByKey 类算子或利用 AQE 提前聚合某个 Executor OOM单 partition 数据量过大看 Shuffle Read 是否集中salting、动态分区裁剪或开 AQE skew join拉取数据频繁失败重试网络抖动或 Executor 退出看 shuffle service 日志调大 retries 和 retryWait开启 external shuffle service小分区太多导致调度开销大分区数远超数据量看 task 平均处理时间用 AQE coalesce 或调低spark.sql.shuffle.partitions这张表在面试里也经常被拿来扩展成你怎么定位一个 shuffle 慢的 Spark 作业。我的个人习惯是先抓住两个数Shuffle Write 总大小和最大 task 的 Shuffle Read 大小。前者决定 IO 瓶颈程度后者决定是否存在倾斜。其他的配置调整基本都要围绕这两个数来展开。6. 我的实操体会与新版本趋势6.1 一些常规文档不会写的坑我在生产环境里见过太多人把spark.sql.shuffle.partitions直接调到 1000以为并行度越高越快。实际上分区数一旦超过集群并行度多余的任务只是排队反而放大调度成本和网络连接数。分区数应该和 Executor 数量、单 Executor 并发能力相匹配而不是凭感觉设置。另一个容易被忽视的是 shuffle 压缩格式。MapReduce 时代很多公司默认开启 map 输出压缩Spark 也默认开启 shuffle 压缩。但压缩算法是有代价的。如果你的作业是 CPU 密集型的压缩可能让 CPU 打满而网络带宽空转如果你的作业是 IO 密集型的压缩反而是性价比最高的手段。推荐的做法是分别用 Snappy 和 LZ4 在小集群上做一次对照测试看整体耗时再决定。MapReduce 和 Spark 的 shuffle 还有一个理念差异很容易引发误解MapReduce 的 reducer 数据是有序的所以很多 Hive 用户会用全局排序来实现某些逻辑。迁移到 Spark 后如果你仍然依赖某个全局顺序或者隐式的有序分组就可能踩坑。Spark 的 shuffle 不保证 key 全局有序除非你显式使用sortByKey或类sort merge join的机制。好在 Spark 多数 aggregations 不依赖顺序但理解这一点能避免不少 SQL 语义上的意外。6.2 新版本都在往哪个方向走Spark 3.x 之后shuffle 的演进重心明显在自适应和减少不必要落盘两个方向。AQE 可以动态改变 reducer 分区数、优化倾斜 join、合并小分区push-based shuffle 试图把部分 shuffle 数据在 map 端聚合之后再推送减少下游直接拉取大量小块文件的压力。remote shuffle service 也在让 shuffle 文件的存储和服务与计算节点解耦进一步避免 Executor 崩溃带来的重算成本。反观 MapReduce 生态新的迭代已经基本进入维护期更多精力放在和计算引擎的整合上。但从原理上讲MapReduce 这种每一步都有明确落点和强排序的模型仍然是理解分布式 shuffle 最好的入门教材。很多 Spark 内部机制比如 spill、merge、分区、buffer底层思路都能从 MapReduce 里找到影子。我个人的体会是不要急着背参数也不要被两个框架的术语差异绕晕。先把一条数据从上游到下游的物理路径走一遍想清楚它在哪个节点、哪个内存区域、哪个文件里待过许多问题自然就通了。下次再有人问你 Spark shuffle 和 MapReduce shuffle 的区别你不需要枚举十点不同只需要说一句MapReduce 用强排序换稳定Spark 用内存和索引换速度而它们最终都要面对同一件事——把数据搬到该去的地方。