Spark 3.5 完整知识体系：从入门到生产调优

---

一、先搞清楚Spark是什么、不是什么

很多人一开始学Spark就被一堆术语砸晕了——RDD、DataFrame、Dataset、Spark SQL、Structured Streaming……还没写代码呢，概念先劝退一半。咱们换个方式理解。

1.1 Spark的本质

Spark就是一个分布式计算引擎。你的数据太大，一台机器处理不完，Spark帮你把任务拆成很多小块，分发到集群里的多台机器上并行跑，最后把结果汇总回来。

它自己不存数据。数据存在HDFS、S3、OSS这些分布式存储里，Spark负责去读、计算、写回。你可以把它想象成一个超级计算器，接了无数个存储的插头，你要做的就是告诉它"算什么、怎么算"。

1.2 核心抽象：从RDD到DataFrame的演进

Spark诞生之初的核心抽象叫RDD（弹性分布式数据集）。RDD说白了就是一组分布在多台机器上的数据集合，你可以对它做各种转换操作（map、filter、reduce之类）。

但RDD有个大问题——它太底层了。你要告诉Spark每一步具体怎么算，比如要手写map函数、自己管理分区、自己操心性能。这对于写代码的人来说是个负担，对于Spark引擎来说也很难做自动优化。

所以在Spark 2.x之后，DataFrame和Dataset成为了主推API，到了3.5.x版本，这个趋势更加明显。现在生产环境里写Spark，90%以上都在用Spark SQL（DataFrame/Dataset API），RDD基本已经进博物馆了。

特性	RDD	DataFrame	Dataset
抽象层次	底层，手写逻辑	高层，声明式，无类型（运行时检查）	高层，声明式，强类型（编译期检查）
性能优化	手动优化，引擎帮不上忙	Catalyst优化器自动优化执行计划	Catalyst + Tungsten优化，与DataFrame共享执行引擎
类型安全	有类型但不完整，运行时才暴露问题	无编译期类型检查（Python/Scala都一样）	编译期就能发现类型错误（仅Scala/Java）
使用场景	遗留代码、极特殊需求	所有常规数据处理场景（Python唯一选择）	Scala/Java生产环境首选
代码量	多	少	少

三者的关系可以这样理解：RDD是地基，DataFrame和Dataset都建在地基之上。在底层实现上，DataFrame和Dataset最终都会被转换成RDD来执行（经过Catalyst优化后的RDD操作）。在Scala中，DataFrame实际上是Dataset[Row]的类型别名——当你写val df: DataFrame时，编译器看到的其实是Dataset[Row]。而Dataset[T]可以携带更具体的类型，比如Dataset[Person]，让你在编译阶段就能捕获类型错误。

说白了：新项目别碰RDD，直接用DataFrame/Dataset。

踩坑点：有些老教程还在大篇幅讲RDD，甚至让你用RDD写word count。学这个没用，面试都不问了，生产更用不上。直接上手DataFrame。

补充说明：Python开发者经常困惑——为什么资料里总说Dataset类型安全？因为在Python这门动态语言中，根本不存在编译期类型检查，所以PySpark只提供了DataFrame API（没有Dataset）。但这完全不碍事，PySpark的DataFrame一样能享受到Catalyst优化器的全部性能优势。Python开发者不必纠结"Dataset更好"的说法，专注用好DataFrame就行。

---

二、Spark SQL：你现在最该学的东西

2.1 DataFrame是什么

DataFrame就是带schema的分布式数据表。你可以把它理解成一个Excel表格，有列名、有类型，但数据存在集群里而不是你本地电脑上。

from pyspark.sql import SparkSession

# 创建SparkSession（这是3.x版本的唯一入口）
spark = SparkSession.builder \
    .appName("demo") \
    .master("local[*]") \
    .getOrCreate()

# 读一个JSON文件，自动推断schema
df = spark.read.json("people.json")
df.show()
df.printSchema()

SparkSession是Spark 3.x里唯一的入口。老版本里还有SparkContext，现在已经不需要你直接操作它了。

2.2 两种编程风格

Spark SQL支持两种写法，底层是一样的，看你喜欢哪种：

风格一：DSL风格（链式调用，更像代码）

# 筛选年龄大于21的人，只取name和age两列
df.filter(df.age > 21).select("name", "age").show()

# 分组统计
df.groupBy("age").count().show()

风格二：SQL风格（写纯SQL字符串）

# 先把DataFrame注册成临时视图
df.createOrReplaceTempView("people")

# 然后直接写SQL
result = spark.sql("SELECT age, COUNT(*) as cnt FROM people WHERE age > 21 GROUP BY age")
result.show()

两种写法性能完全一样，因为最终都会走Catalyst优化器。我个人推荐DSL风格，因为IDE能补全、能检查语法，写SQL字符串拼错了要运行时才报错。

2.3 Rdd vs Dataset vs DataFrame

