从MapReduce到Spark：大数据性能调优关键路径解析

做大数据的人只要经历过MapReduce那个年代大概都忘不了被“跑一批任务等到天亮”支配的感觉。那时候处理几百GB的数据凑一个稳定跑完的作业都算技术活更别说什么实时性、交互式查询。后来Spark出现内存计算的概念打出来很多团队像抓住救命稻草一样从MR迁到Spark但也有人换了Spark以后发现作业照样慢、照样OOM跑起来甚至不如调优过的MapReduce稳。这篇文章不聊空泛的架构演进全部围绕实际调优经验来写。从MapReduce到Spark本质不是换一个框架而是换一套对计算资源、数据Shuffle、内存模型的理解方式。你会看到为什么MapReduce慢、慢在哪个环节Spark快、快在哪里但坑又藏在哪里集群搭好之后哪些参数是真正影响性能的Shuffle怎么做才能不拖后腿Spark OOM怎么一步步排查还有Spark SQL处理外部数据源例如集成Redis、接入达梦数据库时那几个容易忽略的IO瓶颈。内容适合刚入门大数据、准备给现有批处理环境做升级的工程师也适合天天调Spark作业但始终觉得没摸到门道的朋友。1. 为什么性能调优绕不开“从MapReduce到Spark”这段路1.1 先明确要解决的真实问题先定义一个问题我们说的性能调优到底在调什么从MapReduce那个时代走到Spark的时代表面上是换框架、换API本质上是换了一套计算范式。MapReduce把每个计算都拆成Map和Reduce两个阶段数据每一轮都要写入磁盘下一轮再从磁盘拉回来。这个模型的好处是简单、稳定、容错容易做坏处也非常直观计算过程中全是磁盘IO和网络IO时间都花在搬运上。举个我实际接触过的案例。早年间做过一个用户行为日志的分析任务输入是HDFS上大约300GB的文本日志要做一次简单的词频统计和会话切分预处理。如果用MapReduceMap阶段读日志吐中间结果Sort和Shuffle阶段把数据按照Key排序分发Reduce阶段再拉数据聚合成最终结果。整个过程走一遍在大概10台物理机的集群上跑耗时大约是以小时为单位计算的。后来同样的逻辑迁到Spark RDD来写Stage内部能内存计算的部分全部走内存跑完一次作业的时间缩短到分钟级别。这个对比其实并不在于谁的算法更高级而是Shuffle和磁盘落地的次数相差太多。MapReduce里的每个Job几乎强制要求中间结果落盘因为它的容错模型就是“每个Stage重新执行”。Spark通过RDD的血统机制做容错Stage内部能缓存的中间结果直接保留在内存或按需计算不再强制落盘。这就是性能差距的一个根本来源。1.2 性能瓶颈到底在哪重新理解MapReduce的三个“太重”之前有人问过我一个问题MapReduce是不是被时代淘汰了我的看法是这个模型没有淘汰但实现思路确实过重。重在哪里拆开看有三点第一中间结果落盘太重。Map结束后MapOutput要写到本地磁盘Reduce任务启动后再从各个Map任务节点拉取数据这个过程既是磁盘IO又是网络IO。数据量一大整个作业的时间几乎都消耗在落盘和拉取上。第二调度模型太重。MapReduce的每个Task是JVM级别的进程意味着一个Job包含几千个Task就要反复启动几千个JVM。JVM启动本身的开销、序列化、反序列化、上下文切换这些和业务逻辑无关的额外消耗占比不小。Spark引入Executor常驻内存的模型Task以线程方式在同一进程内调度启动开销大幅下降。第三编程范式太重。MapReduce想表达一个多步骤的复杂计算就得把逻辑拆成多个Job串联每个Job之间通过HDFS传递中间数据这种串联模式让复杂分析逻辑代价极高。而Spark用RDD或DataFrame天然支持在单个作业里构建多Stage的DAGStage内部尽可能走内存管道化计算。从需求和成本的角度讲这个演进是必然的。2. 环境搭建Spark集群部署的参数取舍2.1 版本选型与基础配置Spark集群搭建这件事听起来简单真正影响性能的细节经常藏在安装之外。很多教程写spark安装步骤无非是下载二进制包、配置spark-env.sh、slaves文件、启动集群但这几步只能说让集群能跑离“能跑得快”还差得远。版本选型方面我建议直接用当前主线稳定版本比如Spark 3.4.x或3.5.x系列原因是3.0之后的Spark引入了Adaptive Query ExecutionAQE等自动优化能力。AQE可以在运行时根据实际Shuffle输出统计信息重新优化执行计划比如自动处理数据倾斜时拆分Reduce分区、自动合并过小的Shuffle分区这些功能对刚入手调优的人来说非常友好能减少不少手工调参的压力。配套生态需要提前想清楚。如果你主要跑批处理用YARN作为资源调度会比Standalone模式扩展性更好别图省事长期跑Standalone。如果只是本机学习或做小规模实训那Standalone加本地目录没问题。所谓spark集群搭建的“集群”二字起码要覆盖NameNode、DataNode、ResourceManager、NodeManager、Spark的Executor运行节点这些角色否则你在物理环境会踩到很多资源协商上的坑。以我曾经搭过的一个测试环境为例配置大致是这样节点角色配置用途主节点16核 / 64GB内存NameNode ResourceManager Spark Master4个工作节点8核 / 32GB内存DataNode NodeManager Spark Executor运行存储HDFS 3副本数据冗余这个配置并不高但关键问题是跑Spark作业时executor核心数和内存如何分配这决定了你能拿到多少并行度。总资源就这么多分配给每个Executor的核数越少Executor数量越多并行度越高但任务之间的网络Shuffle成本也可能上升。实操中常见的经验值是单个Executor给2~4个core内存按容器内预留一点给YARN overhead后尽量都给Spark使用。2.2 真正影响性能的部署参数安装完Spark之后先别急着跑作业有几个配置参数是必须过一遍的因为它们直接决定资源利用率。以下参数在spark-defaults.conf里设置或提交任务时用--conf传。spark.cores.max 8 # 整个应用最多可用的CPU核心数 spark.executor.memory 8g # 每个Executor的堆内内存 spark.executor.memoryOverhead 2g # 每个Executor的堆外内存 spark.executor.cores 2 # 每个Executor占用的CPU核心数 spark.sql.shuffle.partitions 200 # Spark SQL Shuffle默认分区数 spark.default.parallelism 8 # RDD默认分区数 spark.serializer org.apache.spark.serializer.KryoSerializer很多初学的人会以为spark.executor.memory给得越大越好实际并不完全是。Executor在YARN上申请的内存如果太大JVM的GC会成为新瓶颈。我见到过不少例子把Executor内存调到32G甚至64G结果Full GC频繁整个作业运行期间GC时间占到了30%以上。一般建议单个Executor堆内内存在8G到16G之间对绝大多数批处理作业来说已经足够。KryoSerializer值得单独说。Spark默认的Java序列化器方便但性能很差。对于数据量大、或者使用了自定义类的场景用spark.serializer切换Kryo能够明显减少序列化时间和内存占用。有些网上教程会建议在spark-defaults.conf里全局开启Kryo但如果你的代码里注册了自定义类一定要用spark.kryo.registrator或spark.kryo.classesToRegister提前注册否则Kryo在遇到未注册类时仍然会走落伍的路径性能优势大打折扣。3. Shuffle调优MapReduce和Spark的共同命门3.1 MapReduce侧怎么调Shuffle很多人从MapReduce迁移到Spark后容易忽略一个事实Shuffle仍然是两个框架共同的性能命门。MapReduce的Shuffle包括Map输出端的Sort和Spill以及Reduce端从各节点拉取中间结果的Copy。这里的Sort到底能不能关在Hadoop早期的版本里如果想在Map端做Combine比如本地聚合必须先Sort保证Key有序才能高效合并所以Sort几乎是强制性的。后来Hadoop做了优化有一种Secondary Sort机制但那时写Java代码的复杂度已经让很多人放弃了。在MapReduce里实际能调的Shuffle参数主要这几项mapreduce.task.io.sort.mb决定Map输出缓冲区大小改大了能减少Spill次数mapreduce.reduce.shuffle.parallelcopies决定Reduce端并行拉取Map输出的线程数设得大能加快Copy阶段但网络开销也会增加mapreduce.map.sort.spill.percent默认0.8触发Spill的阈值调大也能减少落盘次数。这些参数作用都是缓解同一个老大难问题数据搬运成本。反过来我们会发现Spark之所以能在Shuffle上更高效是因为Spark引入了HashShuffle和后来的SortShuffle经历过多个版本的迭代。到了Spark 2.x之后默认的Shuffle方案是SortShuffleManager。它把Map的输出按照Partition写入内存缓冲再排序落盘每个Map任务最终生成的数据文件数量不再是“Reduce分区数”那种爆炸式增长而是尽量合并让Reduce端拉取更容易。3.2 Spark侧Shuffle优化和数据倾斜处理落到实际调优场景Spark的Shuffle优化主要关注三点分区数、缓冲大小、以及数据倾斜。spark.sql.shuffle.partitions是Spark SQL作业里出现频率最高的一个参数。理论上说Shuffle分区数越多每个Task处理的数据量越小并行度越高。但分区数太多也会产生大量小的Shuffle文件反而让调度和网络传输开销变大。合理区间建议是Shuffle数据量在几十GB以内时分区数设置在200到500之间通常表现不错如果单个Task处理时间差异巨大再看是否要动态开启AQE。数据倾斜是Spark作业最常见的性能杀手。表现是集群上大部分Task快速跑完一两个Task卡了很久最后整个Job等那个Task。解决倾斜没有万能法常用的手段如下先定位倾斜的Key到底是什么。可用df.groupBy(key).count().orderBy(desc(count)).show()方式快速看分布。如果是热点Key数量不多可以考虑加随机前缀再聚合分两步做第一步给热点Key打散第二步去掉前缀归并真实结果。这个办法对两阶段聚合场景效果比较明显。如果倾斜是发生在Join阶段比如一张小表和一张大表Join可以把小表用BroadcastHashJoin广播避免Shuffle。如果本身是大表Join大表可以先过滤掉脏数据比如空Key再考虑对热点前缀加盐。Spill和GC也是个容易被忽视的问题。Shuffle读阶段如果抛出类似Shuffle file cannot find的异常大概率是Executor在Fetch时内存不够把曾经落盘的Shuffle文件删掉导致。遇到这种问题要增加spark.shuffle.memoryFraction对应的内存或者在Spark 2.x之后直接调spark.memory.fraction同时检查Executor是否频繁GC。4. 内存管理从Spark OOM说起4.1 运行时内存模型拆解Spark作业跑挂的重要原因除了资源没申请够之外很多都和内存模型理解不到位有关。Spark的Executor内存可以拆成三块堆内执行内存Execution Memory、堆内存储内存Storage Memory、以及堆外内存Off-heap Memory。在旧版本里Execution Memory和Storage Memory是按比例静态划分的两边的空间不能互相借用导致经常出现“执行内存不够但存储内存闲置”的尴尬。从1.6开始引入统一内存管理Unified Memory让Execution和Storage可以互相抢占多余的闲置空间这是一个非常大的改进。但统一内存不代表没有OOM。通常说的Spark OOM一部分是JVM堆内存真正不够OutOfMemoryError: Java heap space另一部分是堆外内存耗尽也就是Container killed by YARN for exceeding memory limits这个在日志里很常见。后者的起因往往是spark.executor.memoryOverhead或spark.memory.offHeap.size设置不合理。尤其在使用外部库比如读取Redis、访问JDBC数据源的时候底层库分配的直接内存或线程栈有可能不计入Spark的堆内预算一旦超过YARN容器阈值就会被kill。4.2 常见的OOM场景排查步骤如果你发现Spark作业反复OOM先不要急着一味加Executor内存按下面的顺序排查更高效第一步看是Driver端还是Executor端OOM。Driver端OOM通常是因为collect()把全量结果抓到内存或广播变量太大。Executor端OOM则要去分析Task处理的数据量。第二步看日志确认堆内还是堆外。堆内OOM一般是数据分区过大或代码里有巨大的集合缓存堆外OOM则多半是序列化缓冲或外部资源占用过多。第三步检查是否存在数据倾斜。如果一个处理流程中某个Task拉取的数据量特别大那么就是典型倾斜把倾斜的Key打散或调大并行度比简单加内存效率高得多。这里我额外想强调一个亲身踩过的坑别在RDD和DataFrame的算子内部维护大型集合对象。比如用mapPartitions做外部API调用时把整个Partition的数据攒成ArrayList再批量提交分区的数据量大时这个List会直接在Executor堆内产生巨大压力。我当时就是在一个数据清洗作业里用mapPartitions批量读取Redis的值为了减少网络往返把一个Partition内上万条Key的Value全部攒到List再写库结果整个Executor被撑满日志报的是堆内OOM。后来改成边读边写按批大小限制在1000条以内内存问题立刻消失。5. Spark SQL性能优化与外部数据源集成实践5.1 Spark SQL调优三板斧实际生产环境里大部分人写的不是RDD算子而是Spark SQL。所以Spark SQL的性能调优更需要熟练掌握。我把它归纳成三板斧代码本身的执行计划优化、资源配置与运行时参数优化、外部存储IO优化。第一板斧是执行计划优化。Spark SQL通过Catalyst优化器生成物理执行计划你要做的是学会用explain()去看执行计划确认Join的方式是BroadcastHashJoin还是SortMergeJoin确认有没有不必要的全表扫描有没有可以做下推过滤却没做的情况。动态分区裁剪和文件中列裁剪这些能力在Spark 3.x里默认开启了一部分但某些嵌套子查询还会退化成全量扫描这时候可以通过改写SQL让过滤条件下推。第二板斧是资源配置。AQE在Spark 3.x默认是开启的Spark 3.4里相关参数比较完整包括自动处理Join时数据倾斜、自动合并Shuffle分区、自动切换Join策略。AQE是好东西但对于依赖稳定分区数的报表任务也有可能出现“动态合并后某个大分区还是过重”的情况。出现这种问题时建议固定spark.sql.adaptive.coalescePartitions.enabled为false回到手动设置分区数的方式。第三板斧是外部存储IO优化。这是Spark SQL实战里最容易忽略的地方也是最容易出效果的地方。以Spark集成Redis为例很多人直接写一个UDF在Task内部遍历调用Redis客户端获取数据。这种做法如果键数量少还好键一多网络往返延迟就会被放得很大。较合理的方案是使用Redis的pipeline或批量获取命令或者将需要关联的数据预读并广播到每个Executor变成本地内存查找。当然如果预读数据很大就老老实实做Join。Spark SQL读取外部数据库也有类似道理。拿达梦数据库DM来举例国产化环境中需要对接到Spark时应该怎么处理重点之一是JDBC连接参数。每次读取数据numPartitions决定并行度lowerBound、upperBound决定拆分列范围。如果一张千万级的大表只用一个连接去拉Spark会先给你一个Timeout。给JDBC Reader配置合理的分区数量、每次批量拉取的行数、连接超时时间能明显缓解全量拉取时的压力。另一个容易踩的点是底层驱动不一定支持下游的数据类型读取时尽量用dbtable的方式写子查询提前把列映射到Spark认识的类型上。5.2 一个可参考的实际调优案例把前面这些点穿起来看一个简单的分析案例场景是读取存放在HDFS上的用户行为明细表需要先用Spark SQL做数据清洗然后关联一份从Redis读取的白名单数据最后关联达梦数据库中的用户维度表做宽表输出。整个过程可能会遇到三个层面的问题清洗阶段Shuffle太多、关联Redis太慢、读取达梦库形成单点瓶颈。对应调整的思路清洗阶段先过滤后聚合。提前用filter做行裁剪再用select做列裁剪减少进入Shuffle的数据量Join白名单时如果白名单本身是百万量级以内用广播变量方式将白名单做成Map并在Task内直接lookup而不是对每条明细发起Redis命令。接入达梦维度表时给JDBC源设置合理的分区列尽量使用主键或数值型字段做partitionColumn同时按Where条件切分多个子查询并发读取避免一个连接拉全表。调整之后整个任务从原来的20多分钟缩短到5分钟以内。这个效果并不来自某一种高深的魔法而是每一个IO环节都减掉了多余的搬运和等待。6. 数据倾斜之外的一些“软性”调优点6.1 从MapReduce编程实例里学到的资源利用思维处理过一个传统MapReduce编程实例统计网站上每个页面的PV和UV。这个实例有多个阶段需要两个MR作业串联第一个MR完成PV统计并输出按UV去重后的预聚合数据第二个MR做最终汇总。如果严格按照两个独立的MapReduce作业来跑每个作业都要把中间结果写一遍HDFS再拉一遍整个链路的时间翻倍。后来我们在第一个Reducer里把预聚合做得足够细致让第二个Job的Map阶段几乎只做读取全局排序的意义已经不大这样总耗时下降不少。这个思路放到Spark里同样适用。不要因为Spark支持在DAG里做多Stage就随意把所有操作都串联起来每个Stage之间的宽依赖仍然对应一次Shuffle。写代码之前先在纸上画出数据流标注哪些地方会发生Shuffle哪些宽依赖其实可以转化成窄依赖。比如能用reduceByKey做局部聚合再用全局聚合的地方就别一上来直接groupByKey。这个原则从MapReduce时代到Spark时代都没有变过。6.2 Spark面试常见考点对应的调优理解顺着话题多说几句Spark面试题里经常出现“Spark为什么比MapReduce快”、“数据倾斜怎么解决”、“Spark OOM怎么处理”、“Spark SQL的执行流程是什么”。这些考点看起来分散其实对应的都是一个工程师做调优时需要掌握的基本盘快在计算模型和调度模型数据倾斜是Shuffle不均OOM是内存模型没吃透Spark SQL是在Catalyst优化器之上写SQL。如果能把这几个点想透面试和实际调优基本上都没有太大障碍。反过来说如果只是一味背参数而不知道参数背后的模型换一个集群环境就会手足无措。有个新趋势是DGX Spark这类硬件加速集成方案开始出现它把Spark的批处理跑在GPU 加速硬件上这类方案要求对执行模型的IO瓶颈和调度开销同样有比较深的理解否则也是白搭。遇到这种环境建议从小的数据分析案例入手先在无状态作业里跑通再逐步加复杂依赖否则很难定位新增的瓶颈到底在算子还是调度层。6.3 集群日常维护的调优习惯最后补一些日常操作细节。Spark集群搭好作业也能跑了并不意味着可以躺平。我自己习惯在每类任务跑完后去Spark UI上面看几个数值Executor的GC时间、Shuffle读写量、单个Task耗时分布、以及是否频繁发生Speculative Task重试。这些数据比任何理论都更能说明问题。如果GC时间偏高优先处理代码中多余的缓存对象如果Shuffle数据量远远大于输入数据量说明在Shuffle上游可以做更彻底的预聚合或列裁剪如果Task耗时中位数很低但某些尾巴特别高大概率又是倾斜问题。还有每周清理一次Spark日志和Event Log的习惯防止磁盘占用过高导致节点异常。看起来是运维活但生产环境的速度和稳定性往往来自这些不起眼的地方。我在实际使用中也发现很多人把性能调优理解成“加资源、加并行度”没解决本质的Shuffle和IO问题时资源堆上去只是暂时掩盖问题数据量一涨又被打回原形。真正有效的调优顺序永远是先定位瓶颈在哪一层再对该层做最便宜的优化。MapReduce到Spark的迁移只是提供了更多优化的空间不是终点。遇到新框架、新硬件平台时这个思路依然管用。