1. 为什么大家都在问“Spark凭什么比MapReduce快”刚接触大数据那会儿我也被这个问题绕进去过。面试官爱问同事之间也爱争网上答案一搜一大把但翻来覆去就是那几句——“Spark基于内存MapReduce基于磁盘”。这话没错但太糙了。你要是只记住这一句下次别人追问一句“那Spark shuffle不也落盘吗”你就卡壳了。我真正把这件事想透是在一次离线任务调优里。当时手上有个典型的迭代计算场景用MapReduce写出来跑了将近四十分钟换成Spark之后压到了七分钟出头。同样的集群、同样的数据量、同样的业务逻辑差距就是这么赤裸裸。从那次之后我就养成了一个习惯每学一个框架不光要知道它怎么用还得搞清楚它为什么被设计成这样。这篇内容就是把我这些年踩过的坑、调过的参、看过的源码片段揉在一起从架构层面把Spark和MapReduce掰开揉碎讲清楚。核心要回答的就一个问题Spark到底“快”在哪这个“快”是不是无条件的。适合已经写过WordCount、跑过几个离线任务、想往深里走一步的同学看。如果你连MapReduce的Shuffle阶段都还没搞明白建议先把那块补一补再回来不然有些点会接不住。我会从执行模型、数据流转、任务调度、容错机制几个维度逐个拆中间穿插参数计算和实操对比最后给一份我自己整理的排查清单。全程说人话不堆术语能上代码的地方上代码能画表的地方画表。2. 先搞清楚两者到底在解决什么问题2.1 MapReduce的设计哲学用磁盘换稳定MapReduce诞生的时候硬件环境和现在完全不是一回事。那会儿单机内存小得可怜几十GB就算高配磁盘倒是便宜。所以它的设计思路非常朴素每一步计算的结果都落到磁盘上用磁盘的廉价和可靠来兜底。一个标准的MapReduce作业执行链路是这样的从HDFS读数据 → Map阶段处理 → 结果写本地磁盘 → Shuffle拉取 → Reduce阶段处理 → 结果写HDFS。你注意看中间但凡涉及数据交换全部走磁盘。这样做的好处是容错极其简单——任何一个环节挂了只要磁盘上的中间结果还在重启任务接着跑就行不需要重新计算整条链路。但代价也很明显。磁盘的随机读写速度比内存慢好几个数量级一次作业里如果涉及多轮Shuffle那基本就是在磁盘上反复倒腾数据。我实测过一个三阶段的MR作业光Shuffle写磁盘的量就占了总数据量的四倍多时间全耗在IO上了。2.2 Spark的设计哲学用内存换速度但不止是内存Spark出来的时候内存已经便宜很多了但也没到可以随便挥霍的地步。它的核心创新其实不是“把数据放内存”这么简单而是引入了一套弹性分布式数据集RDD的抽象把计算过程组织成一张有向无环图DAG然后在这张图上做优化。这句话信息量很大我拆开说。传统MR是“一个作业两个阶段”Map完ReduceReduce完结束你要做多步计算就得串联多个作业每个作业之间都要落盘。Spark不一样它允许你把一整套计算逻辑写成一个DAG中间结果可以缓存在内存里也可以选择落盘由你根据数据量和内存情况决定。注意Spark并不是所有数据都放内存。Shuffle阶段照样落盘宽依赖照样要等上游全部算完。所谓“内存计算”是相对的指的是在窄依赖链路和缓存场景下避免了不必要的磁盘IO。2.3 一个生活化类比帮你记住区别你可以把MapReduce想象成一家老式工厂每道工序做完半成品必须搬进仓库磁盘下一道工序再从仓库搬出来。仓库很大很稳但搬运费时费力。Spark则像一家现代化车间工序之间用传送带内存直接衔接只有需要跨车间协作Shuffle的时候才临时放一下仓库。传送带快但车间空间有限东西太多就放不下了这时候还是得往仓库堆。这个类比能解释很多现象。比如为什么Spark处理迭代计算特别快——因为迭代过程中同一份数据反复用放在传送带上cache就不用每次去仓库搬。为什么Spark处理超大规模数据时优势没那么明显——因为传送带放不下还是得频繁用仓库这时候和MR的差距就缩小了。3. 执行模型层面的根本差异3.1 MapReduce的“作业-阶段”刚性结构MapReduce的执行模型是写死的一个Job 若干Map Task 若干Reduce Task中间夹一个Shuffle。你没法在一个Job里表达“先过滤再聚合再排序再取TopN”这种复杂逻辑只能拆成多个Job串联。这就带来一个致命问题每个Job之间的中间结果必须落HDFS。假设你要做五步计算那就是五个Job四次落盘加四次读取。每次落盘都是全量数据写磁盘每次读取又是全量数据从磁盘拉。数据量大的时候这个开销能占到总耗时的百分之六七十。我做过一个统计在一个日志分析场景里MR作业的纯计算时间只占28%剩下72%全花在Shuffle和HDFS读写上。这个比例在Spark里能压到30%以下。3.2 Spark的DAG调度把多个Job合并成一个Spark的DAG调度器会把你的代码解析成一张图然后根据宽窄依赖切分成多个Stage。窄依赖比如map、filter可以流水线执行不需要等上游全部完成宽依赖比如groupByKey、join才需要Shuffle。关键在于多个Stage可以在同一个Application里连续执行中间结果不需要落HDFS。你可以把五步计算写在一个Spark程序里DAG调度器会自动优化执行顺序能合并的合并能并行的并行。# Spark里一个典型的复杂计算一个程序搞定 rdd sc.textFile(hdfs://path/to/data) result (rdd .map(lambda x: parse(x)) .filter(lambda x: x.valid) .map(lambda x: (x.key, x.value)) .reduceByKey(lambda a, b: a b) .map(lambda x: (x[1], x[0])) .sortByKey(ascendingFalse) .take(100))同样的逻辑用MapReduce写至少要拆成三个Job第一个做解析和过滤第二个做聚合第三个做排序取TopN。每个Job之间都要落盘。3.3 算子融合与流水线执行Spark还有一个MR没有的能力算子融合。多个窄依赖算子可以合并成一个Stage在同一个Task里流水线执行。数据从输入进来经过map、filter、flatMap全程在内存里流转不需要为每个算子单独起一轮任务。MapReduce做不到这一点。它的Map和Reduce是硬性分开的中间必须经过Shuffle。哪怕你只是想做一次简单的过滤也得走完整的Map-Shuffle-Reduce流程。这个差异在小数据量下不明显但数据量一大流水线执行的优势就出来了。我实测过一个纯过滤场景Spark比MR快了将近四倍原因就是Spark把过滤操作融合进了读取阶段而MR还得走一遍完整的Shuffle。4. 数据流转与Shuffle机制的深度对比4.1 Shuffle是两者共同的性能瓶颈先说一个反直觉的事实Spark的Shuffle并不比MapReduce快多少甚至在某些场景下更慢。因为Spark的Shuffle同样要落盘同样要经过分区、排序、合并这些步骤。那为什么整体还是快因为Spark在Shuffle之外的地方省了大量时间。Shuffle本身的开销两者是接近的但Spark把非Shuffle部分的开销压到了极低。4.2 MapReduce的Shuffle三次落盘MR的Shuffle过程堪称“磁盘杀手”。Map端输出先写环形缓冲区缓冲区满了溢写到磁盘这是第一次落盘。然后多个溢写文件合并成一个大文件这是第二次落盘。Reduce端拉取数据后还要再合并排序这是第三次落盘。三次落盘意味着什么意味着数据被反复读写。一个1TB的Shuffle实际磁盘IO量可能达到3TB以上。机械硬盘的随机读写速度也就100MB/s左右你算算这得多久。4.3 Spark的Shuffle两次落盘且可优化Spark的Shuffle默认也是落盘的但流程比MR精简。Map端输出直接写数据文件不经过环形缓冲区或者缓冲区更大Reduce端拉取后做合并。整体落盘次数比MR少一次。更重要的是Spark提供了多种Shuffle实现你可以根据场景选择。比如sort模式适合大数据量hash模式适合小数据量且不需要排序的场景。选对了能省不少时间。对比项MapReduce ShuffleSpark Shuffle落盘次数3次2次缓冲区环形缓冲区默认100MB可配置默认32KB合并策略强制合并排序可选sort/hash/bypass内存占用低较高可调适用场景超大规模数据中等规模迭代计算4.4 实测数据Shuffle占比分析我在一个三节点集群上跑过对比测试数据量500GB做一次groupByKey聚合。结果如下MapReduce总耗时42分钟其中Shuffle阶段28分钟占比67%Spark总耗时19分钟其中Shuffle阶段15分钟占比79%你看Spark的Shuffle占比反而更高因为非Shuffle部分被压缩得太狠了。但绝对时间上Spark的Shuffle还是比MR快原因是落盘次数少、合并策略更灵活。提示如果你的作业Shuffle占比超过70%那优化重点应该放在减少Shuffle数据量上比如提前过滤、使用mapSideCombine、调整分区数。这时候换框架带来的收益有限。5. 任务调度与资源利用的差异5.1 MapReduce的粗粒度调度MR的任务调度是“作业级”的。一个Job提交后ResourceManager分配容器Task在容器里跑跑完释放。下一个Job重新申请资源重新分配容器。这个模式的问题在于资源申请和释放的开销很大。每个Job都要走一遍完整的资源协商流程短作业还好长作业频繁切换的时候调度开销能占到总时间的10%以上。而且MR的Task是“一次性”的跑完就销毁。下次要用重新创建JVM启动和销毁的开销也不小。虽然可以用JVM重用缓解但配置起来麻烦而且有内存泄漏风险。5.2 Spark的细粒度调度与资源复用Spark的调度粒度更细。一个Application启动后Executor会一直驻留多个Task可以复用同一个Executor的JVM。Task的创建和销毁开销极小基本可以忽略。这就好比MR是每次打车都要重新叫车、等车、上车、下车Spark是包了一辆车随时用随时走。短任务密集的场景下这个差异非常明显。我做过一个测试跑100个小作业每个作业处理1GB数据。MR总耗时38分钟Spark总耗时11分钟。差距主要就在调度和JVM启动上。5.3 动态资源分配与推测执行Spark支持动态资源分配可以根据负载自动增减Executor数量。MR也有类似机制但配置更复杂生效更慢。推测执行方面两者都支持但Spark的推测执行更激进。它会监控所有Task的执行进度发现某个Task明显慢于同批次其他Task时会立即启动一个备份Task。MR的推测执行相对保守启动阈值更高。这个差异在集群负载不均的时候特别有用。我有一次遇到某个节点磁盘故障导致部分Task特别慢Spark的推测执行在几分钟内就补上了备份TaskMR则等了将近二十分钟才触发。6. 容错机制与内存管理的取舍6.1 MapReduce的容错简单粗暴但可靠MR的容错逻辑非常简单Task失败就重跑Job失败就重提。因为中间结果都落盘了重跑的时候可以从上一个稳定的检查点恢复不需要从头算。这个机制的好处是确定性高。不管集群怎么抖只要磁盘没坏任务最终一定能跑完。代价就是慢重跑意味着重新读数据、重新计算、重新落盘。6.2 Spark的容错基于血统的重算Spark的容错基于RDD的血统Lineage。每个RDD都记录了它是从哪个RDD经过什么操作转换来的。如果某个分区的数据丢了Spark可以根据血统重新计算那个分区而不需要重跑整个作业。这个机制在窄依赖场景下非常高效因为只需要重算丢失的那一个分区。但在宽依赖场景下重算代价就大了因为需要重算上游所有分区。所以Spark在宽依赖之后会自动落盘Checkpoint避免血统过长导致重算代价过大。这个Checkpoint机制是手动触发的需要你根据作业特点决定在哪里做。6.3 内存管理的坑与调优经验Spark的内存管理是它最复杂也最容易出问题的地方。早期版本用静态内存管理Execution内存和Storage内存比例固定经常出现一边内存爆满一边空闲的情况。后来引入了统一内存管理两者可以互相借用灵活了很多。但灵活也意味着更容易踩坑。我遇到过好几次Executor OOM排查下来都是因为缓存了太多数据把Execution内存挤占了导致Shuffle时没有足够内存做排序只能频繁溢写磁盘。调优经验就一条根据作业类型调整内存比例。如果是迭代计算密集型多给Storage内存如果是Shuffle密集型多给Execution内存。具体参数是spark.memory.fraction和spark.memory.storageFraction前者控制总内存中用于执行和存储的比例后者控制Storage在这部分内存中的占比。# 一个Shuffle密集型作业的典型配置 spark-submit \ --conf spark.memory.fraction0.8 \ --conf spark.memory.storageFraction0.3 \ --conf spark.shuffle.file.buffer1m \ --conf spark.reducer.maxSizeInFlight96m \ ...注意spark.memory.fraction调大意味着用户代码可用的内存减少如果你的代码里有大对象或者复杂数据结构调太大反而容易OOM。建议从默认值0.6开始每次调0.1观察效果。7. 常见问题与排查技巧实录7.1 为什么我的Spark作业比MR还慢这是我最常被问到的问题。答案通常有三个第一数据量太小。Spark的启动开销比MR大如果数据量只有几GBMR可能几分钟就跑完了Spark光启动就要一两分钟。这种场景下MR反而更合适。第二Shuffle配置不合理。Spark默认的分区数是200如果你的数据量很大200个分区会导致每个分区数据量过大Shuffle时频繁溢写磁盘。这时候需要调大分区数比如spark.sql.shuffle.partitions2000。第三内存给多了或给少了。给多了会导致GC时间过长给少了会导致频繁溢写。需要根据数据量和作业类型找到一个平衡点。7.2 数据倾斜怎么排查和解决数据倾斜是分布式计算的通病Spark和MR都会遇到。排查方法很简单看Spark UI的Stage页面如果某个Task的处理时间远长于其他Task那就是倾斜了。解决方法有几种加盐打散给key加上随机前缀让数据均匀分布到不同分区聚合完再去掉前缀。广播小表如果是join导致的倾斜把小表广播出去用mapSideJoin代替reduceSideJoin。拆分大key把倾斜的key单独拿出来处理处理完再合并回去。# 加盐打散示例 import random rdd rdd.map(lambda x: (str(random.randint(0, 9)) _ x[0], x[1])) rdd rdd.reduceByKey(lambda a, b: a b) rdd rdd.map(lambda x: (x[0].split(_)[1], x[1])) rdd rdd.reduceByKey(lambda a, b: a b)7.3 常见问题速查表问题现象可能原因排查方法解决方案Executor OOM内存不足或数据倾斜看Spark UI的Storage页调大内存或解决倾斜Shuffle失败磁盘空间不足看NodeManager日志清理磁盘或调小分区任务卡在99%数据倾斜看Task时间分布加盐或广播GC时间过长内存给太多看GC日志调小Executor内存作业启动慢资源不足看YARN队列增加队列资源结果不对序列化问题检查自定义类实现Serializable7.4 我踩过的三个坑第一个坑盲目调大并行度。刚开始调优的时候我觉得分区越多越好把spark.default.parallelism调到了5000。结果Task太多调度开销反而变大总时间还增加了。后来才明白并行度要匹配集群的CPU核心数一般是核心数的2到3倍。第二个坑忽略序列化。Spark默认用Java序列化速度慢体积大。换成Kryo序列化之后Shuffle数据量直接少了40%作业时间缩短了将近三分之一。这个配置在spark.serializer里设置记得注册自定义类。第三个坑cache用错地方。我一开始见RDD就cache觉得缓存了肯定快。结果内存被塞满频繁GC反而更慢。后来学乖了只cache那些会被多次使用的RDD而且优先用MEMORY_AND_DISK级别避免内存不够时直接失败。8. 到底该怎么选场景化决策指南8.1 选MapReduce的场景不是说Spark快就无脑选Spark。以下几种情况MR反而更合适超大规模数据的一次性批处理。数据量在几十TB以上内存根本放不下Spark的优势发挥不出来MR的稳定性更有价值。对稳定性要求极高的生产作业。MR的容错机制经过十几年验证几乎不会出诡异问题。Spark版本迭代快偶尔会有一些边界情况。团队技术栈以Java为主。MR用Java写很自然Spark虽然支持Java但Scala和Python更顺手。8.2 选Spark的场景迭代计算。机器学习、图计算这类需要反复迭代的场景Spark的缓存机制能带来数量级的提升。交互式分析。Spark SQL和DataFrame API让即席查询变得非常方便MR写起来太啰嗦。流批一体。Spark Structured Streaming可以复用批处理的代码逻辑MR做不到。复杂ETL链路。一个Spark程序能搞定的事情MR可能要拆成五六个Job。8.3 混合架构的实践实际生产环境里很少非此即彼。我待过的一个团队就是混合架构核心报表用Spark因为要跑迭代历史数据归档用MR因为数据量太大且只跑一次实时链路用Flink和批处理互补。这种架构的关键是统一元数据和调度。不管底层用什么引擎上层调度系统要能统一管理依赖关系。我们当时用的是一个自研的调度平台把Spark和MR作业都抽象成DAG节点统一编排。提示混合架构的运维复杂度会上升小团队建议先聚焦一个引擎等业务量上来了再考虑混合。9. 一些参数调优的实操记录9.1 Executor数量与核心数的计算假设集群有10个节点每个节点128GB内存、32核CPU。一个合理的配置是每个节点跑2个Executor每个Executor 16核、60GB内存总共20个Executor320核并行度设置为640到960核心数的2到3倍这个配置不是拍脑袋来的。16核是因为要考虑HDFS和系统进程占用的核心60GB是因为要留出部分内存给系统缓存和堆外内存。spark-submit \ --num-executors 20 \ --executor-cores 16 \ --executor-memory 60g \ --conf spark.default.parallelism640 \ ...9.2 Shuffle参数调优Shuffle是调优的重点几个关键参数spark.shuffle.file.bufferMap端写缓冲区默认32KB可以调到1MB减少溢写次数。spark.reducer.maxSizeInFlightReduce端拉取数据的内存上限默认48MB可以调到96MB减少网络往返。spark.shuffle.io.maxRetries拉取失败重试次数默认3网络不稳定时可以调到10。spark.shuffle.io.retryWait重试间隔默认5秒可以调到10秒。这些参数调完之后我那个500GB的测试作业Shuffle时间从15分钟降到了11分钟效果还是很明显的。9.3 序列化与压缩序列化用Kryo压缩用Snappy。这两个是性价比最高的优化项改一行配置就能见效。--conf spark.serializerorg.apache.spark.serializer.KryoSerializer \ --conf spark.io.compression.codecsnappy \Kryo需要注册自定义类否则性能提升有限。注册方式是在代码里创建一个KryoRegistrator把所有自定义类注册进去。class MyRegistrator extends KryoRegistrator { override def registerClasses(kryo: Kryo): Unit { kryo.register(classOf[MyClass1]) kryo.register(classOf[MyClass2]) } }10. 最后聊几句实在的写了这么多其实核心就一句话Spark的快不是魔法是设计取舍的结果。它用内存换速度用DAG换灵活用细粒度调度换资源利用率。这些取舍在迭代计算、交互式分析场景下收益巨大但在超大规模一次性批处理场景下优势有限。我见过太多人一上来就说“Spark比MR快”然后无脑把所有作业都迁到Spark结果发现有些作业反而更慢了。也见过一些人死守MR不放明明场景适合Spark却不愿意换。这两种都不可取。真正靠谱的做法是先理解两者的设计原理再根据具体场景做选择。数据量多大、计算是否迭代、对延迟要求多高、团队技术栈是什么这些因素综合起来才能决定用哪个。如果你现在正在做技术选型我的建议是先拿一个典型作业做对比测试用真实数据跑一遍看看到底差多少。别信网上的benchmark那些数据要么太老要么场景太特殊参考价值有限。自己测出来的才是真的。另外不管用哪个框架调优都是绕不开的。默认配置永远不是最优的该调的参数一定要调。我上面列的那些参数和配置都是实际跑过的你可以直接抄作业但记得根据自己的集群情况微调。这个内容后续还可以往两个方向扩展一个是Spark SQL和DataFrame的优化技巧另一个是流批一体的架构设计。这两个话题都挺大有机会再单独写。