这里有个容易混淆的点。在Scala和Java里，DataFrame其实是Dataset[Row]的别名，而Dataset可以带具体的类型，比如Dataset[Person]。到了Python里，没有Dataset这个概念，只有DataFrame。

语言	可用类型
Scala	DataFrame（无类型）+ Dataset（强类型）
Java	DataFrame + Dataset
Python	只有DataFrame

踩坑点：Python同学看到有些资料里讲Dataset的类型安全优势，不用慌，Python的DataFrame用起来一样爽。类型检查在Python这种动态语言里本来就不是强项，没必要纠结。

2.4 Catalyst优化器：Spark SQL的性能杀手锏

这是Spark SQL比RDD快很多的关键原因。你写了一条查询，Spark不会傻傻地按你写的顺序执行，而是先交给Catalyst优化器做一系列自动优化：

1. 逻辑优化：比如谓词下推（把filter尽量推到数据源端）、列裁剪（只读需要的列）、常量折叠
2. 物理优化：选择最优的执行策略，比如用broadcast join还是shuffle join
3. 代码生成：把优化后的计划生成JVM字节码，减少运行时的开销

这个过程中你什么都不用做，Spark自动搞定。这也是为什么说Spark SQL是"声明式"的——你只告诉它"要什么结果"，至于"怎么算最快"，Spark自己决定。

2.5 常用数据源

Spark SQL能读写的数据源非常多，生产环境最常见的几个：

# Parquet（列式存储，Spark最推荐的格式）
df = spark.read.parquet("s3://bucket/data/")
df.write.parquet("output/")

# ORC（Hive常用，压缩比高）
df = spark.read.orc("hdfs://namenode:8020/data/")

# JSON（日志类数据常用）
df = spark.read.option("multiLine", True).json("logs/")

# CSV（导入导出、对接Excel时常用）
df = spark.read.option("header", True).option("inferSchema", True).csv("data.csv")

# JDBC（直接读关系型数据库）
df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://host:3306/db") \
    .option("dbtable", "users") \
    .option("user", "root") \
    .option("password", "xxx") \
    .load()

# Kafka（流式场景，后面会讲）
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "host:9092") \
    .option("subscribe", "topic_name") \
    .load()

Parquet是首选格式。它是列式存储，Spark读取时可以只加载需要的列，压缩率高，而且自带schema信息。做数仓的话，数据尽量存成Parquet。

踩坑点：读CSV的时候，inferSchema=True会自动推断列类型，但第一次会扫一遍文件，性能慢。如果是生产任务，建议手动指定schema，或者先infer一次把schema存下来复用。另外CSV不存null值信息，空字符串和null分不开，这点很坑。能不用CSV就不用。

2.6 常用Transform操作（Transformation vs Action）

Spark的操作分为两类，这个概念必须理解清楚：

• Transformation（转换）：惰性求值，不会立即执行。比如select、filter、groupBy、join、orderBy。
• Action（动作）：触发真正计算。比如show()、collect()、count()、write.save()。

# 这些都是Transformation，只记录执行计划，不会真的跑
df2 = df.filter(df.age > 21)           # 过滤
df3 = df2.select("name", "age")         # 选列
df4 = df3.groupBy("age").count()        # 分组统计

# 这是一个Action，到这里才会真正触发计算
df4.show()

这个设计叫做惰性求值（Lazy Evaluation），好处是Spark可以先收集完整的执行计划，然后统一优化。但也带来一个坑——如果你的代码报了错，堆栈信息可能指向的是Action那一行，而不是真正出问题的那一行Transform。

踩坑点：很多人喜欢在中间步骤加show()或count()来"看看数据对不对"。这在调试时可以，但生产代码里千万别这么干——每个Action都会触发一次完整的计算，中间加Action等于把流水线切成好几段，性能暴跌。

2.7 Join操作与优化策略

Join是生产环境最耗性能的操作之一，也是最容易出问题的地方。

# 普通的join
df1.join(df2, df1.id == df2.user_id, "inner")

# 左外连接
df1.join(df2, df1.id == df2.user_id, "left")

# 多个条件join
df1.join(df2, (df1.id == df2.user_id) & (df1.dt == df2.dt), "inner")

Spark 3.5.x在Join优化上做了很多事，有几个你需要知道的：

Broadcast Hash Join（广播Join）：

如果一张表很小（默认小于10MB），Spark会自动把这张表广播到所有Executor的内存里，然后在本地做Hash Join，避免Shuffle。这是最快的Join方式。

# 强制使用broadcast join
from pyspark.sql.functions import broadcast
df_large.join(broadcast(df_small), "join_key")

Sort-Merge Join（排序合并Join）：

两张表都很大时的默认策略。两边先按join key排序，然后归并。性能比Shuffle Hash Join稳定，是Spark 3.x的默认选择。

踩坑点：如果你的表很小但没触发broadcast join（比如表大小刚好超过阈值，或者统计信息不准），性能会差几十倍。可以手动调spark.sql.autoBroadcastJoinThreshold（默认10MB，单位是字节），或者明确写broadcast()hint。反过来，如果一张表其实很大但你强行broadcast了，会直接把Executor内存打爆，任务挂掉。

