Spark 核心知识点复习大全（基于PySpark）

模块一：Spark 基础与架构

1. Spark 的架构与核心组件

Driver：负责执行main方法，创建SparkContext，负责作业调度、任务分配和监控。
Executor：运行在Worker节点上，负责执行具体的Task任务，并返回结果给Driver。
Master/Worker：资源管理层面的概念，Master负责资源分配，Worker负责启动Executor。
Application/Master/Worker/Executor/Driver/Task/Stage/Job 的关系需理清。

2. 四种部署模式

Local（本地模式，测试用）
Standalone（Spark自带资源调度框架）
YARN（Hadoop资源调度框架，生产环境最常用）
Mesos（逐渐淘汰）
Spark on Yarn模式的优点：统一资源调度，按需分配，支持多种计算框架混跑。

3. Spark 任务执行流程（重点）

Driver 创建 SparkContext，向资源管理器（如YARN）申请资源。
ApplicationMaster 启动，申请Executor资源。
Executor 启动后向 Driver 注册。
Driver 将代码转化为 DAG，划分Stage（根据宽窄依赖），生成TaskSet。
将 Task 分发到 Executor 执行。

cover: 附件/Halo封面/database.jpg

模块二：RDD 编程核心（转换算子与行动算子）

1. Value 类型转换算子

`map` vs `mapPartitions` (重点思考题)

区别：map 每次处理一条数据；mapPartitions 每次处理一个分区的数据（传入迭代器）。mapPartitions 效率更高（函数调用开销少，可共享变量），但容易导致 OOM。
思考题完整代码演示：

from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession

# 初始化环境
spark = SparkSession.builder.appName("Review").master("local[*]").getOrCreate()
sc = spark.sparkContext

rdd = sc.parallelize([1, 2, 3, 4], 2)

# map: 针对每个元素
map_rdd = rdd.map(lambda x: x * 2)
print("map:", map_rdd.collect()) # [2, 4, 6, 8]

# mapPartitions: 针对每个分区（迭代器）
def process_partition(iterator):
    yield sum(iterator)
map_part_rdd = rdd.mapPartitions(process_partition)
print("mapPartitions:", map_part_rdd.collect()) # 结果是分区的和 [3, 7]

# mapPartitionsWithIndex: 带分区索引
def process_index(index, iterator):
    yield (index, list(iterator))
print("mapPartitionsWithIndex:", rdd.mapPartitionsWithIndex(process_index).collect())

# flatMap: 压平
flat_rdd = sc.parallelize(["hello world", "hello spark"])
print("flatMap:", flat_rdd.flatMap(lambda x: x.split(" ")).collect())

# glom: 将每个分区形成数组
print("glom:", rdd.glom().collect())

# groupBy
print("groupBy:", rdd.groupBy(lambda x: x % 2).mapValues(list).collect())

# filter
print("filter:", rdd.filter(lambda x: x > 2).collect())

# sample (withReplacement, fraction, seed)
print("sample:", rdd.sample(False, 0.5, 42).collect())

# distinct
print("distinct:", sc.parallelize([1, 1, 2, 2]).distinct().collect())

# coalesce vs repartition (coalesce默认不shuffle，repartition必定shuffle)
print("coalesce:", rdd.coalesce(1).getNumPartitions())
print("repartition:", rdd.repartition(4).getNumPartitions())

# sortBy
print("sortBy:", sc.parallelize([3, 1, 2]).sortBy(lambda x: x, ascending=False).collect())

双 Value 类型转换算子

rdd1 = sc.parallelize([1, 2, 3])
rdd2 = sc.parallelize([3, 4, 5])

# intersection 交集
print("intersection:", rdd1.intersection(rdd2).collect()) # [3]
# union 并集
print("union:", rdd1.union(rdd2).collect()) # [1,2,3,3,4,5]
# subtract 差集
print("subtract:", rdd1.subtract(rdd2).collect()) # [1,2]
# zip 拉链（要求分区数及元素个数一致）
rdd3 = sc.parallelize(["a", "b", "c"])
print("zip:", rdd1.zip(rdd3).collect()) # [(1,'a'), (2,'b'), (3,'c')]

