本章目标：能讲清 Spark 的 Driver/Executor、Job/Stage/Task、RDD/DataFrame、惰性求值和 shuffle，并在本地跑通第一个 PySpark 程序。 前置知识：读完 Hadoop 基础；会一点 Python（变量、函数、列表）即可。 预计用时：2.5 小时。
1. 为什么需要它
上一章说过 MapReduce 慢在四处：中间结果反复落盘、复杂任务要串多个作业、固定只有 Map/Reduce 两阶段、API 表达力弱。Spark 针对这四点做了改进：
| MapReduce | Spark | |
|---|---|---|
| 中间结果 | 主要落磁盘 | 优先放内存，放不下再溢写磁盘 |
| 执行模型 | 固定 Map/Reduce 阶段 | DAG，多个转换可以连成一条流水线 |
| 编程接口 | Java Map/Reduce 类 | RDD、DataFrame、SQL、Streaming |
| 迭代计算 | 每轮重新读盘写盘 | 可以 cache 中间结果反复用 |
| 延迟 | 分钟级起步 | 批处理秒到分钟，交互查询更快 |
关键认知：Spark 快，不是“因为内存所以不读磁盘”。shuffle 仍然要落盘、走网络，数据一大照样慢。Spark 的真正优势是：
- 惰性求值，能看到完整计算链后再优化。
- DAG 把多个操作合并，减少中间落盘和作业启动。
- DataFrame/SQL 共用 Catalyst 优化器，能自动做列裁剪、谓词下推、join 重排。
- 同一套 API 覆盖批、SQL、流、机器学习。
Spark 也不是万能的：几百行数据用 pandas 更快；毫秒级点查用 MySQL/Redis；严格事务用数据库；常驻毫秒级流处理通常选 Flink。Spark 的主战场是大规模离线批处理和数仓 ETL。
2. 核心概念
2.1 架构：Driver、Executor、Cluster Manager
Driver（main 进程，SparkSession/DAG Scheduler）
| 申请资源、下发任务
v
Cluster Manager（local / YARN / K8s / Standalone）
| 分配 Executor
v
Executor（JVM 进程，多个 core 跑多个 Task，负责 cache 和 shuffle 读写）
Driver 是“大脑”：运行 main()，创建 SparkSession，把代码翻译成执行计划，按 shuffle 切 Stage，调度 Task 并收集结果。Driver 挂了，应用就结束。
Executor 是“工人”：每个 Executor 是一个 JVM 进程，包含若干 core，一个 core 同时跑一个 Task；它执行 Task、缓存数据、读写 shuffle 文件。应用结束后被回收。
Cluster Manager 只负责资源分配：local[*] 用本机线程模拟，YARN 用 Container，Kubernetes 用 Pod，Standalone 是 Spark 自带模式。
注意：
local[*]不是集群。Driver 和 Executor 在同一个 JVM 里，只是用多个线程模拟并行。它的作用是学习语法和验证逻辑，不能用来推断真实集群的性能。
2.2 Job、Stage、Task
这三个词是 Spark 执行模型的核心，关系是“包含”：
Action -> Job ->（按 shuffle 边界切）Stage ->（按分区切）Task
例子：df.filter(...).groupBy("order_status").count().show()。
show() 是 Action，触发一个 Job。groupBy 需要把相同状态的数据拉到一起，产生 shuffle，所以 Job 至少被切成两个 Stage：shuffle 前做过滤和局部聚合，shuffle 后做最终聚合。每个 Stage 有多少分区，就发多少个 Task。
面试高频：Task 数不等于数据行数，而等于分区数。分区太少，机器闲着；分区太多，调度开销大。
2.3 RDD、DataFrame、Dataset
| RDD | DataFrame | Dataset | |
|---|---|---|---|
| 是什么 | 不可变的分布式对象集合 | 带 schema 的分布式表 | 带类型的分布式数据集 |
| schema | 无 | 有 | 有 |
| 优化器 | 无，靠开发者 | Catalyst + Tungsten | Catalyst + Tungsten |
| API 风格 | map、filter、reduceByKey |
select、filter、groupBy |
DataFrame + 强类型对象 |
| PySpark | 支持 | 支持 | 不直接支持 |
RDD 是最底层抽象：数据被切成分区，分布在不同机器上；不可变，每次转换生成新 RDD；靠 lineage（血缘）记录“从哪来、经过哪些转换”，丢了可以重算。RDD 适合非结构化数据、需要精细控制分区和算子的场景。
DataFrame 是“分布式的表”，有列名和类型，接口像 pandas。你写 filter、groupBy，Spark 先记成逻辑计划，再由 Catalyst 优化。生产里 90% 的场景用 DataFrame 或 Spark SQL。
Dataset 是 DataFrame 的强类型版本，主要在 Scala/Java 中使用；PySpark 没有真正的 Dataset，DataFrame 就等价于 Dataset[Row]。面试被问到，照实说“PySpark 里主要用 DataFrame”。
2.4 惰性求值：转换与行动
Spark 把操作分成两类：
- 转换（Transformation）：
select、filter、withColumn、join、groupBy、orderBy、distinct、repartition。它们只记录“要做什么”，不立刻计算。 - 行动（Action）：
show、count、collect、first、take、write。它们才触发真正执行。
df2 = df.filter(F.col("price") > 100).select("order_id", "price") # 什么都没发生
df2.show() # 到这里才提交 Job、读数据、计算
这种设计叫惰性求值。好处是 Spark 能看到完整链条再优化：比如你先 select 三列再 filter，优化器可能把过滤下推到读取阶段，只读需要的列；如果走一步算一步，就没有这种全局视角。
代价是调试时容易误判：df.count() 写在循环里，每次循环都重算一遍；df.cache() 之后不调用 Action，缓存也不会真正发生。用 df2.explain(True) 可以看逻辑计划、优化后计划和物理计划。
2.5 宽窄依赖、DAG 与 shuffle
窄依赖：父分区的数据只流向一个子分区，不需要跨机器搬数据。map、filter、union、coalesce、广播 join 都属于窄依赖，可以在一个 Stage 内流水线执行。
宽依赖：一个子分区依赖多个父分区，必须把数据按 key 重新分发，产生 shuffle。groupBy、distinct、repartition、orderBy、非广播 join 都会触发宽依赖。
read -> filter -> select -> [shuffle] -> groupBy -> agg -> show
|<------ Stage 1 ------>| |<---- Stage 2 ---->|
窄依赖流水线 宽依赖边界
shuffle 的过程：Map 端按 key 的 hash 把数据分成若干份写到本地磁盘；Reduce 端从所有 Map 端拉取属于自己的那一份，再归并、排序、聚合。它贵在磁盘 I/O、网络传输、序列化和排序。Spark SQL 默认 spark.sql.shuffle.partitions=200，这个值对 GB 级和 TB 级数据不可能都合适，后面调优章会讲怎么改。
DAG 就是这些转换组成的有向无环图。Stage 的边界就是宽依赖；同一个 Stage 内的窄依赖可以合并成流水线，减少中间数据。某个分区丢失时，Spark 沿 lineage 重算该分区，不需要重跑整个作业。
2.6 Spark SQL 与 Catalyst
Spark SQL 不只是“写 SQL 的接口”，它是一套查询优化和执行引擎。DataFrame API 和 SQL 最终都会变成同一棵逻辑计划树：
SQL/DataFrame -> 未解析逻辑计划 -> Analyzer（目录、列名、类型）
-> 优化后逻辑计划 -> Optimizer（谓词下推、列裁剪、join 重排）
-> 物理计划（按代价选）-> Whole-stage codegen -> 执行
常见优化：
- 谓词下推：能早点过滤就早点过滤，少读数据。
- 列裁剪：只读用到的列，Parquet/ORC 列式存储收益明显。
- 常量折叠：
1 + 2在编译期算成3。 - join 重排/广播：小表广播，避免大表 shuffle。
- 代码生成：把多个算子合成一段 Java 字节码，减少虚函数调用。
所以“DataFrame 比手写 RDD 快”通常不是因为 API 本身，而是因为 Catalyst 能自动做这些优化。你用 spark.sql() 和 DataFrame API 写同一个查询，执行计划往往一样。
2.7 容错：丢了就重算
Spark 的容错主要靠 lineage：每个 RDD/DataFrame 都记得自己从哪来、做过哪些转换。某个分区所在的机器挂了，Driver 重新调度这个分区，沿血缘从可用的父分区重算。
血缘太长时重算很贵，解决办法是 checkpoint：把中间结果写到可靠存储（HDFS/S3），切断血缘。checkpoint 会真正物化并丢弃部分 lineage，代价比 cache 高；cache 只是为了复用，血缘还在。此外还有 Task 重试和推测执行：慢 Task 可能被复制到另一台机器，谁先完成用谁的结果，但这不能替代真正的倾斜处理。
3. 动手示例
3.1 安装 Java 17 和 PySpark
Spark 跑在 JVM 上，先确认 Java：
java -version
课程统一用 Java 17 + Spark 3.5.5：Java 17 是当前 LTS，Spark 3.5 对它的兼容性已经稳定。没装就去 Adoptium 下载 JDK 17，安装时勾选设置 JAVA_HOME 和 PATH；没勾选就手动配好，再关掉终端重开。
安装 PySpark 时锁定版本，避免不同机器装到不同大版本：
pip install pyspark==3.5.5
python -c "import pyspark; print(pyspark.__version__)"
注意：PySpark 3.5.5 需要 Python 3.8+。如果 Python 3.13 装完启动报 Python worker 错误，换 Python 3.12。版本不匹配时不要硬调，先换版本。
3.2 第一个程序：spark_hello.py
在项目根目录新建 scripts/spark_hello.py：
# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
spark = (SparkSession.builder
.appName("hello")
.master("local[*]")
.getOrCreate())
spark.sparkContext.setLogLevel("ERROR")
# RDD 版 WordCount
rdd = spark.sparkContext.parallelize([
"hello hadoop", "hello spark", "hadoop spark",
])
counts = (rdd.flatMap(lambda line: line.split())
.map(lambda word: (word, 1))
.reduceByKey(lambda a, b: a + b))
print("WordCount:", sorted(counts.collect()))
# DataFrame 版读 CSV
df = (spark.read.option("header", True)
.option("inferSchema", True)
.csv("data/sample/orders_demo.csv"))
df.printSchema()
df.show(5)
print("订单行数:", df.count())
spark.stop()
运行：
python scripts/spark_hello.py
WordCount 输出固定为：
WordCount: [('hadoop', 2), ('hello', 2), ('spark', 2)]
订单行数应为 12。第一次启动要起 JVM，慢 30 秒到 1 分钟是正常的。
3.3 用 explain 看惰性求值
把下面几行加到程序里：
from pyspark.sql import functions as F
filtered = (df.filter(F.col("status") != "canceled")
.select("order_id", "status", "order_date"))
filtered.explain(True) # 只打印计划，不读全表
filtered.show(5) # 这里才真正计算
explain 会打印逻辑计划和物理计划，不需要背每个节点，先找三样东西：
Filter是否下推到FileScan附近（谓词下推）。FileScan的 schema 是否只剩用到的列（列裁剪）。- 是否有
Exchange（shuffle 标志）；看到BroadcastExchange说明小表被广播。
3.4 local 和真实集群的差别
local[*] 只用于学习：分区在本机线程里跑，没有网络传输、副本和资源队列。真实提交通常长这样（概念示例，本地不执行）：
spark-submit \
--master yarn \
--deploy-mode cluster \
--executor-memory 8g \
--executor-cores 4 \
scripts/job.py
本地调通的代码不代表集群上能直接跑：路径要从本地文件换成 HDFS/对象存储，资源参数要按集群配，日志要去 YARN 上看。这些差异在调优章展开。
4. 常见坑
| 坑 | 表现 | 原因与处理 |
|---|---|---|
在循环里调 count()/show() |
越跑越慢，日志刷屏 | 每次 Action 都触发一个 Job；先算一次再复用，或写临时表 |
collect() 拉全量 |
Driver OOM | 结果大时用 write 落盘，或先聚合再 show |
以为 cache() 立刻生效 |
内存没变化 | cache 是惰性的，要等第一个 Action；用完 unpersist() |
以为 local[*] 是集群 |
性能判断失真 | 它只是本机多线程，不能模拟网络和资源竞争 |
| 分区数一直用默认 200 | 小数据任务过多，大数据单任务过重 | 按数据量调 spark.sql.shuffle.partitions，或开 AQE |
| Python UDF 写复杂逻辑 | 比内置函数慢很多 | 优先用 pyspark.sql.functions；必要时用 pandas UDF/Arrow |
| 版本不匹配 | Python worker failed、JVM 启动失败 | Java 17 + 匹配的 PySpark/Python 版本，先换版本再调代码 |
Windows 写文件报 winutils.exe |
读写本地 Parquet 失败 | 只读练习可继续；要写文件用 Hive 兼容目录、WSL2 或 Docker |
5. 练习
练习 1：区分转换和行动
下面哪些是 Transformation，哪些是 Action？
df.select("order_id")
df.filter(F.col("price") > 100)
df.count()
df.groupBy("order_status").count()
df.write.parquet("data/out/orders")
df.orderBy("order_date")
df.first()
参考答案
Transformation：select、filter、groupBy(...).count()、orderBy；groupBy("x").count() 返回新 DataFrame，仍是转换。
Action：df.count()、df.write.parquet(...)、df.first()。区分方法：返回值是不是新的 DataFrame/RDD；把结果交给 Driver 或写出存储的，就是 Action。
练习 2：数 Job、Stage、Task
一段代码只有一个 Action，源数据有 4 个分区，中间做了一次 groupBy，shuffle 后仍是 4 个分区。至少有几个 Stage？大致多少个 Task？
参考答案
一个 Action 触发一个 Job。groupBy 产生一次 shuffle，所以至少 2 个 Stage：shuffle 前一个，shuffle 后一个。
Task 数按 Stage 内的分区数算：前后都是 4 个分区时，大约 4 + 4 = 8 个 Task。实际还受 map 端预聚合、AQE 合并分区、文件分片影响，不能只背固定数字。关键：Stage 看 shuffle 边界，Task 看分区数。
练习 3：用 RDD 写 WordCount
用 spark.sparkContext.parallelize 造三行文本，统计每个单词出现次数，按单词排序输出。
参考答案
rdd = spark.sparkContext.parallelize([
"hello hadoop", "hello spark", "hadoop spark",
])
result = (rdd.flatMap(lambda line: line.split())
.map(lambda word: (word, 1))
.reduceByKey(lambda a, b: a + b)
.sortBy(lambda x: x[0])
.collect())
print(result)
输出：[('hadoop', 2), ('hello', 2), ('spark', 2)]。
flatMap 把一行拆成多个单词；map 打标；reduceByKey 先在 Map 端局部聚合再 shuffle，比 groupByKey 更省网络。sortBy 是宽依赖，会再触发一次 shuffle。
练习 4：解释为什么没有立刻读数据
你写了：
df2 = df.filter(F.col("order_status") == "delivered").select("order_id")
运行后什么输出都没有，也没有报错。为什么？怎样让它真正执行？
参考答案
filter 和 select 都是 Transformation，只构建执行计划，不触发计算。Spark 采用惰性求值，等 Action 才提交 Job。
加一个 Action 即可，例如 df2.show(5)、df2.count() 或 df2.write.parquet(...)。调试时可以用 df2.explain(True) 看计划，但 explain 本身不会读全表。
练习 5：什么时候该广播小表
大表 order_items 要和 32,951 行的 products（约 2.4 MB）按 product_id join。应该用普通 join 还是广播 join？怎么确认广播生效？
参考答案
products 约 3.3 万行、2.4 MB，远小于默认广播阈值（10 MB），适合广播：
from pyspark.sql.functions import broadcast
result = items.join(broadcast(products), "product_id", "left")
result.explain(True)
在物理计划里找 BroadcastHashJoin 和 BroadcastExchange。如果仍是 SortMergeJoin，可能是统计信息不准、表超过阈值或 hint 被限制。广播表会复制到每个 Executor，不要对几 GB 的表强行广播。
练习 6：给一条 ETL 标出 shuffle 点
需求：读订单和明细，过滤取消单，按订单汇总金额，再按品类算 GMV，最后写 Parquet。列出主要步骤，并标出哪些步骤会 shuffle。
参考答案
- 读
orders、order_items、products，过滤取消单和脏明细：读文件 + 窄依赖，不 shuffle。 - 明细 join 订单、join 商品：商品表小且广播时不 shuffle，否则 join 会 shuffle。
- 按
order_id汇总金额，再按category汇总 GMV：两次 groupBy，都会 shuffle。 - 写出 Parquet：按分区写、写前重分区会 shuffle。
优化方向：小表广播、先过滤再 join、能合并的聚合尽量合并、写前控制分区数。广播 join 和预聚合能减少搬数据，但聚合本身通常仍需要 shuffle。
6. 面试延伸
面试题：Driver 和 Executor 的区别是什么？
是什么：Driver 运行 main()，负责创建 SparkSession、生成 DAG、切 Stage、调度 Task、收集结果；Executor 是工作进程，执行 Task、缓存数据、读写 shuffle 文件。
解决什么问题：把“指挥”和“干活”分开。Driver 只做计划和协调，不碰全量数据；Executor 可以横向扩展，数据分片并行处理。
我的实操边界：我在本地模式写过 PySpark，Driver 和 Executor 在同一个 JVM 里；理解集群模式下由 Cluster Manager 分配 Executor，但没有在生产 YARN 上排查过 Driver OOM 和 Executor 丢失。
面试题：惰性求值有什么好处？
是什么：Transformation 只记录计划，Action 才触发执行。
解决什么问题：Spark 能看到完整计算链，统一做谓词下推、列裁剪、join 重排和流水线合并，避免产生不必要的中间结果；也方便把多个窄依赖合并进一个 Stage。
我的实操边界：我用 explain(True) 看过过滤和列裁剪是否下推，也踩过在循环里反复 count() 的坑；没有做过 Catalyst 源码级开发。
面试题：shuffle 为什么慢？
是什么：shuffle 是把数据按 key 重新分发到不同分区的过程，发生在 groupBy、join、distinct、repartition、orderBy 等宽依赖处。
解决什么问题：它让相同 key 的数据聚到一起，才能做聚合和 join；代价是 Map 端写磁盘、Reduce 端走网络拉取，还要序列化、反序列化和排序。
我的实操边界：我在本地小数据上观察过 Exchange 节点和 Stage 切分，理解 shuffle 的读写过程；没有处理过 TB 级 shuffle 的磁盘和网络瓶颈，调优参数只了解常用项，见 Spark 调优。
7. 本章验收
- 能画出 Driver、Executor、Cluster Manager 的关系，并说出各自职责
- 能解释 Job、Stage、Task 的包含关系，以及 Stage 按什么切分
- 能说出 RDD、DataFrame、Dataset 的区别，知道 PySpark 主要用哪个
- 能各举 5 个 Transformation 和 Action，并解释惰性求值
- 能说出窄依赖和宽依赖的区别，指出哪些操作会触发 shuffle
- 能运行
spark_hello.py，看到 WordCount 结果和 12 行示例订单 - 能用
explain(True)找到 Filter、FileScan、Exchange 三类关键节点 - 完成 6 道练习，至少 4 道能独立写出答案
下一章
PySpark 实战：用 DataFrame 和 Spark SQL 把电商数据真正算成指标，并与 DuckDB 对账。