2.8 窗口函数（Window Function）

窗口函数是数据分析的利器，在Spark SQL里完全支持。典型场景：分组TopN、累计求和、移动平均。

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, sum, avg

# 定义窗口：按category分区，按amount降序排列
window_spec = Window.partitionBy("category").orderBy(col("amount").desc())

# 每个品类取金额Top3
df.withColumn("rn", row_number().over(window_spec)) \
  .filter(col("rn") <= 3) \
  .show()

# 累计求和（从分区起始到当前行）
window_cumsum = Window.partitionBy("user_id").orderBy("dt").rowsBetween(Window.unboundedPreceding, Window.currentRow)
df.withColumn("cumsum_amount", sum("amount").over(window_cumsum)).show()

窗口函数的执行效率一般都不错，但要注意窗口定义里orderBy的字段最好是分区有序的，否则可能触发额外的排序开销。

踩坑点：窗口函数里的rowsBetween和rangeBetween行为不一样。rowsBetween是按物理行数算，rangeBetween是按值范围算。比如你想算"往前7天的累计"，应该用rangeBetween，而不是rowsBetween。搞混了结果会错得莫名其妙。

---

三、Structured Streaming：流处理的正确打开方式

3.1 为什么不用Spark Streaming（DStream）了

早期的Spark流处理模块叫Spark Streaming，基于DStream（Discretized Stream），本质上就是把流数据切成一小批一小批（micro-batch），每批当作一个RDD来处理。

这个方案有几个硬伤：

• 延迟高：再快也是秒级，做不到毫秒级
• API和Spark SQL不统一：批处理和流处理要写两套代码
• 事件时间处理弱：对数据乱序、迟到的支持不好
• 已经被官方标记为legacy，不再积极开发

结论：新项目直接用Structured Streaming，DStream别学了。

3.2 Structured Streaming的核心理念

Structured Streaming的设计哲学很简洁：把流数据当成一张不断增长的表。每来一批新数据，就相当于往表里插了几行。你的查询逻辑和批处理完全一样，Spark帮你处理"数据源源不断来"这件事。

# 读取Kafka流
stream_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "orders") \
    .option("startingOffsets", "latest") \
    .load()

# 解析JSON数据
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType

schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_id", StringType()),
    StructField("amount", IntegerType()),
    StructField("event_time", TimestampType())
])

orders = stream_df.select(from_json(col("value").cast("string"), schema).alias("data")).select("data.*")

# 按1分钟窗口统计订单金额
from pyspark.sql.functions import window, sum

result = orders \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(window(col("event_time"), "1 minute")) \
    .agg(sum("amount").alias("total_amount"))

# 输出到控制台（调试时用）
query = result.writeStream \
    .outputMode("update") \
    .format("console") \
    .trigger(processingTime="10 seconds") \
    .start()

query.awaitTermination()

注意几个关键概念：

• Watermark（水位线）：解决数据迟到问题的机制。你声明"事件时间超过水位线10分钟的数据我就不要了"，Spark据此决定什么时候可以安全地输出窗口结果并清理状态。

• outputMode：append（只输出新增的行）、update（输出有更新的行）、complete（输出全部结果，只适合聚合结果不大的场景）。
• trigger：多久触发一次计算。不设置的话Spark会尽快处理。

3.3 三种输出模式怎么选

输出模式	含义	适用场景
append	只输出新插入的结果行	无聚合、或带watermark的窗口聚合
update	输出被更新的行（包括新增和修改）	有聚合的流式查询
complete	每次触发都输出完整的结果表	结果集很小的聚合查询

踩坑点：新手最容易搞混的是append和update。如果你的查询里有聚合（groupBy），但输出模式用了append，Spark会报错——因为聚合结果需要更新，不是简单append。反过来，如果不用聚合，update不会报错但可能不如append直观。

3.4 流式Join

Structured Streaming支持流和静态表的Join、流和流的Join。

# 流DataFrame Join 静态DataFrame
stream_df.join(static_df, "join_key", "left")

# 流-流Join（两个Kafka topic按时间窗口Join）
stream1.join(
    stream2,
    (stream1.join_key == stream2.join_key) & 
    (stream1.event_time.between(stream2.event_time - expr("INTERVAL 10 MINUTES"), stream2.event_time)),
    "inner"
)

流-流Join是Structured Streaming的亮点，也是难点。你需要定义好时间约束，否则Spark需要把两边的数据永远留在状态里，内存会爆炸。

踩坑点：流-流Join必须有一边定义watermark，或者两边都定义。否则Spark不知道该什么时候丢弃旧状态。还有一个坑：Join条件里时间范围的写法很容易出错，建议用between明确指定上下界。

3.5 Checkpoint：流任务的"救命稻草"

流式任务会长期运行，机器挂了、代码升级了怎么办？Checkpoint机制帮你保存状态。