2. Key-Value 类型转换算子

`reduceByKey` vs `groupByKey` (重点思考题)

区别：reduceByKey 会在map端进行局部聚合（Combine），大大减少Shuffle落盘数据量；groupByKey 不会进行map端聚合，会将所有数据全量拉取，极易导致OOM。开发中优先使用 reduceByKey。

`reduceByKey`, `foldByKey`, `aggregateByKey`, `combineByKey` 的区别

reduceByKey：分区内和分区间计算规则相同，第一个数据不参与计算（作为初始值）。
foldByKey：相比reduceByKey，提供一个初始值（zeroValue），分区内和分区间计算规则相同。
aggregateByKey：提供初始值，分区内和分区间计算规则可以不同。
combineByKey：最底层的接口，允许返回值的类型与输入类型不一致（自定义初始值结构）。

kv_rdd = sc.parallelize([("a", 1), ("b", 2), ("a", 3), ("b", 4)], 2)

# reduceByKey
print("reduceByKey:", kv_rdd.reduceByKey(lambda x, y: x + y).collect())

# groupByKey
print("groupByKey:", kv_rdd.groupByKey().mapValues(list).collect())

# aggregateByKey (zeroValue, seqOp, combOp)
# 分区内取最大值，分区间相加
print("aggregateByKey:", kv_rdd.aggregateByKey(0, lambda x, y: max(x, y), lambda x, y: x + y).collect())

# foldByKey (zeroValue, func)
print("foldByKey:", kv_rdd.foldByKey(0, lambda x, y: x + y).collect())

# combineByKey (createCombiner, mergeValue, mergeCombiners)
# 计算平均值 (sum, count)
def createCombiner(v): return (v, 1)
def mergeValue(acc, v): return (acc[0] + v, acc[1] + 1)
def mergeCombiners(acc1, acc2): return (acc1[0] + acc2[0], acc1[1] + acc2[1])
combined = kv_rdd.combineByKey(createCombiner, mergeValue, mergeCombiners)
print("combineByKey (sum, count):", combined.collect())

# sortByKey
print("sortByKey:", kv_rdd.sortByKey(ascending=True).collect())

# join 和 leftOuterJoin
rdd_a = sc.parallelize([("a", 1), ("b", 2)])
rdd_b = sc.parallelize([("a", "apple"), ("c", "cat")])
print("join:", rdd_a.join(rdd_b).collect()) # [('a', (1, 'apple'))]
print("leftOuterJoin:", rdd_a.leftOuterJoin(rdd_b).collect()) # [('b', (2, None)), ('a', (1, 'apple'))]
print("cogroup:", rdd_a.cogroup(rdd_b).mapValues(lambda x: (list(x[0]), list(x[1]))).collect())

3. Action 行动算子

action_rdd = sc.parallelize([1, 2, 3, 4, 5])

print("reduce:", action_rdd.reduce(lambda x, y: x + y)) # 15
print("collect:", action_rdd.collect()) # [1, 2, 3, 4, 5]
print("count:", action_rdd.count()) # 5
print("first:", action_rdd.first()) # 1
print("take:", action_rdd.take(2)) # [1, 2]
print("takeOrdered:", action_rdd.takeOrdered(2)) # [1, 2] (升序取前2)
print("aggregate:", action_rdd.aggregate(0, lambda acc, v: acc + v, lambda a, b: a + b)) # 15
print("fold:", action_rdd.fold(0, lambda x, y: x + y)) # 15

kv_action = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
print("countByKey:", kv_action.countByKey().collect()) # [('a', 2), ('b', 1)]

# foreach (无返回值，常用于写入外部系统)
action_rdd.foreach(lambda x: print("Foreach print:", x))

