1. 为什么大家都在聊Spark先搞清楚它到底是什么做数据这行只要你的数据量到了每天几GB、几十GB甚至TB级别Spark这个名字就绕不开。早期很多人从Hadoop的MapReduce入门辛辛苦苦写一堆Java代码跑词频统计跑一个中型数据集要等上十几分钟甚至半小时写任务像做批改作业改一行逻辑要重新提交整个任务。直到有人把Spark的示例丢过来同样的逻辑同样的机器跑完也就一两分钟那种震撼感我到现在还记得。Spark是一套统一的大数据处理引擎核心卖点是内存计算和DAG调度。简单说它把计算过程中的中间结果优先放在内存里而不是像旧方案那样做完一步就写一次磁盘所以迭代类、多阶段类任务的速度提升非常明显。它并提供了一组非常易用的高级API覆盖Java、Scala、Python、R等多种语言底层统一走同一套执行引擎。再加上它天生支持批处理、流计算、图计算和机器学习这几个场景你不需要为不同需求上多套系统一套框架里就能解决大部分问题。这篇文章适合谁如果你是刚接触Spark想搞明白RDD、DataFrame、DAG这些名词到底是什么意思如果你已经能写出简单的Spark任务却总在提交时踩中内存、序列化或数据倾斜的坑又或者你是团队里负责搭建数据处理方案的人想快速判断Spark到底适不适合你的场景——那我这篇总结应该能帮你省掉不少摸索时间。我下面的内容会直接讲架构、代码和排障经验不绕弯不会用什么花哨的黑话。2. 核心架构与运行原理Driver、Executor和DAG是怎么配合的2.1 角色分工谁在管调度谁在干苦力先画个人物关系图我尽量用大白话描述一个Spark集群在跑任务的时候集群里会同时存在两类角色一个是Driver一个是Executor。Driver是整个任务的“大脑”。它负责把你的代码翻译成一个执行计划算清楚有多少个任务、任务之间有什么依赖再把这些任务分发下去。它还负责跟踪每个任务跑到了什么状态最后把结果汇总回来。如果Driver进程挂了整个应用直接失败非常脆弱。Executor是实实在在干活的“工人”。它启动后会在自己的JVM里执行Driver分配过来的任务。每个Executor可以并发跑多个Task所谓并发度取决于你给它分配了多少核心。Executor之间的数据互相不共享各自有独立的内存区域。这也带来一个结果不同Executor之间要做数据传输时必须走“洗牌”流程也就是shuffle把数据重新分区这个过程往往是整个作业最吃资源、最容易出问题的地方。从下图上理解这个模型文字形式Driver在最顶层往下连接了多个Executor每个Executor里又挂着多个TaskTask处理的数据块来自集群里的分布式文件块或内存缓存。你可能会问为什么非要有Driver和Executor的区分不能直接每个节点各跑各的吗因为有“全局视角”的需求。你要做聚合、排序、join这类操作必须依赖一个中枢节点来确定“谁的数据发给谁”。Driver就是做这个中枢的。这个模型也带来了一个隐含约束Driver的内存和网络带宽可能会成为大规模任务的瓶颈所以绝对不要在Driver里用collect把海量结果拉到本地。2.2 RDD、DataFrame与Dataset三种数据抽象到底怎么选接触Spark一定会先遇到RDD这个概念全称是弹性分布式数据集。我第一次学时看了好几篇文章名字太绕我差点以为是什么高端数据结构。后来想通了它其实就是“一个分区的集合”每个分区可以放到不同的节点上Spark对这个集合做各种变换操作时会自己跟踪整个演变过程一旦某个分区丢失可以靠血缘关系重新算出来这就是所谓“弹性”的含义。RDD提供了一大堆底层转换算子像map、flatMap、filter、reduceByKey这些非常灵活。但是它的坏处也很明显没有任何schema信息Spark不知道每个元素里面是什么结构所以想做优化也无从下手。而且你用RDD写代码处处要操心序列化、分区函数、内存缓存策略很容易写着写着就变成性能调优师了这显然不适合大多数人。所以Spark后来推出了DataFrame。你可以把它理解成一张分布式的表每一行就是一条记录每一列都有明确的类型。有了schema之后Spark的优化器就能自动做一些列裁剪、谓词下推、甚至重新排列join顺序的优化。我见过不少同学把RDD代码改写为DataFrame后什么都没调性能凭空快了两三倍这就是引擎优化的价值。Dataset是我跟你说的进阶项。在Scala/Java里Dataset是强类型的编译期会帮你检查字段是否写错。在Python里DataFrame其实就是Dataset[Row]的别名所以你在PySpark里基本不用考虑Dataset的事情。我的建议是新任务一律优先用DataFrame API只有遇到特别复杂的自定义逻辑、优化器搞不定的时候再考虑用RDD。这个选型方案是我跑了几年业务后最省心的答案。2.3 DAG调度为什么Spark能快起来DAG中文叫有向无环图。这大概是Spark最核心的设计之一。你的代码在Driver端会被解析成一系列transform操作这些操作不会被立刻执行而是形成了一个计算图节点是每一步操作边是数据依赖关系。这里最关键的概念叫做惰性求值。惰性求值的意思是像map、filter、flatMap这类转换操作并不会真正去读数据、做计算它们只是记录“我要做什么变换”。只有遇到action操作比如count、collect、saveAsTextFileSpark才会真正划分阶段并启动任务去跑。这个设计很巧妙它让Spark能看到整张计算图然后做出全局优化。比如你连着写了几个map操作Spark可以把它们合并成一个阶段里的连续管道数据在一个任务里一路流下去不用反复落盘。还是以WordCount为例来说明textFile读入文件产生一个RDDflatMap压平map变成键值对reduceByKey做聚合。这四步在图上会形成一个计算链条。DAG调度器会把这个链条从reduceByKey处切开前面属于宽依赖要先完成shuffle才能进入最后的聚合阶段。我见过很多新手想当然认为“代码写了4步就要分4个阶段”其实它只分了2个阶段这就是DAG合并的效果。理解这个机制后你再看“为什么Spark比MapReduce快”就能抓住重点了MapReduce每个map和reduce之间必须落盘中间结果还要复制到下一阶段天然就是重的。而Spark尽量把所有中间数据放内存加上DAG的管道化大量算子能在一个任务中连续执行不必要的I/O开销就省掉了。3. 手把手搭建本地Spark运行环境新手也能快速跑起第一个任务3.1 环境准备其实就三个东西我建议初学者先在本地搭一套环境至少不用申请集群资源改代码验证逻辑特别方便。本地模式不需要分布式文件系统不需要YARN只需要三样东西JDK、Python如果你想用PySpark和一个Spark安装包。JDK必须装因为Spark源码是跑在JVM上的。安装后要配置JAVA_HOME环境变量最好装Java 8或Java 11这类长期支持版本。Python一般用3.8以上就行本地环境里Python的作用是提供PySpark的接口层真正干活的是底层的JVM进程。装好这些之后去Spark官网下载对应的安装包注意二进制包分为“Hadoop版本配套”的类型。即使你还没装Hadoop也要选一个带“Hadoop 3.x”字样的表示该版本编译时适配的Hadoop协议本地模式能正常用。把压缩包解压到某个目录比如/opt/spark或者D:\spark然后配置环境变量SPARK_HOME并加到PATH里就完成了。整个过程顺利的话十分钟内搞定。3.2 启动local模式并验证一条命令告诉你环境OK没有进入Spark安装目录的bin目录执行spark-shell如果你是Python用户执行pyspark看到控制台出现大段Spark标志以及一个scala或提示符说明环境没问题。在pyspark里可以顺手做个验证输入下面这段代码rdd sc.parallelize([1, 2, 3, 4, 5]) print(rdd.sum())能输出15说明从分布式转化到本地执行链路是通的。我第一次跑通的时候心里还嘀咕这也没啥神奇的啊其实这时候还没有真正涉及分布式计算只是验证了一整套环境栈能正常协作。本地模式的核心参数叫master在代码或shell里写local[2]这个2代表模拟2个线程来并行执行任务。你可以试着改这个数字会发现并发变大后即使同一份数据执行时的任务划分也不一样这对后面理解分区和并行度很有帮助。3.3 本地模式的适用边界别高估它的能力本地模式干活方便但必须说清楚它的边界。你本地电脑就一套内存和CPU即使设置local[N]也只是在进程里多开几个线程模拟并发真正的跑去分布式存储、多节点传输、动态资源申请是不可能的。所以本地模式适合用来做开发自测、调试逻辑、小数据量验证不适合拿来做真实性能测试。我见过有人用本地模式测了一个几GB数据的任务发现挺快于是就把同样配置搬上集群结果集群反而更慢。原因很简单本地模式下数据都在同一台机器shuffle只在内存里完成网络开销几乎为零而上了集群网络带宽、磁盘竞争、调度延迟都成了变量。所以本地模式的测试结果只能用来验证功能和逻辑不能用来预言集群性能。4. 从WordCount到核心算子拆解看懂这几行代码能省不少弯路4.1 WordCount的完整代码与运行效果WordCount是数据框架的“Hello World”我拿PySpark版来拆。先用spark-shell的完整语法来写更贴近真实提交任务from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(WordCount Demo) \ .master(local[2]) \ .getOrCreate() lines spark.sparkContext.textFile(file:///tmp/input.txt) words lines.flatMap(lambda line: line.split( )) pairs words.map(lambda word: (word, 1)) counts pairs.reduceByKey(lambda a, b: a b) results counts.collect() for word, cnt in results: print(f{word}: {cnt}) spark.stop()注意两件事。第一SparkSession是统一入口老教程里还在用SparkContext直接起步现在更推荐用SparkSession它内部已经帮你管理了SparkContext。第二textFile读数据时如果数据在本地前面要加file://前缀如果数据在HDFS上写hdfs://路径即可。我经常看到有人直接写本地路径结果程序报FileNotFound其实就是协议头没写。collect这个action会把所有结果拉到Driver端。上面这个例子我故意用collect因为方便打印。但如果你处理的数据集是几十GB甚至更大绝对不要collect它会直接撑爆Driver内存。生产环境应该用saveAsTextFile或写入数据库让结果落到持久化存储里。4.2 核心Transformations逐个拆解下面是我认为新手最应该先理解的几个核心操作也是日常开发用得最多的几个。flatMap把每一个输入元素映射为0到多个输出元素且输出会被压扁成一个新集合。拿上面例子来说如果一行文本是“hello world”flatMap后变成两个单词“hello”和“world”如果是空行就什么都不产生。map一对一的变换每个输入产生一个输出。例子中的(words)变成(word, 1)就是在每个单词外面套了一层元组。filter把结果为False的元素丢弃。比如你想去掉空字符串可以写words.filter(lambda w: len(w.strip()) 0)。reduceByKey按key做局部预聚合相同key的数据会先在一个分区里合并然后再做全局合并。这个过程涉及shuffle并行度有讲究。数据里的不同key数量很大时shuffle的量也会很大所以如何调整分区数通常是个性能瓶颈点。distinct和groupByKey也是高频操作。但groupByKey在设计上会把某个key对应的全部value都汇集到一块如果某个key特别大倾斜问题立刻暴露。能用reduceByKey先聚合的优先用reduceByKey别用groupByKey硬扛。4.3 阶段划分与shuffle你写的代码在集群里到底经历了什么写完代码提交后Driver会先构建DAG。遇到shuffle算子如reduceByKey、join、distinctDAG调度器会在它跟前切出一个stage。stage之间是宽依赖意思是某个分区的数据要传输给多个下游分区这个环节必须等前一个stage完全执行完。拿WordCount的DAG来说整个流程被切成两个stage。第一个stage包括textFile、flatMap、map这些不需要shuffle的算子每个Executor在自己的分区里完成本地转换。第二个stage从reduceByKey开始先把各个分区的相同key数据通过网络重新汇聚到一组新分区上再由新分区做最终聚合然后才是saveAsTextFile输出。这个阶段数目直接决定作业的执行时长。你可以在Spark UI的Jobs页面看到每个作业拆成了几个Stage点击任意Stage能看到对应任务的执行时间、Shuffle读写大小、本地性情况。我很建议你在本地跑一次代码打开浏览器看一次Spark UI它比任何文字教程都直观。5. 常见问题与排查技巧实录三个高频坑一次性讲完先说结论Spark新手遇到的报错七八成是这三类——内存溢出、Shuffle阶段偶发失败、任务数据严重倾斜。下面我逐个说现象和排查方向。5.1 Executor内存溢出报错就去找内存配比现象是执行过程中某个Executor直接崩掉日志里出现OutOfMemoryError或者看到“Container killed by YARN for exceeding memory limits”。原因主要有两种一是JVM堆内内存不够二是堆外内存超了。堆内内存一般通过配置spark.executor.memory来控制如果你的任务里缓存了大量数据集到内存就要合理加大这个值如果你处理的是超大join或groupBy也可能需要在shuffle时把spark.shuffle.memoryFraction调大。不过从较新的Spark版本开始内存管理改成了统一内存模型静态调memoryFraction的方式已经不太使用更应该关注的是spark.executor.memory、spark.executor.memoryOverhead这两个参数。我建议你在排查时看Spark UI里每个Stage的Shuffle Spill信息。如果显示大量数据溢出到磁盘说明内存给少了或者分区数不合理。如果每个任务的内存都正常但整个TaskSet就是跑不动那大概率是Executor并发太多导致每个Executor分到的可用内存太少。调并发度的方法一是调spark.executor.cores二是调spark.sql.shuffle.partitions默认是200把它调整到集群能承担的数量。5.2 数据倾斜同一个任务其他Executor都跑完了就它在磨洋工数据倾斜是我这三年里遇到最多、也最恼火的问题。现象就是在某个Stage里大部分Task秒级完成剩下一个Task要跑十几分钟甚至更久整个作业被拖死。原因是某些key的量大得离谱比如空值key、热点用户ID、超大分区字段所有数据都压到一个分区去处理。几种常见解法我对齐整理一下定位先在Spark UI里看Stage各个Task的处理时长和Shuffle Read量如果某个Task的Shuffle Read明显高于中位数基本就是倾斜了。定位到倾斜key后最简单的办法是过滤掉异常数据比如把空值、无意义key直接过滤或单独处理。如果倾斜key是热点但没有异常可以尝试加盐salting把原来单一的key随机拆成多个子key比如key加随机数后缀分成10份这样所有数据会被分到更多分区去计算最后再聚合一次去掉盐。代价就是多一轮聚合但总时间往往能降一个量级。还可用广播变量如果一个表很小另一个表特别大做join的时候可以把小表广播到每个Executor避免shuffle。这样大表每个分区本地就能完成匹配速度提升非常明显。这是我处理join类倾斜的首选方案。5.3 Shuffle阶段FAILED网络和磁盘背锅现象是报错信息里看到ExecutorsLostFailure、FetchFailedException。这种情况多半发生在shuffle拉取阶段某个Executor数据拉不到原因可能是节点崩溃、网络抖动、或者磁盘满了。排查方向先看失败的Executor日志是网络原因还是磁盘原因。如果是磁盘空间不足清理临时目录如果是节点崩溃确认是不是因为之前某个Executor内存溢出导致被YARN杀掉那就又回到内存排查那条路。另外把spark.shuffle.service.enabled打开并启用外部shuffle服务可以减少Executor动态变化时shuffle文件失联的概率。5.4 调试小技巧把一个作业分成几段验证最后分享一个调试习惯不要写一个几百行的Spark作业然后指望一次跑通。我习惯先把核心链路用最小数据量跑通确认每一步输出的schema、样例数据都符合预期再把数据量逐步放大。在PySpark里我经常用show()先看前20行用take(5)看数组或者用explain()查看SparkSQL的执行计划确认优化器干了什么。如果你在开发环境经常提交、被杀、再提交我建议用spark-submit时开启--deploy-mode client模式这样Driver跑在本地日志直接打到控制台排障比cluster模式直观得多。等到代码稳定了再切回cluster模式让Driver跟着应用走避免本地断网或关电脑导致任务失败。6. 分享一点我个人的实践体会如果你今天刚接触Spark千万别被那一大堆配置参数吓到。刚开始就记住几条最核心的优先用DataFrame API、本地用小数据集验证逻辑、提交任务前在UI上看一眼Stage划分、遇到性能问题先怀疑数据倾斜而不是盲目调内存。这套思路能让你避开我当年踩过的大部分坑。还有一个小技巧是在调试shuffle相关问题时很管用的把spark.sql.shuffle.partitions从默认的200调小先观察单个Task的数据量是否会暴增。如果暴增了说明当前分区数偏少shuffle后的每个分区数据量过大如果没暴增但总耗时还是没变化那你就要往数据倾斜的方向查了。这个动作虽然简单却经常能帮我在几分钟内判断出问题的方向。Spark的学习曲线确实比单纯写业务代码要陡一点但只要你亲手跑通一个作业、亲手在UI里观察一次Stage划分、亲手被一次OutOfMemoryError折磨过并把它解决掉这套东西基本就长在你身上了。接下来你可以去折腾SparkSQL、Streaming和MLlib它们底层的原理都还是一样的Driver、Executor和DAG那套故事。