query = result.writeStream \
    .outputMode("update") \
    .format("parquet") \
    .option("path", "s3://bucket/output/") \
    .option("checkpointLocation", "s3://bucket/checkpoints/") \
    .start()

Checkpoint目录里保存了：

• 当前处理到Kafka的哪个offset
• 各个算子的中间状态（比如聚合的累加值）
• 查询的执行计划

如果任务挂了，只要checkpoint目录还在，重启任务时会自动从断点续传。

踩坑点：

1. Checkpoint目录一旦设置就不能改，改了等于从零开始。如果一定要改，需要把旧checkpoint里的offset信息想办法迁移。
2. 代码逻辑变了，checkpoint可能不兼容。比如你改了groupBy的字段，旧状态对新逻辑来说是无效的。Spark这时候会报错，解决方案是换个checkpoint路径重新启动。
3. Checkpoint文件会越来越多，需要定期清理。Spark 3.5支持设置checkpointLocation的保留策略，或者用外部定时任务清理过期文件。

---

四、Spark运行架构：你的代码是怎么跑起来的

4.1 核心组件全貌

不管你用YARN、Kubernetes还是Standalone部署，Spark应用的架构都遵循统一的计算模型。理解这套模型是排查问题、调优性能的基础。

完整的架构包含以下角色：

Driver（驱动程序）

Driver是整个Spark应用的"大脑"，运行着你的main()函数。它不负责实际的数据计算，但承担以下关键职责：

• 解析你的代码，构建DAG（有向无环图），描述计算任务的依赖关系
• 通过DAGScheduler将DAG按Shuffle边界切分成多个Stage
• 通过TaskScheduler向集群申请资源，把具体的Task分发到Executor上执行
• 汇总Task的执行结果，返回给用户程序
• 在运行期间持续监控各个Executor的健康状态

Driver是单点——如果Driver进程挂了，整个Spark应用就终止，Executor也会跟着退出。生产环境中务必确保Driver部署在高可用节点上，且配置spark.driver.memory时要给足内存（建议4GB起步），因为DAG的元数据、广播变量、collect()回来的结果都存在Driver端。

Cluster Manager（集群管理器）

Cluster Manager负责统筹集群的物理资源（CPU、内存），本质是一个资源调度器。Spark本身不自带资源管理功能，需要依赖外部系统：

• YARN：Hadoop生态的传统选择，国内大厂主流
• Kubernetes：云原生时代的主流，弹性伸缩能力强
• Standalone：Spark自带的简易调度器，仅适合测试

注意：Cluster Manager只管"给Executor分配多少资源、放在哪台机器上"，不参与Spark的任务调度和计算逻辑。

Executor（执行器）

Executor是真正干活的进程，运行在集群的工作节点上。每个Executor是一个独立的JVM进程，由Cluster Manager根据Spark的申请而启动。Executor内部维护一个线程池，每个线程负责执行一个Task。也就是说：

• 一个Worker节点可以运行多个Executor进程
• 一个Executor进程内部可以同时运行多个Task线程
• 一个Task线程对应RDD的一个分区数据

默认情况下，Executor的每个CPU核心（core）跑一个Task。比如你设置spark.executor.cores=4，那这个Executor同时可以跑4个Task，这4个Task共享该Executor的内存（spark.executor.memory）。

这种"进程内多线程"模型的优势在于：同一个Executor内的多个Task可以共享内存中的广播变量和缓存数据（cache/persist），避免了进程间通信的开销。

下图展示了完整的Spark运行架构与DAG调度流程，涵盖了从RDD转换链、DAG构建、Stage划分、组件交互到Shuffle数据流的全过程：

DAGScheduler（DAG调度器）

DAGScheduler是Driver内部的核心组件。当你调用一个Action时，DAGScheduler会：

1. 从后往前回溯你的Transformation链条，构建完整的DAG
2. 根据RDD的依赖关系（宽依赖vs窄依赖，见4.2节）将DAG切分成Stage
3. 把每个Stage封装成一组Task（Task数量等于该Stage最后一个RDD的分区数），提交给TaskScheduler

TaskScheduler（任务调度器）

TaskScheduler负责把DAGScheduler发来的Task实际分发到各个Executor上。它会：

1. 向Cluster Manager申请Executor资源
2. 根据数据本地性（data locality）原则，尽量把Task分发到数据所在的节点（移动计算而不移动数据）
3. 监控Task的执行状态，失败时重试（默认重试4次）
4. 推测执行（speculative execution）：如果某个Task跑得明显比同Stage的其他Task慢，会在另一个Executor上启动一个副本，谁先完成用谁的结果

直观理解：Driver是总设计师，DAGScheduler是施工图纸分解员，TaskScheduler是包工头，Executor是工人，Cluster Manager是人力资源部。总设计师不搬砖，但没有他，工人们不知道房子要怎么盖。

4.2 宽依赖与窄依赖：Stage划分的依据

理解宽依赖和窄依赖是理解Spark执行模型的关键。