cover: 附件/Halo封面/database.jpg

模块三：Shuffle 机制与 Join 策略

1. Spark Shuffle（重点）

导致Shuffle的算子：repartition, groupByKey, reduceByKey, join, distinct, sortByKey 等。
Hash Shuffle：早期机制，产生大量小文件（M * R个），磁盘IO和内存开销大。
Sort Shuffle：优化后机制，每个Map Task将所有结果写入一个数据文件和一个索引文件，减少文件数量。包含普通Sort Shuffle和Bypass Sort Shuffle（当分区数小于200时触发，不进行排序，直接写文件）。
MR与Spark的Shuffle过程对比：MR的Shuffle是必须落盘的（除非用MapReduce的优化），Spark的Shuffle基于内存和磁盘，且Spark的Shuffle机制有Hash和Sort两种。

2. Spark Join 的三种策略

Broadcast Hash Join (Map端Join)：大表Join小表。将小表广播到所有Executor，在Map端完成Join，避免Shuffle。（最推荐，效率最高）
Shuffle Hash Join：大表Join大表（且一张表比另一张小很多，但不够小到广播）。按照Join Key进行Hash重分区，然后在各分区内进行Hash Join。
Sort-Merge Join：大表Join大表。先按照Join Key进行Shuffle重分区并排序，然后进行归并排序Join。是Spark默认的Join策略。

cover: 附件/Halo封面/database.jpg

模块四：Spark 核心概念与容错机制

1. RDD、DataFrame、DataSet 的区别

RDD：底层数据抽象，强类型，面向对象，支持编译时类型检查，但序列化开销大（JVM对象）。
DataFrame：以列式存储（Schema）组织的分布式数据集，弱类型（Row），自带优化器（Catalyst），执行效率高，但缺乏编译时类型安全。
DataSet：结合了RDD的强类型和DataFrame的优化器（Spark 1.6引入，Python目前不支持DataSet，只有Scala/Java支持）。

2. RDD 的理解与创建方式

弹性分布式数据集。不可变、可分区、可并行计算。
创建方式：
1. 从集合创建：sc.parallelize()
2. 从外部存储创建：sc.textFile()
3. 从其他RDD转换：rdd.map()

3. 宽窄依赖

窄依赖 (Narrow Dependency)：父RDD的一个分区只对应子RDD的一个分区（如 map, filter, union）。支持流水线优化，分区恢复效率高。
宽依赖 (Wide Dependency)：父RDD的一个分区对应子RDD的多个分区（如 groupByKey, reduceByKey），会产生Shuffle。是划分Stage的依据。

4. RDD 容错机制

Lineage (血统)：利用宽窄依赖关系重新计算丢失的分区。
Cache 缓存：将RDD数据缓存到内存/磁盘（persist / cache），避免重复计算。注意：Cache是懒执行，必须在Action算子触发后才会真正缓存。
CheckPoint 检查点：将RDD数据持久化到可靠的分布式文件系统（如HDFS）。不仅切断血统，还提供容错。sc.setCheckpointDir("hdfs://…")。

5. 其他核心概念

数据本地性：PROCESS_LOCAL > NODE_LOCAL > RACK_LOCAL > ANY。计算向数据移动。
block 与 partition 的关系：HDFS的block是物理存储单位（默认128MB），Spark的partition是逻辑计算单位。读取HDFS文件时，通常一个block对应一个partition（除非使用压缩文件等特殊格式）。
Unified Memory Management：统一内存管理模型，执行内存（Execution，用于Shuffle, Join等）和存储内存（Storage，用于Cache）可以互相借用，提高内存利用率。
如何处理不能被序列化的对象：实现 Serializable 接口；使用 transient 关键字修饰不需要序列化的字段；将对象定义为静态（在Scala中）；或者将其封装在闭包外并在Executor端初始化。

cover: 附件/Halo封面/database.jpg

模块五：Spark SQL 与 DataFrame

