Spark基础实战：从MapReduce到内存计算与作业调优

很多朋友问我Spark基础该怎么学的时候我一般会先反问一句你处理数据的方式还是算完一步存一步的那种吗如果答案是是那Spark真正值得你花时间的地方就不是那几个API怎么背而是它为什么能改变这种节奏。这篇内容我按自己早期带团队上手Spark的经验来写不堆概念尽量把为什么要这样设计作业跑到集群上之后到底发生了什么讲清楚适合刚接触分布式计算、想系统建立Spark基础的读者也适合已经会写Spark SQL但总在调优时一头雾水的人。1. 谈Spark不能不先聊MapReduce内存计算到底改变了什么1.1 为什么中间结果落盘是最大的性能杀手在Spark出现之前大家处理大规模数据主要靠MapReduce那套思路一个作业拆成Map阶段和Reduce阶段每个阶段处理完的结果先写到磁盘上下一个阶段再从磁盘读出来继续算。这种设计的好处是简单、稳每个环节失败了大不了重新读一遍磁盘里的数据但代价也非常明显——磁盘读写成了整个作业的咽喉。拿一个典型的迭代式算法举例比如K-means聚类或PageRank这类算法要反复扫描同一份数据每迭代一次就要把中间结果重新落盘一次。数据量一旦上了几个TB每一轮迭代花在序列化、网络传输、磁盘IO上的时间远远大于真正计算的时间。我第一次帮某公司优化一个用户画像作业时他们的做法就是每个MapReduce步骤都把中间表写到HDFS上下一段再load进来整个链路跑一次要三个多小时其中至少一半时间都耗在了重复读写上。Spark做的事情其实也不神秘它把中间结果尽量留在内存里让同一个作业的不同阶段之间不再频繁走磁盘。所谓内存计算本质上就是让数据在管道里流动起来而不是每过一个算子就去仓库里搬一次货。这也是为什么Spark官方资料里常说它比MapReduce快10到100倍——这个数字当然有水分但如果你面对的恰好是迭代式计算或者需要反复join、filter的场景量级上的提升是真实的。1.2 RDD与血缘把计算留在内存故障恢复靠的是重算而不是备份Spark里最基础的数据抽象是RDD弹性分布式数据集。弹性这个词容易劝退新人其实说白了就是数据切成了很多个分区分散在多台机器的内存里一旦某个分区丢失或算坏了Spark不会像MapReduce那样从磁盘备份恢复而是沿着这条数据当初的血缘Lineage重新算一遍。血缘是个什么概念你可以把它理解成一张数据是怎么一步步变出来的族谱。比如你从源数据读进来一个RDD然后做filter、map、reduceByKey每一步都会生成一个新的RDD新RDD知道自己的父RDD是谁、做了什么变换。如果某个分区在计算过程中炸掉了Spark就回溯血缘链从最早的源头重新执行那几条变换只重算丢失的分区而不是把整份数据倒腾一遍。这套设计的巧妙之处在于它把容错的代价从存转移到了算。内存不够用或者节点挂掉的时候宁可花点CPU重新算也省去了冗余备份带来的大量磁盘和网络开销。当然代价也很直接如果血缘链特别长、某个上游特别重重新计算可能比读磁盘还慢。所以后来DataFrame的优势之一就是能让你在写代码阶段就看到一条相对优化的血缘链而RDD时代经常要自己手动做persist或checkpoint来控制这个风险。1.3 在学Spark之前你最好先具备的几块拼图基础教程最怕的就是读者连前置知识都没准备好看什么都像天书。学Spark之前我建议先确认自己懂四样东西第一至少会用一种编程语言Python或Scala都行Java稍微累一点第二知道什么是分布式文件系统不用会运维但得理解数据其实是分成块放在不同机器上的第三懂一点SQL因为Spark里最主流的写法最后还是落在DataFrame和SQL上第四对Linux基本命令行不陌生因为无论本地调试还是提交集群作业都绕不开shell。这些都不需要精通。我带过的A同学之前只写过一段时间的Pandas第一次跑PySpark时连分区是什么都没概念但按先搞懂数据在哪、怎么算、怎么容错这条主线走两三天就能独立写ETL了。真正需要你花时间的不是API记忆而是理解作业被切分成什么样子、哪些操作会触发真的计算、哪些操作只是记了一笔账。2. 解剖一次作业的执行链路Driver、Executor、Shuffle的真实分工2.1 从SparkContext到Task一次作业的完整旅程很多教程讲架构就画一张Driver和Executor的图但读者看完还是不知道作业是怎么跑起来的。我这里用旅程的方式说一遍。你写好的代码提交后最先启动的是一个叫Driver的进程它是整个作业的大脑。Driver里有一个SparkContext负责创建RDD、累积算子、把计算逻辑组织成一个DAG有向无环图。DAG生成之后DAGScheduler登场。它会把DAG按照依赖关系切成多个Stage。切分的判断依据很简单如果这一步操作不需要把数据重新洗牌比如map、filter就可以留在同一个Stage里连续执行一旦遇到reduceByKey、join这类需要把相同key的数据聚合到同一个节点上的操作就会产生一次shuffle这地方就成了Stage的边界。Stage切好之后TaskScheduler再把每个Stage拆成一个个Task每一个Task负责处理一个分区。这些Task会分发给集群里的Executor去执行。Executor是真正干活的进程每个Executor里按配置有几个CPU核心并行跑Task。整个过程里Driver只管调度和汇总结果实际的数据计算全部发生在Executor侧。理解这个分层抓Bug的时候你就知道该往哪儿看Driver挂了多半是代码逻辑或内存收集的问题Executor挂了多半是数据本身或者资源配置的问题。2.2 Shuffle整个Spark作业里最贵的那一步Shuffle这个词基础阶段怎么强调都不为过。它的意思是某个算子需要把上游不同分区里的数据按照某个规则通常是key重新组织让相同key的数据跑到同一个下游分区去。这个过程必然涉及网络传输因为数据原本分散在不同节点上。我见过太多新手把Spark想得太理想以为一切都是内存计算直到看到某个join跑了一个小时才意识到shuffle是最容易暴露问题的地方。它不只是网络IO还包括序列化、压缩、磁盘溢出写入。默认情况下Shuffle Write出来的中间文件还是要落本地磁盘的目的是让下游Task可以稳定拉取这也是Spark并非完全不落盘这句话的来源。想判断一个作业里shuffle重不重最直接的办法是看Spark UI上每个Stage的Shuffle Read和Shuffle Write大小。如果Shuffle Write动辄几百GB而整体数据量也就这么多那多半是代码里产生了不必要的shuffle。基础阶段的你要养成的习惯是凡是能让数据留在原地、减少搬运的操作优先用凡是需要全局重分配的先问自己一句——这个key真的需要那么细粒度吗2.3 部署模式怎么选local、standalone还是资源调度框架开发环境调试和正式环境提交用的部署模式完全不一样。本地学习期直接用local[*]就够了它会在你本机起一个SparkContext用当前机器的所有核心模拟分布式环境。注意local模式只是为了开发方便它没有真正的节点容错也不会有网络shuffle的瓶颈所以本地跑得飞快不代表上集群也能跑得快。真正进集群之后有两种主流选择一种是Spark自带的Standalone模式架构最简单Master和Worker开起来就能用适合内部小集群和教学环境另一种是交给统一的资源调度框架来管理比如YARN或Kubernetes。用统一调度框架的好处是你同一套集群上可能既跑Spark实时流作业又跑别的计算任务资源能统一分配、按队列隔离。很多公司最终都走到这一步。选哪种主要看你现有的运维体系基础教程阶段不用纠结知道--master参数这几种填法就行部署模式master参数示例特点适合场景本地调试local[*]进程内模拟无网络开销开发、单元测试Standalonespark://host:7077部署简单自带资源管理中小型独立集群调度框架yarn/k8s://...资源统一管理多租户生产环境大集群无论哪种模式Driver进程都会占用一定资源--deploy-mode cluster时Driver本身也跑在集群内某个节点上方便看到完整日志开发期建议用client模式日志直接打在提交机器上排查问题更直观。2.4 分区数基础阶段最值钱的一个认知打开任何一份Spark作业你都能在UI上看到Executors、Tasks、Partitions这几个指标。分区数决定了你这个作业的并行度分区太少集群节点闲着没事干分区太多任务调度、序列化反而拖累性能甚至产出大量小文件。经验上有一个很粗糙的公式一个Executor核心配置2到3个Task比较合理所以总分区数大致等于executor数 × 单个executor的核心数 × 2~3。但是Spark里有两个独立影响分区数的参数别搞混了一是输入数据的切片粒度读HDFS时由文件块大小决定比如128MB一个块对应一个分区二是shuffle之后的分区数由spark.sql.shuffle.partitions控制默认是200。很多新手遇到数据倾斜就去调executor内存其实首先应该看两件事shuffle分区数是不是被盲目设成了几千上万输入数据读取时是不是因为文件压缩格式问题导致分区极度不均匀。这些判断不需要读懂Spark源码只要会看UI并按分区分布验证就能解决相当一部分性能问题。3. RDD、DataFrame、Dataset到底怎么选兼谈Catalyst优化器3.1 三个API的本质差异类型安全、执行优化和易用性初学者最容易在RDD、DataFrame、Dataset这三个名字之间纠结。先说结论今天写Spark业务代码90%的场景都该用DataFrame或者DatasetRDD只在你需要底层控制时才出手。RDD是我们前面反复提到的核心抽象它直接暴露map、filter、flatMap这类函数式接口操作的对象是Java或Scala对象。它最大的优点是灵活你可以在里面放任意自定义类的对象缺点也很明显——没有schema信息Spark看不出你每个字段到底怎么用所以很难做优化。DataFrame就是在RDD外面套了一层结构化信息相当于给数据加了列名和类型。有了这层表结构Spark的Catalyst优化器就能动刀了。比如你只需要两列Catalyst在读取时直接做列裁剪column pruning源数据里其他列根本不加载比如你过滤条件能下推它就尽量把filter压到靠近数据源的地方先执行。这些优化你自己手写RDD代码时基本享受不到。Dataset则是类型安全版的DataFrame在编译期就帮你检查字段类型对不对主要服务于Scala和Java用户。PySpark里其实没有严格对应的Dataset概念DataFrame就是主接口。所以如果你是Python党不必纠结Dataset重点把DataFrame和SQL用到烂熟就行。3.2 Catalyst优化器到底帮你做了什么Catalyst听起来高大上干的事其实很朴素你写的DataFrame代码会被它转换成逻辑执行计划再经过规则优化、物理计划选择最后生成可执行代码。几个核心优化要心里有数谓词下推把filter尽量往数据源推减少读入数据量。列裁剪只保留后续用到的列减少IO和内存占用。常量折叠where age 18 1这种表达式提前算成age 19。代码生成把整段表达式编译成高效字节码避免逐行解释执行。这也是为什么同一个逻辑你用DataFrame写会比用RDD手写快出一大截。Catalyst甚至能自动做部分join重排把大表和小表的join顺序调整成更合理的执行方式。所以基础阶段我强烈建议能用SQL表达的就不用算子能用DataFrame表达的就别退化到RDD。提示不要一看到RDD就觉得过时了。某些场景——比如自己实现一个特殊的图算法、或者扩展第三方解析逻辑——RDD的低层级控制依然不可替代。只是它应该是你的最后选择而不是默认选项。3.3 从groupByKey到reduceByKey算子选择为什么能决定作业生死算子这块所有教程都会提transformation是惰性执行、action才会真正触发计算但真正让新手栽跟头的不是这个而是算子之间巨大的性能差异。最经典的例子就是groupByKey和reduceByKey。groupByKey会把同一个key对应的所有value原封不动地聚到一个内存里再交给下游处理。如果某个key下面挂了上百万条记录这一下就是内存灾难。reduceByKey则聪明得多它在shuffle之前先对每个分区内的同key数据做一次本地预聚合把能减掉的数据量先减掉再跨节点传输。我帮A同学排查过一个案例他写了一个日志聚合作业按用户ID把行为序列收集起来用的是groupByKey结果一个executor动不动OOM。改成reduceByKey对行为做初步合并、再在map侧做二次聚合之后Shuffle Write数据量直接从几十GB掉到几个GB。这就是选错算子带来的量级差异。类似的还有join时要不要广播小表默认情况下Spark会做shuffle join但如果一个小表只有几十MB用broadcast把它复制到每个executor内存里就能整个省掉一次大shuffle。Spark 3之后Adaptive Query ExecutionAQE会自动做很多这类决策但你仍然应该理解背后的原理——自动优化不是万能的它也需要你提供合理的初始配置。4. 从第一个能跑的WordCount到提交集群完整上手流程4.1 最小WordCount的代码拆解理论再丰富不亲手动一个作业等于没学。我用PySpark给你写一个最标准的WordCount这是几乎所有Spark教程的第一个例子但它足够让你看清几个关键机制from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(WordCountDemo) \ .master(local[*]) \ .getOrCreate() lines spark.read.text(hdfs:///data/input/words.txt).rdd.map(lambda r: r[0]) words lines.flatMap(lambda line: line.split( )) pairs words.map(lambda word: (word, 1)) counts pairs.reduceByKey(lambda a, b: a b) # action触发计算 results counts.collect() for word, cnt in results: print(f{word}: {cnt}) spark.stop()这段代码里藏着几个基础知识点。第一flatMap和map的区别map一条输入对应一条输出flatMap一条输入可以炸出多条输出所以切词用的是flatMap。第二reduceByKey就是前面说的带预聚合的算子word count用它而不是groupByKey是刻意为之。第三collect()会把所有结果拉到Driver端在Demo里没问题生产环境绝对不要这么干——数据量一大Driver内存直接爆炸正确做法是counts.write.save(...)写到目标存储。4.2 提交作业时的参数到底怎么填在本地IDE里跑通只是第一步作业最终是要提交到集群的。用一个最小提交命令来做参数说明spark-submit \ --master yarn \ --deploy-mode cluster \ --num-executors 20 \ --executor-cores 4 \ --executor-memory 8g \ --driver-memory 2g \ --conf spark.sql.shuffle.partitions200 \ wordcount.py这里每个参数都值得琢磨。--num-executors填20--executor-cores填4意味着你有80个并发Task能力再结合shuffle分区200来看就大致吻合分区数核心数×2~3倍的经验值。--executor-memory 8g指的是每个executor堆内内存这个值不是拍脑袋定的要结合你数据单个分区的大小和算子复杂度来估算后面我会详细说。--conf可以覆盖一堆运行时参数前面说的spark.sql.shuffle.partitions就是最常见的。还有一个容易被忽略的参数是spark.executor.memoryOverhead它代表executor堆外内存如果作业里有大量自定义序列化或者使用了很多第三方库这个值不够会报奇怪的OOM。提示调参时一次只动一个变量。很多人上来把executor、分区、内存全改一遍跑通了也不知道是哪个参数生效跑崩了更不知道是哪个参数导致的。我见过最稳的做法是先按默认参数跑一遍记录Spark UI上的执行时间和shuffle量再逐项调整并对比。4.3 提交之后先学会看Spark UI作业一旦跑起来你最高频使用的工具就是Spark UI默认端口4040。不要觉得看UI是运维的事开发不懂UI调优无从谈起。你在UI上最需要关注四个地方一是Jobs列表看当前有几个作业、哪个阶段最耗时二是Stage详情看某个Stage内Task的时间分布如果绝大多数Task一分钟跑完就有一两个Task跑了十几分钟这是数据倾斜的典型信号三是Shuffle Read/Write指标前面已经说过它是判断shuffle开销的最直观数据四是Executors页面确认executor有没有频繁挂掉重建以及GC时间是不是高得离谱。一个健康的作业Task执行时间分布应该相对均匀GC占比在个位数百分比较常见Shuffle大小跟数据总量在一个合理的数量级。如果你看一个作业某个Stage绝大多数Task都很快、只有几个Task特别慢先别急着调内存大概率是分区数据不均匀或某些分区遇到了昂贵的计算路径这种问题靠调高executor内存往往只是掩盖症状。5. 内存参数与数据倾斜新手最常踩的两个坑5.1 Spark内存模型不是玄学先知道内存花在哪了所有OOM问题的排查都该从内存到底是怎么被划分的开始。Spark 3.0之后Executor的内存主要分两大块执行内存ExecutionMemory和存储内存StorageMemory共享一块统一的区域默认比例大概是各占一半由spark.memory.fraction0.6决定总可用比例。执行内存管的是shuffle、join、聚合这类中间数据存储内存管的是缓存、广播变量这类数据。理解这两个区域的竞争关系很重要。如果你大量使用persist(StorageLevel.MEMORY_ONLY)缓存中间结果存储内存占比高shuffle阶段可能就没内存用了执行内存会反过来把缓存数据挤出内存。Spark的应对机制是执行内存需要时可以直接抢占存储内存的已缓存数据让它们溢写磁盘。所以不要无脑缓存——缓存本身是有代价的被驱逐的数据写磁盘反而增加IO。OOM有三种常见表现对应三种不同病因报错/现象真正原因常用处理方向Executor Lost / OOM单分区数据过大或算子内存爆炸增大executor内存、增加分区数、换预聚合算子Driver OOMcollect()收集了太多数据改用写存储必须小结果才允许collect堆外内存报错overhead设置不足提高spark.executor.memoryOverhead5.2 数据倾斜先识别再对症下药如果说内存是新手第一道坎那数据倾斜几乎就是第二道。表现形式前面提过一堆Task秒跑完一两个Task跑半天甚至直接把executor拖死。根因通常是一两个key的数据量远超其他key导致分配到这些key的分区成了瓶颈。对症下药的套路按实际操作频率排序有五招第一招提高shuffle分区数让倾斜的那个key不至于独占一个分区。治标但有时立竿见影。第二招在倾斜key上加随机前缀加盐把一个大key拆成多个小key分散到多个分区算完再合并。第三招广播小表把join里等值key的那张小表广播到每个executor彻底避免因为大表key倾斜造成的shuffle爆炸。第四招把倾斜key单独拆出来处理统计一下哪些key是巨无霸把这些key的数据过滤出来后走单独的join或聚合逻辑剩下的正常处理最后union回来。第五招如果用的是Spark 3以上记得把AQE打开默认开启它提供的coalescePartitions和skewJoin能自动把过小的分区合并、把倾斜的join拆分。A同学前几天处理一个订单表关联用户维度的作业订单表里有个默认用户ID占了几千万条普通join下那个分区直接OOM。我们用第四招把默认用户ID单独拆出来做filter broadcast关联主体数据走正常join最后union合并整个作业从半小时缩到三分钟。关键点在于先统计确认了倾斜key而不是瞎调参数。5.3 小文件问题结合作业设计提前预防当你的Spark作业产出写入HDFS这类分布式存储时最容易出现的次生灾害就是大量小文件。原因很简单写文件的Task数太多或者shuffle分区数太大每个分区写出来只有几十KB到几MB远远小于128MB的块大小。小文件多会带来两重代价一是后续读取作业因为要处理大量小块启动大量Task调度开销暴涨二是NameNode上文件元数据压力变大整个集群的性能都会被拖累。预防思路有三个写之前用repartition(n)或coalesce(n)控制输出分区数按目标文件大小反推n写之后如果没有严格的分区约束需求用OPTIMIZE这类操作合并或者干脆在分区表场景下把输出按业务分区date或category组织每个业务分区单独控制文件数。这套知识并不复杂但在基础阶段养成写出文件前先想分区数的习惯后面进生产环境你会感激自己。最后再分享一个我自己的习惯学Spark基础别急着追新特性先把一个作业从提交到产出全过程中每一阶段数据量变化这件事画出来。我每次接手新集群或新作业都会在Spark UI上截图记录每个Stage的Shuffle大小、耗时、GC占比攒几份之后你就会形成自己的体感看到数据量你就能大概预估它该跑多久看到某个指标不对你就能快速定位是shuffle问题、倾斜问题还是内存问题。这种能力不是靠刷文档刷出来的就是靠一个个作业跑出来的。找一个你手上最常用的日志处理需求用PySpark或Scala把它从单机搬到集群再对着UI逐段优化一遍比看十篇教程都有用。