窄依赖（Narrow Dependency）

父RDD的每个分区最多被一个子RDD的分区使用。也就是说，子RDD的分区可以直接在本地从父RDD的少量分区计算出来，不需要跨节点拉取数据。典型的窄依赖操作：

• map()、filter()、flatMap()
• union()、sample()
• 某些join()（当两个RDD已经按相同的key分区过，且分区数相同时）

宽依赖（Wide Dependency，也叫Shuffle Dependency）

父RDD的每个分区被多个子RDD的分区使用。这意味着必须把父RDD的数据按照key重新分发到各个节点，这个过程就是Shuffle。典型的宽依赖操作：

• groupByKey()、reduceByKey()、aggregateByKey()
• sortByKey()、repartition()、coalesce()（增加分区时）
• 大部分join()（两个RDD未按join key分区时）
• distinct()

为什么这个区分如此重要？ 因为窄依赖的操作可以在一个线程内流水线式地完成（比如map之后紧跟着filter），不需要等待其他分区。而宽依赖的操作必须等所有父分区都准备好，数据Shuffle到正确位置后才能开始。DAGScheduler正是依据这个边界来划分Stage——每个Stage内部全是窄依赖，Stage之间由宽依赖（Shuffle）隔开。

一句话记忆：窄依赖是一对一或少数对少数，数据不走网络；宽依赖是一对多，必须Shuffle数据。

4.3 DAG构建与Stage划分流程

当你写了一段Spark代码并触发Action后，Spark内部经历了以下完整的调度流程：

1. 构建DAG：从最后一个RDD（Action触发的那个）开始反向回溯，分析所有Transformation的依赖关系，形成一张有向无环图。

2. 划分Stage（Shuffle为界）：DAGScheduler从后往前遍历DAG，遇到宽依赖就切一刀。最后一个Stage叫ResultStage（负责生成最终结果），前面的Stage都叫ShuffleMapStage（负责为下游Stage准备数据）。

3. 生成Task：每个Stage会被拆分成多个Task。Task的数量等于该Stage最后一个RDD的分区数。比如你的RDD有200个分区，那这个Stage就有200个Task。

4. Task调度：TaskScheduler把Task提交给Executor，优先选择数据所在的节点（数据本地性）。

5. Task执行：Executor的线程池执行Task，窄依赖的Transformation会在一个Task内流水线执行，直到遇到需要Shuffle输出的地方。

6. Shuffle Write/Read：ShuffleMapStage的Task把输出数据按key哈希写到本地磁盘（Shuffle Write）；下游Stage的Task从各个节点拉取属于自己的数据（Shuffle Read，也叫Fetch）。

7. 结果汇总：ResultStage的Task把最终结果返回给Driver。

这个流程在Spark UI的"Jobs"和"Stages"页面可以完整看到，是排查性能问题的核心依据。

4.4 Task、Stage、Job的关系

Spark的执行层次：

• Job：每个Action触发一个Job。比如你先df.count()再df.write.save()，这是两个Job。
• Stage：Job内部按Shuffle操作（比如groupBy、join、repartition）划分为多个Stage。Stage之间需要等上一个Stage全部完成才能开始下一个。
• Task：Stage内的最小执行单元。你的数据有N个分区，这个Stage就有N个Task。

Shuffle是性能杀手。因为Shuffle需要把数据按照key重新分发到各个节点，涉及大量网络IO和磁盘IO。写Spark代码的核心优化思路之一就是减少Shuffle。

一句话总结：Action触发Job，Shuffle切分Stage，分区数决定Task数。

踩坑点：看Spark UI的时候，如果发现某个Stage特别慢，大概率是Shuffle stage。这时候可以看Shuffle Read/Write的大小，如果数据量巨大，考虑：

1. 前面加filter减少数据量
2. 用broadcast join代替shuffle join
3. 检查是否有数据倾斜（某个key的数据特别多）

4.5 部署模式

现在生产环境主流的部署方式有三种：

1. YARN（Hadoop生态的传统选择）

# Client模式：Driver跑在你提交任务的机器上
spark-submit --master yarn --deploy-mode client app.py

# Cluster模式：Driver跑在YARN的某个NodeManager上
spark-submit --master yarn --deploy-mode cluster app.py

• Client模式适合交互式开发（比如pyspark shell），因为Driver在你本地，能看到实时日志。
• Cluster模式适合生产任务，Driver由YARN管理，你提交完任务就可以关掉终端。

踩坑点：Client模式下，如果你的本地机器挂了，整个Spark应用就挂了。所以生产任务一定用Cluster模式。

两种模式的区别看这张图就明白了：

2. Kubernetes（云原生时代的主流选择）

K8s部署越来越流行，特别是已经在用云基础设施的团队。

spark-submit \
  --master k8s://https://k8s-apiserver:443 \
  --deploy-mode cluster \
  --conf spark.kubernetes.namespace=spark \
  --conf spark.kubernetes.container.image=spark:3.5.0 \
  local:///opt/spark/app.py