1. Spark SQL 是什么？与传统 SQL 的区别

Spark SQL是基于Spark引擎的分布式SQL查询引擎。传统SQL通常运行在单一节点（如MySQL）或依赖专门的MPP架构（如ClickHouse）。Spark SQL可以处理PB级数据，支持结构化、半结构化和非结构化数据，并且可以无缝与Spark生态（MLlib, GraphX）结合。

2. 核心组件与 Hive 集成

核心组件：Catalyst Optimizer（优化器）、Tungsten（钨丝计划，内存和CPU优化）。
与Hive集成：通过 enableHiveSupport() 开启，读取Hive Metastore，直接使用Hive的元数据和SQL语法。

3. 处理 JSON 数据与交互式分析

# 完整可跑：Spark SQL 处理 JSON 及交互式分析
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SparkSQLReview").master("local[*]").getOrCreate()

# 准备 JSON 数据
json_data = """
{"name": "Alice", "age": 25, "city": "NY"}
{"name": "Bob", "age": 30, "city": "LA"}
{"name": "Charlie", "age": 35, "city": "NY"}
"""
with open("people.json", "w") as f:
    f.write(json_data)

# 读取 JSON
df = spark.read.json("people.json")
df.printSchema()
df.show()

# 注册临时视图进行交互式分析
df.createOrReplaceTempView("people")
result = spark.sql("SELECT city, AVG(age) as avg_age FROM people GROUP BY city")
result.show()

cover: 附件/Halo封面/database.jpg

模块六：Spark 调优与数据倾斜处理（重难点）

1. Spark 调优通用方向

资源调优：增加Executor内存、核心数、并行度（spark.default.parallelism）。
算子调优：使用 mapPartitions 代替 map，使用 foreachPartitions 代替 foreach，使用 reduceByKey 代替 groupByKey，使用 filter 后 coalesce 减少分区。
序列化调优：使用 KryoSerializer。
内存调优：调节统一内存管理比例（spark.memory.fraction）。
数据倾斜调优（重点）

2. 数据倾斜（Data Skew）

问题发现与定位：查看Spark Web UI，发现某个或某几个Task执行时间特别长（长尾效应），或者发生OOM。通常是由于某个Key的数据量过大导致。

如何缓解数据倾斜（完整PySpark代码演示核心逻辑）：

方案一：提高 Shuffle 并行度

增加 spark.sql.shuffle.partitions（默认200），让原本分配给一个Task的多个Key分配给多个Task，从而减轻单个Task的压力。

spark.conf.set("spark.sql.shuffle.partitions", "1000")

方案二：过滤异常数据

如果倾斜的Key是无效数据（如null，或者测试数据），直接过滤掉。

# 假设倾斜的是空字符串
df_filtered = df.filter("join_key != ''")

方案三：自定义 Partitioner (针对RDD)

将原本倾斜的Key打散。例如，给Key加上随机前缀。

import random
# 给大表的倾斜key加上随机前缀（0-9）
rdd_big = sc.parallelize([("hot_key", 1), ("hot_key", 2), ("normal_key", 3)])
rdd_big_salted = rdd_big.map(lambda x: (str(random.randint(0, 9)) + "_" + x[0], x[1]))
# 小表需要扩大N倍，每条数据都加前缀
rdd_small = sc.parallelize([("hot_key", "A"), ("normal_key", "B")])
rdd_small_expanded = rdd_small.flatMap(lambda x: [(str(i) + "_" + x[0], x[1]) for i in range(10)])
# 然后进行Join
joined = rdd_big_salted.join(rdd_small_expanded)
# 去掉前缀后再进行后续聚合
result = joined.map(lambda x: (x[0].split("_", 1)[1], (x[1][0], x[1][1])))
print(result.collect())

方案四：Map 端 Join (Broadcast Join)

小表扩大N倍（或者直接广播小表），大表不用动。

