先放结论：Spark不是用来替代SQL的，而是用来替代那些单机处理不了、又不想写一堆MapReduce的分布式计算场景。我见过太多人一上来就背API，结果连“作业为什么慢”“Executor为什么OOM”都说不清楚。这篇教程我会按自己实际跑生产链路的经验来写，从定位、架构、环境、算子，到调优、踩坑，一次讲透。
我是从接手一个每天处理几TB流量日志的离线任务开始系统用Spark的，前后折腾了一年多。当时最痛苦的不是写不出逻辑，而是写出来的作业要么跑不动、要么动不动就Executor丢失。后来我把Spark的源码层面机制和参数一个个试过来，才慢慢摸清楚它的脾气。这篇文章就按我自己的学习路径组织，没有教科书式的废话，全部以“能跑、能调、能排查”为目标。
1. Spark到底解决了什么问题：先看清它的定位
1.1 从MapReduce的痛点说起
很多人不理解Spark为什么能“快”，其实快是相对Hadoop MapReduce而言的。MapReduce把每个计算步骤拆成Map和Reduce两个阶段，中间结果必须落到磁盘，下一阶段再去读。你要是写一个多步骤的ETL，每一步都要落盘一次，I/O开销非常夸张。我记得第一次用MapReduce跑一个六步清洗任务，单日数据量也就几十GB，却跑了近两个小时，大部分时间都耗在读盘和写盘上。
Spark的思路是：尽量把计算留在内存里，中间结果能缓存就缓存，能不落盘就不落盘。它把一系列操作构建成一张DAG（有向无环图），在行动算子触发后统一调度执行。同样那个六步清洗任务，换成Spark后首跑大概二十分钟，后续因为数据能复用甚至更快。这个差距在数据量越大的场景下越明显。
1.2 Spark的定位：一个统一的计算引擎
Spark不是数据库，也不是存储系统，它本身不存数据，只是计算引擎。它能读HDFS、对象存储、本地文件、JDBC数据源，也能读Kafka做流式处理，然后计算结果写到各种目标。这意味着你原来存储在HDFS或数据仓库里的数据，都可以直接用Spark来做批量计算、特征工程、报表加工。
它还给了你一套统一的API，支持Scala、Java、Python、SQL。很多团队是这种搭配：离线ETL用Spark SQL，复杂特征逻辑用PySpark，实时链路用Structured Streaming，一套框架覆盖大部分需求，不用在多个引擎之间切换。这也是很多数据平台选它做离线计算底座的核心原因。
1.3 哪种场景不要用Spark
光知道Spark能做什么还不够，还得知道它不适合什么。Spark定位是批量计算和准实时处理，不是低延迟OLTP查询。你要是想做一个线上接口，要求几百毫秒返回结果，那应该用关系型数据库或者OLTP引擎，而不是Spark。另外，如果数据量就几百MB，单机Pandas完全能处理，引入Spark反而增加运维成本和调度开销。
我做过的判断标准大概是：单机内存能轻松装下且计算逻辑不复杂，就不上Spark；数据量上TB、需要分布式计算资源，或者需要跑超过一小时的批处理，才值得用Spark。学会判断边界，能省掉很多不必要的复杂度。
2. Spark的骨架：Driver、Executor、RDD与Stage调度
2.1 两种角色：Driver和Executor
Spark集群运行作业时，角色划分其实很清晰。Driver是你的作业“大脑”，负责解析代码、生成执行计划、调度任务、汇总结果。Executor是真正干活的“工人”，分布在集群各个节点上，负责执行具体的task，并把结果返回给Driver。
这个划分直接影响你后面的排错思路。Driver挂了，整个作业失败；Executor挂了，Spark会尝试重新调度失败的任务，但如果频繁Executor丢失，就要考虑资源或内存设置。我在生产环境看到过不少案例，Driver内存给太小，日志直接显示“Driver堆内存溢出”，连调度都完成不了。所以配置资源时，Driver和Executor要分开考虑。
2.2 RDD、DataFrame与Dataset的取舍
很多人刚学Spark时被三个数据抽象搞晕，我先给一个直观结论：RDD是最底层的抽象，DataFrame是带Schema的分布式表，Dataset是类型安全的DataFrame（主要在Scala/Java里体现价值）。实际工作中，Python开发者基本只用DataFrame；Scala/Gava开发者偶尔用Dataset做编译期检查；RDD更多用于底层API或需要细粒度控制的地方。
用PySpark写数据处理时，我会默认用DataFrame。原因是Spark SQL的Catalyst优化器能对DataFrame做很多自动优化，比如谓词下推、列剪枝；而RDD算子不具备这些优化，性能差距很大。有一段时间我帮同事review代码，他用RDD的map/filter处理几亿行数据，作业跑得很慢，改成DataFrame后速度提了不少，靠的就是Catalyst的优化能力。
2.3 DAG与Stage：一个作业在集群里到底怎么跑
这是理解Spark性能的核心。你写的转换操作会被记录成血缘关系（lineage），构成DAG；行动算子触发后，Driver会把DAG拆分成多个Stage，Stage之间以Shuffle为边界。
Shuffle是什么呢？我用一个快递分拣的例子来解释：假设每个快递员手上有一些包裹（分区数据），现在需要按省份重新分组，每个人必须把包裹送到对应省份的货架上，这个过程就是Shuffle。Shuffle会把所有数据重新洗牌、跨节点传输，开销很大。所以写Spark作业时，一个朴素的优化思路就是尽可能减少Shuffle次数，能一次算子搞定的不要拆成两步。
窄依赖（map、filter等）和宽依赖（groupBy、join等）的区别，就是判断是否需要Shuffle的关键。窄依赖每个分区只需从父RDD的少量分区获取数据，宽依赖则要跨很多分区重新分配，也是Stage划分的依据。理解这个机制后，你看到Spark UI上的Stage列表时，就能立刻判断哪些步骤贵、哪些便宜。
3. 本地搭环境：从下载Spark到首个作业跑通的完整流程
3.1 安装前的版本选择
开发学习阶段，我建议直接在你自己的电脑上搭一套local模式的Spark环境，没必要上来就搭集群。版本选择上，我用的是Spark 3.x配Python 3.8+，文件的API变化不大，但性能和改进明显好于旧版本。另外要注意，Spark对Java版本有要求，我用的是Java 8/11（Spark 3.2+默认支持Java 8/11/17，但生产环境稳一点选8或11）。
还有一个易忽略的点：Python 3.12等较新版本，可能和某些PySpark版本存在兼容性问题。我的建议是用Python 3.8到3.10之间，这期间踩坑最少。
3.2 下载与启动Spark
去Spark官网下载预编译的发行版，一般选择“Pre-built for Apache Hadoop 3.2 and later”这个选项，解压后放到指定目录。然后配置环境变量，把SPARK_HOME/bin加入PATH。
启动本地交互环境验证：
cd $SPARK_HOME ./bin/pyspark打开PySpark Shell后，敲一下：
sc.version能输出版本号，就说明环境没问题。你也可以在Shell里快速验证一个WordCount：
rdd = sc.textFile("README.md") rdd.flatMap(lambda line: line.split(" ")) \ .map(lambda word: (word, 1)) \ .reduceByKey(lambda a, b: a + b) \ .take(10)这一步能让你直观体会到RDD的转换和行动操作是怎么配合的。
3.3 写第一个独立脚本并用spark-submit提交
交互式Shell适合验证思路，生产作业往往是用独立脚本提交。我习惯用spark-submit统一提交。先写一个最简单的脚本：
from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("WordCount") \ .getOrCreate() df = spark.read.text("hdfs:///data/input/words.txt") words = df.selectExpr("explode(split(value, ' ')) as word") counts = words.groupBy("word").count().orderBy("count", ascending=False) counts.show() spark.stop()提交命令如下：
$SPARK_HOME/bin/spark-submit \ --master local[2] \ --name WordCount \ wordcount.pylocal[2]表示本地用2个CPU核心来模拟并行执行。如果你想看Spark UI，提交时给--master local[*]，然后访问http://localhost:4040，能看到作业的Stage、Executor情况，这是个很好的学习工具。
3.4 生产提交常用参数格式
等你要提交到集群时，参数会复杂很多。我贴一份生产环境常用的模板：
spark-submit \ --master yarn \ --deploy-mode cluster \ --name etl_daily_report \ --conf spark.executor.memory=4g \ --conf spark.executor.cores=2 \ --conf spark.executor.instances=20 \ --conf spark.driver.memory=2g \ --conf spark.sql.shuffle.partitions=400 \ --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \ s3://path/to/job.py这里每个参数都是有讲究的，后面调优章节我再展开。先记住：executor内存、executor核数、executor数量、driver内存、shuffle分区数，是最常调的五个参数。
4. 日常最常用的DataFrame算子：这些能覆盖八成需求
4.1 读取与写出：数据源格式决定一半效率
实际开发里，DataFrame最常打交道的是读写。读CSV时，我会指定schema而不是让它自动推断，尤其是大文件，自动推断会额外扫描一遍数据，性能损失很大。
from pyspark.sql.types import StructType, StructField, StringType, IntegerType schema = StructType([ StructField("user_id", StringType(), True), StructField("event", StringType(), True), StructField("cnt", IntegerType(), True) ]) df = spark.read \ .option("header", True) \ .schema(schema) \ .csv("s3://data/events/")写出时，coalesce和repartition是最容易混淆的两个算子。coalesce会尝试在减少分区时不发生全量Shuffle，适合把大量小分区合并成大文件；repartition则一定会重新分区，适合需要强制均衡的场景。我通常这样用：输出到数据仓库前，限定输出文件数量：
df.coalesce(8).write.mode("overwrite").parquet("s3://output/result.parquet")4.2 高频转换算子示例
以下是我在数据清洗任务中几乎每天都会用到的算子，配了实际用例：
select：只选择需要的列，上下游衔接时省内存。withColumn：基于已有列生成新列。filter：过滤条件行，常配合col函数。groupBy+agg：做分组聚合。join：多表关联，务必注意小表放左/右以及Join类型。dropDuplicates：去重，注意指定字段。
from pyspark.sql.functions import col, sum, when result = (df .filter(col("cnt") > 0) .withColumn("is_high", when(col("cnt") > 100, 1).otherwise(0)) .groupBy("user_id") .agg(sum("cnt").alias("total_cnt"), sum("is_high").alias("high_cnt")) .orderBy("total_cnt", ascending=False))这里有个细节：filter要尽量前置，越早过滤数据，后续算子处理的数据量越小。withColumn多次调用会生成新列，但不会修改原DataFrame，所以要用链式写法或重新赋值。
4.3 窗口函数与UDF
窗口函数在计算分组TopN、环比、累计值时非常有用。我用一个取每个用户最近一次事件的例子：
from pyspark.sql.window import Window from pyspark.sql.functions import row_number window_spec = Window.partitionBy("user_id").orderBy(col("event_time").desc()) latest_df = df.withColumn("rn", row_number().over(window_spec)) \ .filter(col("rn") == 1) \ .drop("rn")UDF要注意性能。普通的Python UDF会有序列化和逐行调用的开销，大数据量下比较慢。优先用内置函数和SQL表达式；实在要自定义逻辑，建议用Pandas UDF（pandas_udf）成批处理。
4.4 与SQL的互转
如果你的团队更习惯写SQL，Spark完全支持。createOrReplaceTempView之后就能用spark.sql查询：
df.createOrReplaceTempView("events") spark.sql(""" SELECT user_id, COUNT(*) AS cnt FROM events WHERE cnt > 0 GROUP BY user_id """).show()我个人的习惯是：复杂逻辑优先用DataFrame算子链，可读性差时转SQL，两者结合效率最高。过度追求一种写法，反而会让代码难维护。
5. 提升性能的四个突破口：内存、并行度、连接和缓存策略
5.1 先看Spark UI，再改参数
不要一上来就盲目调参。第一步永远是看Spark UI的Stage页面，确认瓶颈在哪个Stage、是计算慢还是Shuffle慢、是否有数据倾斜。我在排障时先关注三个指标：某个Stage的输入数据量、Shuffle读写量、单个Task的最大耗时。如果所有Task时间都长，说明数据量大或资源不足；如果只有个别Task异常慢，优先怀疑数据倾斜。
5.2 Executor内存与并行度的关系
Executor内存不能简单拍脑袋。一个Executor里能同时跑的Task数，由spark.executor.cores决定。比如设置executor.cores=2，那么一个Executor内同时最多跑2个Task。每个Task能分到的内存，大致是executor.memory / cores的某一部分（还要扣除一些Overhead）。如果Task需要的内存超过这个配额，就会触发GC频繁或直接OOM。
我常用这套估算思路：假设处理10GB数据，并行度为200，每个Task大概处理50MB，那每个Executor给4~8GB一般够用；如果数据分布不均匀，某个Task可能吃几百MB甚至更多，这时要检查是否有倾斜问题，而不是盲目加内存。
5.3 shuffle分区数：最常被忽视的参数
spark.sql.shuffle.partitions默认是200，但这个默认值在小作业和大作业下都容易不合适。小作业分区太多浪费资源，大作业分区太少导致单个Task处理数据量过大。
我的经验是：以每个Task处理100~500MB数据为参考来设分区数。比如你Shuffle后要处理20GB数据，分区数设在40~200之间比较合理。数据量波动比较大的任务，我习惯把分区数统一设为Executor总核数的2~3倍。如果使用Spark 3.x，可以考虑开启自适应查询执行：
--conf spark.sql.adaptive.enabled=true它会根据运行情况动态调整reducer分区数和Join策略，省去不少手动调优。
5.4 Join优化离不开广播变量
Join是Spark里最贵的操作之一，也是最容易出性能问题的环节。小表Join大表时，强烈建议用广播。广播会把小表发送到每个Executor节点，避免Shuffle。触发广播的条件是表大小小于spark.sql.autoBroadcastJoinThreshold，默认10MB。你可以手动调大或直接用Hint：
from pyspark.sql.functions import broadcast result = big_df.join(broadcast(small_df), "key", "left")最适合广播的就是维度表，比如几百MB以内的客户表、城市表。超大表之间Join必须Shuffle，这时尽量让两张表的数据在Join前就按相同规则分区，能减少Shuffle量。
5.5 persist与checkpoint的取舍
如果你要在同一个作业里多次复用同一个DataFrame，就该考虑persist，而不是每次都重新计算。默认StorageLevel是MEMORY_AND_DISK，内存不够会溢写到磁盘。如果你明确知道某个DataFrame只需用一次，那就不要persist，白白占内存。
checkpoint和persist不同，它会把数据实际保存到可靠存储（比如HDFS）并切断血缘。血缘链太长时，某个节点失败重算代价极大，用checkpoint可以提前备份关键中间结果。它的代价是写盘开销大，所以只在关键节点用，不要每个步骤都checkpoint。
6. 生产环境跑了一年之后，必须记住的这些坑
6.1 数据倾斜：先分辨是“真倾斜”还是“资源不足”
数据倾斜表现为某一个或某几个Task运行时间特别长，其他Task早就跑完了，整个Stage等这几个Task。我遇到过一次，一个分组聚合作业，数据里有个“默认渠道”字段，占了60%的数据量，导致所有数据挤在同一个Task上。
定位办法很简单：打开Stage详情页，看每个Task处理的Input和Shuffle Read数据量，如果出现一个Task读了几GB而其他Task只读几MB，基本可以判定是倾斜。
6.2 数据倾斜解法（一）：两阶段聚合
两阶段聚合是应对“热点Key”的经典办法。思路是先给Key加一个随机前缀，把它拆成多个Key做局部聚合，再去掉前缀做全局聚合。
from pyspark.sql.functions import concat, lit, rand # 第一阶段：加随机前缀，局部聚合 stage1 = (df.withColumn("salt", (rand() * 10).cast("int")) .withColumn("salted_key", concat(col("key"), lit("_"), col("salt"))) .groupBy("salted_key") .agg(sum("cnt").alias("partial_cnt"))) # 第二阶段：去掉前缀，全局聚合 result = (stage1.withColumn("key", regexp_replace(col("salted_key"), "_\\d+$", "")) .groupBy("key") .agg(sum("partial_cnt").alias("total_cnt")))随机前缀的粒度要控制好。前缀太少效果不明显，前缀太多小文件变多。一般按热点程度取10~100即可。
6.3 数据倾斜解法（二）：广播小表与过滤倾斜Key
如果倾斜是因为某个Key关联了一个巨大的小表，可以考虑把这个小表广播掉；如果倾斜Key本身就是脏数据、无意义数据，比如空字符串、Null、默认值，优先在计算前过滤掉。有一回我们链路里一个“未登录”用户占了一半数据，直接把这类用户单独处理，整体作业时间降了40%。
还有一个思路是“先过滤、再聚合、后合并”，核心思想是分别处理正常Key和热点Key，最后union。
6.4 Executor OOM：大多数时候不是内存不够
很多人遇到OOM第一反应是加内存，但我发现大多数时候OOM的根因不是内存总量不够，而是某个Task的中间数据暴涨，比如groupBy后某个Key的数据量特别大，或者collect把全量结果拉到Driver。前者按数据倾斜处理；后者要避免对超大DataFrame使用collect，改用take取抽样。
另外，当Executor上用了大量persist(MEMORY_ONLY)，内存不够时会出现反复GC。这时不如改成MEMORY_AND_DISK，把放不下的数据溢写磁盘，比反复GC稳定得多。
6.5 小文件问题：平台性能的隐形杀手
Spark写数据时，如果并行度很高，输出文件会很多，可能每个文件只有几MB。下游读这些数据时，打开文件、扫描元数据的开销非常大。我们之前写过一次任务，输出18000个Parquet小文件，下游Hive查询全表扫描跑了很久，改成coalesce(200)后文件数量降到200个，查询时间降了将近一倍。
控制输出文件数的通用做法是：写入前用coalesce减少分区（无Shuffle），或者用repartition按目标字段重新分区（有Shuffle）。如果写入的是分区表，尽量按分区粒度控制，比如每个分区一个文件。
6.6 其他容易踩的“小毛病”
- 序列化问题：闭包里用了不支持序列化的对象，作业在Driver端会报
Task not serializable。解决办法是避免在闭包里使用外部非序列化对象，或改用Kryo序列化。 - 时区问题：Spark默认按系统时区解析时间，如果上游数据是UTC，下游需要本地时间，要在spark-session里配好
spark.sql.session.timeZone，否则时间字段会差8小时。 - Parquet schema兼容：同一张表后来加了新字段，用旧schema读会报错或丢弃字段，最好用
mergeSchema=true来兼容演进。 - Spark.sql.shuffle.partitions改完之后没效果：确认是不是已经开了AQE，
spark.sql.adaptive.coalescePartitions.enabled会动态调整分区数，手动参数可能被覆盖。
6.7 最后一点运维经验：把Spark UI纳入日常习惯
我在每次作业跑完后，都会顺手翻一下Spark UI，看有没有Task卡顿、Shuffle量异常、GC时间过长的现象。这些日志级细节，往往比跑一遍EXPLAIN更能看出来真实问题。平时开发阶段多用explain()看执行计划，生产排查阶段多依赖Spark UI，两者结合，Spark作业基本跑得稳。
这套打法跑了一年后，我们离线链路的作业平均耗时降低了近一半，Executor OOM和倾斜导致的失败基本绝迹。如果你正在学Spark或刚接手Spark作业，我希望这篇教程能帮你少走几个月的弯路。按这个顺序去学，先懂原理再上手，碰到问题就知道往哪查。最后再分享一个小习惯：每个作业跑完，在Spark UI里截一张Stage耗时分布图存档，积累一段时间后回看，你会对自己的数据规律有更清晰的认识。