K8s的优势是弹性好、资源隔离强、和微服务生态统一。劣势是比YARN复杂，需要团队有K8s运维能力。

3. Standalone（测试环境用用就行）

Spark自带的简单集群管理器。配置简单，但功能弱（没有资源隔离、没有队列调度）。生产环境不推荐。

---

五、性能调优：让Spark跑得更快

5.1 数据倾斜：最头疼的问题

数据倾斜的表现：大部分Task几秒就跑完了，就那么几个Task跑了几十分钟甚至几小时，进度条卡在99%不动。

根本原因是某个key的数据量远大于其他key，导致处理这个key的Task特别慢。

诊断方法：看Spark UI的Stage详情，如果某个Task的处理数据量是其他Task的几十倍以上，就是倾斜了。

解决方案：

1. 加盐（Salting）：给倾斜的key加随机前缀，打散到多个Task处理，最后再合并结果。

from pyspark.sql.functions import rand, lit, concat, col

# 给key加0-9的随机前缀
df.withColumn("salted_key", concat(col("skew_key"), lit("_"), (rand() * 10).cast("int"))) \
  .groupBy("salted_key").agg(sum("value").alias("partial_sum")) \
  .withColumn("original_key", col("salted_key").substr(1, expr("length(salted_key) - 2"))) \
  .groupBy("original_key").agg(sum("partial_sum").alias("total"))

2. 两阶段聚合：先局部聚合再全局聚合。上面的加盐方案本质就是两阶段聚合。

3. broadcast join：如果是Join导致的倾斜，且其中一张表不大，用broadcast join避免Shuffle。

4. 调整并行度：有时候倾斜是因为分区数太少，数据太集中。增加spark.sql.shuffle.partitions（默认200，可以调到500-1000）。

踩坑点：加盐方案虽然有效，但代码会变得很丑。Spark 3.x引入了AQE（Adaptive Query Execution），可以自动处理部分倾斜场景。建议先开AQE试试，搞不定再手动加盐。

5.2 AQE（自适应查询执行）

Spark 3.x最重要的性能特性之一，3.5版本已经非常成熟。生产环境强烈建议开启。

spark = SparkSession.builder \
    .appName("demo") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .getOrCreate()

AQE在运行时做三件事：

1. 动态合并分区（Coalesce Partitions）：如果某些分区数据量很小，运行时会自动把它们合并，减少Task数量，降低调度开销。
2. 动态切换Join策略：运行时如果发现某张表比预期的小，自动切换成broadcast join。
3. 动态处理数据倾斜：检测到倾斜后自动做拆分优化。

踩坑点：AQE只在有Shuffle的查询里生效。如果你的查询全是窄依赖（比如filter+select），AQE帮不上忙。另外AQE是基于统计信息做决策的，如果数据分布变化很大，可能需要跑几次才能收敛到最优计划。

5.3 广播变量（Broadcast Variables）

如果你有一个比较大的配置表（比如几千到几百万行的维度表），每个Task都需要查它，用广播变量可以避免每个Task都拷贝一份。

# 把小表广播出去
broadcast_lookup = spark.sparkContext.broadcast(lookup_dict)

# 在UDF里使用
def enrich_data(key):
    return broadcast_lookup.value.get(key, "unknown")

# 注意：广播变量大小有限制，超过会报错。大对象考虑用broadcast join替代

广播变量序列化后的单个块大小由spark.broadcast.blockSize控制（默认4MB），但广播变量的总大小实际受限于序列化后的对象不能超过约2GB（RPC消息上限），以及不能明显挤占Executor内存。如果维度表太大，直接用broadcast join（spark.sql.autoBroadcastJoinThreshold默认10MB）会更可控。

踩坑点：广播变量是不可变的，创建后不能修改。另外不要在广播变量里塞太大的对象，容易OOM。

5.4 Cache和Persist

如果你同一个DataFrame会被多次使用（比如既要统计又要Join），可以把它缓存起来，避免重复计算。

# 默认缓存到内存
df.cache()

# 更细粒度的控制
df.persist(StorageLevel.MEMORY_AND_DISK)

StorageLevel选项：

级别	内存	磁盘	序列化	说明
MEMORY_ONLY	✓	✗	✗	默认，最快，内存不够时重新计算
MEMORY_AND_DISK	✓	✓	✗	内存不够存磁盘，不会重算
MEMORY_ONLY_SER	✓	✗	✓	序列化存内存，省空间但取的时候需要反序列化
OFF_HEAP	✗	✗(堆外内存)	✓	用Tachyon等堆外存储

踩坑点：

1. cache/persist是惰性的，不会立即执行，需要跟一个Action（比如count()）才会真正缓存。
2. 缓存占用内存，如果内存不够，Spark会把分区evict到磁盘（如果StorageLevel含DISK）或者重新计算。监控好Executor的内存使用。
3. 用完记得unpersist()，特别是在notebook交互式环境里，否则内存一直被占着。