from pyspark.sql.functions import broadcast
# 小表被广播，大表不发生Shuffle
df_large.join(broadcast(df_small), "join_key")

方案五：加盐局部聚合 + 去盐全局聚合（两阶段聚合）

适用于 groupByKey 类型的聚合操作。先给Key加随机前缀进行局部聚合，再去掉前缀进行全局聚合。

# 假设 rdd 为 [("hot", 1), ("hot", 1), ("hot", 1)…]
rdd = sc.parallelize([("hot", 1), ("hot", 1), ("hot", 1), ("normal", 1)])

# 1. 加盐局部聚合
rdd_salted = rdd.map(lambda x: (str(random.randint(0, 9)) + "_" + x[0], x[1]))
local_agg = rdd_salted.reduceByKey(lambda x, y: x + y)
# 此时结果类似: [("1_hot", 2), ("5_hot", 1), ("3_normal", 1)]

# 2. 去盐全局聚合
result = local_agg.map(lambda x: (x[0].split("_", 1)[1], x[1])).reduceByKey(lambda x, y: x + y)
print(result.collect()) # [('hot', 3), ('normal', 1)]

方案六：Reduce 端 Join 转化为 Map 端 Join

同方案四，使用 Broadcast。

cover: 附件/Halo封面/database.jpg

复习建议

手写代码：重点面试常考 WordCount、TopN、两表Join 以及 数据倾斜加盐处理，要能达到白板手写 PySpark 的水平。
理解原理：不要只背概念。要能讲清楚 reduceByKey 为什么比 groupByKey 好？宽窄依赖是怎么划分Stage的？Shuffle过程具体是怎么落盘和拉取的？
结合场景：对于调优部分，要理解“为什么会倾斜”以及“每种解决方案的优缺点和适用场景”（例如加盐会导致小表膨胀占用内存，Broadcast要求小表不能太大）。

Spark 核心知识点复习大全（基于PySpark）

模块一：Spark 基础与架构

1. Spark 的架构与核心组件

2. 四种部署模式

3. Spark 任务执行流程（重点）

cover: 附件/Halo封面/database.jpg

模块二：RDD 编程核心（转换算子与行动算子）

1. Value 类型转换算子

map vs mapPartitions (重点思考题)

双 Value 类型转换算子

2. Key-Value 类型转换算子

reduceByKey vs groupByKey (重点思考题)

reduceByKey, foldByKey, aggregateByKey, combineByKey 的区别

3. Action 行动算子

cover: 附件/Halo封面/database.jpg

模块三：Shuffle 机制与 Join 策略

1. Spark Shuffle（重点）

2. Spark Join 的三种策略

cover: 附件/Halo封面/database.jpg

模块四：Spark 核心概念与容错机制

1. RDD、DataFrame、DataSet 的区别

2. RDD 的理解与创建方式

3. 宽窄依赖

4. RDD 容错机制

5. 其他核心概念

cover: 附件/Halo封面/database.jpg

模块五：Spark SQL 与 DataFrame

1. Spark SQL 是什么？与传统 SQL 的区别

2. 核心组件与 Hive 集成

3. 处理 JSON 数据与交互式分析

cover: 附件/Halo封面/database.jpg

模块六：Spark 调优与数据倾斜处理（重难点）

1. Spark 调优通用方向

2. 数据倾斜（Data Skew）

方案一：提高 Shuffle 并行度

方案二：过滤异常数据

方案三：自定义 Partitioner (针对RDD)

方案四：Map 端 Join (Broadcast Join)

方案五：加盐局部聚合 + 去盐全局聚合（两阶段聚合）

方案六：Reduce 端 Join 转化为 Map 端 Join

cover: 附件/Halo封面/database.jpg

复习建议

Spark 核心知识点复习大全

文章目录

`map` vs `mapPartitions` (重点思考题)

`reduceByKey` vs `groupByKey` (重点思考题)

`reduceByKey`, `foldByKey`, `aggregateByKey`, `combineByKey` 的区别