企业官网建设流程全解析

很多朋友问我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参数这几种填法就行：

无论哪种模式，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或者Dataset，RDD只在你需要底层控制时才出手。

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 Execution（AQE）会自动做很多这类决策，但你仍然应该理解背后的原理——自动优化不是万能的，它也需要你提供合理的初始配置。

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.partitions=200 \ 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.fraction=0.6决定总可用比例。执行内存管的是shuffle、join、聚合这类中间数据；存储内存管的是缓存、广播变量这类数据。

理解这两个区域的竞争关系很重要。如果你大量使用persist(StorageLevel.MEMORY_ONLY)缓存中间结果，存储内存占比高，shuffle阶段可能就没内存用了，执行内存会反过来把缓存数据挤出内存。Spark的应对机制是，执行内存需要时可以直接抢占存储内存的已缓存数据，让它们溢写磁盘。所以不要无脑缓存——缓存本身是有代价的，被驱逐的数据写磁盘反而增加IO。

OOM有三种常见表现，对应三种不同病因：

5.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逐段优化一遍，比看十篇教程都有用。