5.5 关键配置参数速查

参数	默认值	说明
spark.sql.shuffle.partitions	200	Shuffle时的分区数，数据量大时调大
spark.sql.autoBroadcastJoinThreshold	10MB	自动广播Join的阈值
spark.sql.adaptive.enabled	false	AQE开关，3.5建议开
spark.executor.memory	1g	每个Executor的堆内存
spark.executor.cores	1	每个Executor的CPU核数
spark.sql.files.maxPartitionBytes	128MB	每个分区最大文件大小

踩坑点：spark.sql.shuffle.partitions默认值200在很多场景下太小了。如果你的输入数据是TB级别，Shuffle后200个分区意味着每个分区几GB，任务会跑得很慢。一般建议根据集群核数和Shuffle数据量调整，常见值是400-2000。但也不是越大越好，分区太多会导致调度开销增加、小文件问题。

---

六、综合实战：电商订单分析

6.1 场景描述

假设你在一个电商公司做数据开发，有两个数据源：

1. 订单事实表（Parquet格式，存在S3上）：包含订单ID、用户ID、商品ID、下单时间、金额、状态
2. 用户维度表（MySQL）：包含用户ID、用户等级、注册城市

需求：

• 离线批处理：统计各城市、各用户等级在最近30天的订单总额和订单量
• 实时流处理：从Kafka读取实时订单，按5分钟窗口统计各城市的订单金额，写入另一个Kafka topic供下游消费

整体架构如下：

6.2 数据准备

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("ECommerceAnalytics") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# ========== 订单事实表（模拟生成，实际是你数仓里的表）==========
order_schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_id", StringType()),
    StructField("product_id", StringType()),
    StructField("order_time", TimestampType()),
    StructField("amount", DoubleType()),
    StructField("status", StringType())  # paid, cancelled, refunded
])

# 实际生产环境直接读Parquet
orders_df = spark.read.schema(order_schema).parquet("s3://data-warehouse/orders/dt=2025-01-*")

# ========== 用户维度表（从MySQL读）==========
users_df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://mysql-host:3306/warehouse") \
    .option("dbtable", "dim_users") \
    .option("user", "analytics") \
    .option("password", "xxxxx") \
    .load()

users_df = users_df.select("user_id", "user_level", "city")

6.3 离线批处理：城市+用户等级统计

# 过滤有效订单（已付款），Join维度表，按城市和用户等级聚合
result_df = orders_df \
    .filter(col("status") == "paid") \
    .filter(col("order_time") >= current_date() - expr("INTERVAL 30 DAYS")) \
    .join(broadcast(users_df), "user_id", "left") \
    .fillna({"city": "未知", "user_level": "L0"}) \
    .groupBy("city", "user_level") \
    .agg(
        sum("amount").alias("total_amount"),
        count("*").alias("order_count"),
        avg("amount").alias("avg_amount"),
        countDistinct("user_id").alias("unique_users")
    ) \
    .orderBy(desc("total_amount"))

# 结果写回数据仓库
result_df.write \
    .mode("overwrite") \
    .parquet("s3://data-warehouse/report/city_user_level_stats/")

# 同时写入MySQL供BI工具读取
result_df.write \
    .mode("overwrite") \
    .format("jdbc") \
    .option("url", "jdbc:mysql://mysql-host:3306/report") \
    .option("dbtable", "city_user_level_stats") \
    .option("user", "analytics") \
    .option("password", "xxxxx") \
    .save()

代码说明：

• 用broadcast(users_df)做广播Join，因为用户维度表通常不大（百万级别），避免Shuffle
• fillna处理Join不上维度的脏数据，防止null影响展示
• 结果同时写Parquet（供后续分析）和MySQL（供BI工具），一份计算多种用途

6.4 实时流处理：5分钟窗口城市订单统计

# 从Kafka读实时订单数据
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "realtime_orders") \
    .option("startingOffsets", "latest") \
    .option("failOnDataLoss", "false") \
    .load()

# 解析订单JSON
order_stream_schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_id", StringType()),
    StructField("city", StringType()),
    StructField("amount", DoubleType()),
    StructField("event_time", TimestampType())
])

orders_stream = kafka_df \
    .select(from_json(col("value").cast("string"), order_stream_schema).alias("data")) \
    .select("data.*") \
    .filter(col("amount").isNotNull())  # 过滤脏数据

# 5分钟窗口聚合，水位线10分钟
windowed_stats = orders_stream \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(
        col("city"),
        window(col("event_time"), "5 minutes")
    ) \
    .agg(
        sum("amount").alias("window_amount"),
        count("*").alias("window_order_count")
    ) \
    .select(
        col("city"),
        col("window.start").alias("window_start"),
        col("window.end").alias("window_end"),
        col("window_amount"),
        col("window_order_count")
    )

# 结果写回Kafka供下游消费
# 先把结果转成JSON字符串
output_df = windowed_stats.select(
    to_json(struct("*")).alias("value")
)

query = output_df.writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("topic", "city_order_stats_5min") \
    .option("checkpointLocation", "s3://checkpoints/city_order_stats/") \
    .outputMode("update") \
    .trigger(processingTime="1 minute") \
    .start()

query.awaitTermination()

代码说明：

• failOnDataLoss false：Kafka topic如果被删了重建（offset不连续），不会直接报错。生产环境建议加这个，但要配合监控告警。
• withWatermark("event_time", "10 minutes")：允许数据迟到10分钟。比如一个event_time=10:00的数据在10:08才到，仍然会被统计到10:00-10:05的窗口里。超过10:10之后，10:00的窗口才会被关闭输出。
• trigger(processingTime="1 minute")：每1分钟触发一次输出。如果这1分钟内没有新数据，也会空触发一次（输出空的update）。

---

七、生产环境踩坑合集（血泪总结）

7.1 小文件问题

如果你的输入数据有很多小文件（比如每个几KB），Spark会为每个文件创建一个分区，导致Task数量爆炸，调度开销远大于计算开销。

解决：

# 读取时合并小文件
spark.read.option("mergeSchema", "true").parquet("path/")

# 或者写入时控制文件大小
df.coalesce(10).write.parquet("output/")  # 合并成10个文件
df.repartition(10).write.parquet("output/")  # 重新分区后写

coalesce vs repartition：coalesce只是减少分区数，不会触发Shuffle，效率高但可能数据不均匀；repartition会Shuffle，数据更均匀但开销大。

踩坑点：Hive表长期积累会产生大量小文件。可以定期做INSERT OVERWRITE合并，或者用Spark的OPTIMIZE命令（如果用Delta Lake）。

7.2 内存溢出（OOM）

Executor OOM是最常见的报错。原因通常是：

1. 数据量太大，单分区数据量超过Executor内存
2. 用了collect()把大数据拉回Driver
3. 缓存了太多数据没unpersist
4. 复杂的UDF占用了大量内存

排查方法：看Spark UI的Executors页面，找Memory栏爆掉的节点。看日志里的java.lang.OutOfMemoryError。

解决：

• 增加Executor内存：--conf spark.executor.memory=8g
• 减少并发度：spark.executor.cores=2（每个Executor少跑几个Task）
• 避免collect()大数据，改用write.save()
• 复杂的聚合操作前先filter、减少列

7.3 Kafka Offset管理

Structured Streaming的Kafka offset默认存在checkpoint里。但如果你需要用外部系统管理offset（比如Zookeeper、Redis），可以用startingOffsets和自定义的offset管理逻辑。

踩坑点：Kafka consumer group ID由Spark自动生成，如果你用外部工具（比如kafka-consumer-groups.sh）查消费进度，可能找不到对应的group。可以通过groupIdPrefix配置统一前缀。

7.4 时区问题

Spark处理时间戳时，默认用JVM的时区。如果集群机器时区不统一，或者你的数据是UTC但Spark按本地时区解析，结果会差8小时（中国时区）。

解决：

spark.conf.set("spark.sql.session.timeZone", "UTC")  # 统一用UTC
# 或者显示转换
df.withColumn("local_time", from_utc_timestamp(col("event_time"), "Asia/Shanghai"))

7.5 Schema Evolution（Schema演变）

数仓表运行久了，上游可能会加字段、改类型。Parquet和ORC都支持schema evolution（增字段没问题），但改类型可能导致读取失败。

踩坑点：建议用Delta Lake（Databricks开源）或Apache Iceberg替代裸Parquet。它们提供了完整的ACID事务、schema evolution、time travel查询能力。Spark 3.5对这两个格式的支持已经非常成熟。

7.6 Notebook里的坑

如果你用Jupyter、Zeppelin或Databricks Notebook写Spark代码：

1. 变量覆盖了之前的DataFrame，但旧DataFrame还在缓存里：Notebook是按单元格执行的，你执行了df = df.filter(...)，旧的df引用没了，但如果之前cache过，内存不会自动释放。需要手动df.unpersist()。

2. SparkSession不会自动重启：你改了配置重新创建SparkSession，但旧的还在运行。需要先spark.stop()，或者整个重启kernel。

3. 隐式的Action导致重复计算：Notebook里每个Cell末尾的变量会自动show()，这会触发Action。如果一个DataFrame依赖链很长，你在中间Cell看了眼数据，等于多跑了一遍。

---

八、写在最后

Spark 3.5.x和2.x相比，核心变化其实就几个方向：

1. Spark SQL（DataFrame/Dataset）是唯一推荐的API，RDD进博物馆了（除非特定场景）
2. Structured Streaming替代DStream，批流一体是主流
3. AQE让性能优化更自动化，手动调优的负担减轻很多
4. Spark Connect提供了新的客户端-服务端架构，远程连接更方便
5. PySpark的体验大幅提升，Arrow优化、UDTF、更好的错误提示

如果这篇文章对你有帮助，欢迎点赞收藏。有问题评论区见～

---