第19课：PySpark｜性能全维度调优【参数调优、SQL调优、数据倾斜根治方案】

一、课前导读

经过前面18节课的学习，你已经掌握了PySpark的绝大多数核心技能：从RDD、DataFrame到Spark SQL，从离线数仓到实时流计算。你写的程序已经能够正确地运行，并输出期望的结果。但是，在真正的生产环境中，“能跑”只是第一步，“跑得快”才是关键。你可能遇到过这样的问题：

• 明明集群有几百个核心，但任务执行时却只有一个Task在运行，其他人都在等待。
• 一个小小的Join操作，居然触发了TB级的Shuffle，磁盘被撑爆。
• 99%的Task都在几秒内完成，却有1%的Task卡了半个小时，拖垮了整个作业。
• 开启动态资源分配后，Executor不断申请和释放，浪费了大量资源。

这些问题归根结底都是性能问题。Spark的性能调优是一个系统工程，涉及资源配置、数据序列化、内存管理、并行度设置、数据倾斜处理、SQL优化等方方面面。很多开发者只知其然不知其所以然，盲目调整参数，往往事倍功半。

本节课将系统性地讲解PySpark性能调优的完整知识体系。我们会从资源参数调优（Executor、内存、并行度）开始，然后深入Spark SQL优化（Catalyst、AQE、分区剪枝），最后重点攻克大数据领域最棘手的问题——数据倾斜，提供多种行之有效的治理方案（加盐打散、广播Join、自定义分区器等）。学完这节课，你将具备诊断性能瓶颈、主动优化Spark作业的能力，成为一名真正的PySpark专家。

二、学习目标

完成本节课的学习后，你将能够：

1. 掌握核心资源配置参数：合理设置Executor数量、内存、核心数、并行度等
2. 理解内存管理模型：区分堆内内存、堆外内存，调优spark.memory.fraction等参数
3. 优化序列化：使用Kryo序列化替代Java序列化，提升网络传输效率
4. 应用Spark SQL优化技巧：利用AQE、动态分区剪裁、Join策略选择等
5. 根治数据倾斜：识别倾斜场景，使用加盐、广播、自定义分区器等多种手段解决
6. 优化Shuffle：减少Shuffle数据量，使用map端预聚合、调整shuffle分区数
7. 使用监控工具：通过Spark UI、事件日志、Metrics系统定位性能瓶颈
8. 综合调优案例：完成一个从慢到快的真实业务优化过程

三、核心理论知识点

四、原理通俗讲解

4.1 资源调优：让你的钱花在刀刃上

集群资源是有限的，就像你有一笔预算，需要合理地分配给各个项目。如果给一个任务分配了太多Executor，其他任务可能无资源可用；如果分配太少，任务又跑得慢。关键是要平衡并行度和资源利用率。

Executor vs Core vs Memory：

• 每个Executor是一个JVM进程，负责执行Task。过多的JVM进程会增加调度开销。
• 每个Core可以并发执行一个Task。一个Executor的Core数决定了它可以同时跑多少个Task。
• 内存大小决定了能缓存多少数据、能容纳多大的Shuffle数据。

调优黄金法则：Executor总数 = 总Core数 / 每个Executor的Core数，每个Executor内存建议4-8GB，Core数建议3-5个。

4.2 内存管理：Spark的“厨房”怎么安排？

Spark Executor的内存就像一个大厨房，分为几个区域：

• 执行内存：炒菜区，用于Shuffle、Join、排序等操作，需要频繁“颠勺”。
• 存储内存：冰箱，用于缓存RDD、DataFrame、广播变量。
• 用户内存：存放用户数据结构和UDF内部对象。

Spark 1.6+的统一内存管理模型允许执行内存和存储内存互相借用（如果对方空闲），提高了内存利用率。但如果配置不当，可能导致频繁的垃圾回收（GC）或溢写磁盘。

4.3 数据倾斜：少数“钉子户”拖垮整个任务

数据倾斜是最常见也最头疼的性能问题。想象一下，你要统计每个省份的人口，但某个省份（比如广东）的人口远多于其他省份，那么负责处理广东数据的那个Task就会非常慢，其他Task早已完成，只能干等。

Spark中的倾斜通常发生在Shuffle阶段：相同key的数据汇聚到同一个分区，导致该分区的数据量巨大。解决思路就是打散：给倾斜的key添加随机前缀，让它们分散到多个分区，然后分两步聚合。

五、重点概念拆解

5.1 核心资源配置参数详解

spark-submit\--masteryarn\--deploy-mode cluster\--num-executors20\# Executor数量（YARN模式）--executor-cores4\# 每个Executor的CPU核心数--executor-memory 8g\# 每个Executor的堆内存--confspark.executor.memoryOverhead=2g\# 堆外内存（占总内存10-20%）--confspark.driver.memory=4g\# Driver内存--confspark.default.parallelism=100\# 默认并行度（Shuffle后分区数）--confspark.sql.shuffle.partitions=200# SQL中Shuffle分区数

调优建议：

• Executor内存不要超过32GB（否则JVM GC暂停时间过长）。推荐8-16GB。
• Executor的Core数通常取3-5，避免超过5导致HDFS I/O吞吐瓶颈。
• 并行度（分区数）应设置为总Core数的2-3倍，让资源充分利用且减少调度开销。
• 堆外内存至少分配10%（尤其在使用UDF或处理大字段时）。

5.2 内存管理参数

5.3 序列化优化

默认使用Java序列化（org.apache.spark.serializer.JavaSerializer），性能较差。建议使用Kryo序列化：

spark=SparkSession.builder \.config("spark.serializer","org.apache.spark.serializer.KryoSerializer")\.config("spark.kryo.registrationRequired","false")\.config("spark.kryo.unsafe","true")\.getOrCreate()

Kryo可以将数据序列化为更紧凑的字节数组，速度也更快。如果知道要序列化的类，可以注册以进一步提升性能。

5.4 Shuffle优化参数

5.5 SQL优化要点

开启AQE（Adaptive Query Execution）：Spark 3.x默认开启，能够动态合并shuffle分区、动态调整Join策略、处理数据倾斜。

spark.conf.set("spark.sql.adaptive.enabled","true")spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled","true")spark.conf.set("spark.sql.adaptive.skewJoin.enabled","true")spark.conf.set("spark.sql.autoBroadcastJoinThreshold","10485760")# 10MB

Join策略选择：

• 小表（<10MB）应使用广播Join（BroadcastHashJoin），避免Shuffle。
• 中等表（10MB-100MB）可考虑广播。
• 大表对大表使用SortMergeJoin。

分区剪枝：在查询中务必加上分区字段的过滤，如WHERE dt='2024-01-01'。

列剪枝：只SELECT需要的列，避免读取无关数据。

5.6 数据倾斜治理方案

方案一：加盐打散（两阶段聚合）
适用于groupBy/reduceByKey等聚合操作。

# 给key加随机前缀（0-9）df_with_salt=df.withColumn("salted_key",concat(col("key"),lit("_"),(rand()*10).cast("int")))# 第一次聚合（局部）partial=df_with_salt.groupBy("salted_key").agg(sum("value"))# 去掉盐，第二次聚合result=partial.withColumn("original_key",split(col("salted_key"),"_")[0])\.groupBy("original_key").agg(sum("sum(value)"))

方案二：广播Join
适用于大小表Join，倾斜的key存在于大表，但小表可以广播。

frompyspark.sql.functionsimportbroadcast result=large_df.join(broadcast(small_df),"key")

方案三：拆分倾斜key
将倾斜的key单独处理，与普通key分开Join后再合并。

# 识别热点key（例如出现次数>1000）hot_keys=df.groupBy("key").count().filter("count > 1000").select("key").collect()# 分离热点和非热点数据hot_df=df.filter(col("key").isin([k[0]forkinhot_keys]))normal_df=df.filter(notcol("key").isin([k[0]forkinhot_keys]))# 对热点数据用广播Join或其他方式hot_result=hot_df.join(broadcast(dim_df),"key")normal_result=normal_df.join(dim_df,"key")# 合并result=hot_result.union(normal_result)

方案四：自定义分区器
对于RDD或DataFrame的repartitionByRange，可以自定义分区逻辑，将热点key分散。

六、易错点避坑

6.1 盲目增加Executor数量

增加Executor会导致调度开销增大，且每个Executor都会申请内存，可能超出集群容量。应先增加Core数或并行度，而非Executor数。

6.2 忽略数据本地性

如果Task被调度到没有数据的节点，需要从远程拉取数据，产生网络开销。应观察Spark UI的Locality Level，若大量为NODE_LOCAL或RACK_LOCAL，考虑调整spark.locality.wait参数。

6.3 动态资源分配关闭导致资源浪费

生产环境强烈建议开启动态资源分配，配合External Shuffle Service，让空闲Executor自动释放。

6.4 过早使用collect()或toPandas()

在调试时使用collect()或toPandas()会将所有数据拉到Driver，大数据集下必定OOM。应使用take()或采样。

6.5 忽视小文件问题

写入数据时如果分区数过多或每个分区数据量太小，会产生大量小文件，给HDFS NameNode带来压力。应使用coalesce()控制输出文件数。

七、完整实战案例

本案例将从一个慢速的Spark作业开始，逐步应用各种调优技巧，最终使其性能提升5倍以上。我们模拟一个常见的电商场景：计算每个商品的销售总额，并关联商品类别。

7.1 初始代码（未优化）

# ============== optimization_before.py ==============# 未优化的版本：全量读取、Shuffle大、数据倾斜未处理frompyspark.sqlimportSparkSessionfrompyspark.sql.functionsimportcol,sumasspark_sumimporttime spark=SparkSession.builder \.appName("OptimizationBefore")\.master("yarn")\.getOrCreate()# 生成模拟数据：1亿条销售记录，其中某个商品ID（P_9999）占比30%（倾斜）# 实际生产应从Hive读取，这里简化data=[]foriinrange(100_000_000):product_id=f"P_{i%100000}"ifi%3==0:# 约33%的倾斜product_id="P_9999"amount=random.randint(1,1000)data.append((product_id,amount))df=spark.createDataFrame(data,["product_id","amount"])# 简单的groupBy聚合start=time.time()result=df.groupBy("product_id").agg(spark_sum("amount").alias("total_amount"))result.count()# 触发行动print(f"执行耗时:{time.time()-start}秒")

问题：单次Shuffle，数据倾斜严重，未使用AQE，未设置合适的shuffle分区数。

7.2 优化后代码

# ============== optimization_after.py ==============# 优化版本：AQE、广播变量、加盐打散、参数调优frompyspark.sqlimportSparkSessionfrompyspark.sql.functionsimportcol,sumasspark_sum,concat,lit,rand,split,exprfrompyspark.sql.typesimport*importtime# 创建SparkSession时配置大量优化参数spark=SparkSession.builder \.appName("OptimizationAfter")\.master("yarn")\.config("spark.sql.shuffle.partitions","200")\.config("spark.sql.adaptive.enabled","true")\.config("spark.sql.adaptive.coalescePartitions.enabled","true")\.config("spark.sql.adaptive.skewJoin.enabled","true")\.config("spark.sql.autoBroadcastJoinThreshold","10485760")\.config("spark.serializer","org.apache.spark.serializer.KryoSerializer")\.config("spark.sql.adaptive.skewedJoinThreshold","10485760")\.config("spark.dynamicAllocation.enabled","true")\.config("spark.dynamicAllocation.minExecutors","5")\.config("spark.dynamicAllocation.maxExecutors","50")\.getOrCreate()# 生成相同的数据（略，与之前一致）# 方法1：直接依靠AQE处理倾斜（Spark 3.x自动处理）start=time.time()result=df.groupBy("product_id").agg(spark_sum("amount").alias("total_amount"))result.count()print(f"仅开启AQE耗时:{time.time()-start}秒")# 方法2：手动加盐打散（更彻底）print("\n使用加盐打散...")# 给product_id加随机盐（0-9）salted_df=df.withColumn("salted_key",concat(col("product_id"),lit("_"),(rand()*10).cast("int")))# 第一次聚合partial=salted_df.groupBy("salted_key").agg(spark_sum("amount").alias("partial_sum"))# 去掉盐，二次聚合final=partial.withColumn("product_id",split(col("salted_key"),"_")[0])\.groupBy("product_id").agg(spark_sum("partial_sum").alias("total_amount"))final.count()print(f"加盐打散耗时:{time.time()-start}秒")# 验证结果一致性spark.stop()

7.3 SQL调优示例

假设我们有订单表（orders）和用户表（users），需要统计每个用户的订单总额。

# 未优化SQLsql=""" SELECT u.user_id, SUM(o.amount) as total FROM orders o JOIN users u ON o.user_id = u.user_id GROUP BY u.user_id """df=spark.sql(sql)# 优化后：启用AQE，使用广播Join（如果users表小于10MB）# 自动识别小表并广播，无需改写

7.4 监控与诊断

通过Spark UI观察：

• Stages页面：查看长尾Task，确认数据倾斜。
• Storage页面：查看缓存命中率。
• SQL页面：查看物理计划，确认是否使用了BroadcastJoin、分区裁剪等。
• Executors页面：查看GC时间、Shuffle读写量，判断内存是否不足。

八、代码逐行解析

8.1 资源配置参数

.config("spark.sql.shuffle.partitions","200")

设置Shuffle分区数，避免默认200对于小数据量过多或对于大数据量过少。

8.2 AQE关键参数

.config("spark.sql.adaptive.enabled","true").config("spark.sql.adaptive.coalescePartitions.enabled","true").config("spark.sql.adaptive.skewJoin.enabled","true")

AQE会在运行时动态优化：合并小分区、处理倾斜Join、动态切换Join策略。

8.3 加盐打散实现

salted_df=df.withColumn("salted_key",concat(col("product_id"),lit("_"),(rand()*10).cast("int")))partial=salted_df.groupBy("salted_key").agg(spark_sum("amount"))final=partial.withColumn("product_id",split("salted_key","_")[0]).groupBy("product_id").agg(spark_sum("partial_sum"))

通过随机盐将倾斜key分散到10个分区，先局部聚合，再去盐全局聚合。注意：盐的粒度需根据倾斜程度调整。

九、业务场景落地应用

9.1 场景一：大表Join大表优化

当两张大表Join时，无法广播。常见优化：

• 使用分桶（bucket）预先存储数据，避免Shuffle。
• 使用相同的分区器，让数据在物理上对齐。
• 使用Bloom Filter预过滤。

9.2 场景二：多维度聚合中的倾斜

使用rollup或cube时，某些维度组合可能产生大量数据。可使用spark.sql.adaptive.skewJoin.enabled自动处理。

9.3 场景三：流计算中的数据倾斜

Structured Streaming中，窗口聚合也可能倾斜。可以使用加盐或增加分区。

9.4 场景四：机器学习特征工程中的倾斜

在One-Hot编码或TF-IDF中，高频词可能导致倾斜。使用局部聚合或采样。

十、常见报错排查

10.1Container killed by YARN for exceeding memory limits

原因：Executor实际内存超过申请值（包括堆外内存）。

解决：增加spark.executor.memoryOverhead，或减少spark.executor.memory。

10.2FetchFailedException

原因：Shuffle数据拉取失败，可能是节点故障或网络问题。

解决：增加重试次数spark.shuffle.io.maxRetries，开启外部shuffle服务。

10.3OutOfMemoryError: GC overhead limit exceeded

原因：JVM GC时间过长，通常因为内存中对象过多（如缓存了大量数据）。

解决：减少缓存数据量，使用MEMORY_AND_DISK，或增加内存。

10.4org.apache.spark.shuffle.MetadataFetchFailedException

原因：Shuffle阶段某个Map任务输出丢失。

解决：检查Executor日志，排查磁盘故障；增加任务重试次数。

十一、本节课知识点总结

调优参数速查表

数据倾斜治理决策树

是否Join引起倾斜？ ├─ 是 → 小表是否<10MB？ │ ├─ 是 → 广播Join │ └─ 否 → 拆分倾斜key单独处理 或 加盐打散（Join需特殊处理） └─ 否（聚合引起） → 加盐两阶段聚合

性能优化流程

1. 基准测试：运行作业，记录耗时和资源消耗。
2. 监控分析：通过Spark UI定位瓶颈（长尾Task、大Shuffle、高GC）。
3. 参数调优：调整资源配置、并行度、内存、序列化。
4. 代码优化：减少Shuffle、使用内置函数、避免UDF、优化Join。
5. 倾斜处理：识别倾斜key，选择合适方案。
6. 验证迭代：对比优化前后效果。

十二、课后思考作业

作业一：理论理解题

1. 解释Spark统一内存管理模型中执行内存和存储内存的借用机制，以及可能导致的相互淘汰问题。

2. AQE是如何动态合并Shuffle分区的？它如何判断一个分区是否需要合并？

3. 请列举至少三种数据倾斜的场景，并分别说明适合的解决方案。

作业二：代码实践题

1. 编写一个PySpark程序，生成一个严重倾斜的数据集（一个key占80%），分别使用加盐两阶段聚合和AQE自动处理（Spark 3.x），对比执行时间和Shuffle数据量。

2. 使用explain查看一个Join的执行计划，判断是否使用了BroadcastJoin。如果没有，如何强制广播？

3. 模拟一个Shuffle Spill的场景（内存不足导致溢写磁盘），调整spark.sql.shuffle.partitions和spark.memory.fraction观察溢写量的变化。

作业三：场景应用题

某日志分析平台每天处理TB级日志，核心计算是SELECT device_id, COUNT(*) FROM logs GROUP BY device_id。目前任务运行在100节点集群（每个节点16核64G），发现Shuffle阶段严重倾斜，少数设备ID（如空字符串）占50%数据。

请给出完整的优化方案，包括：

1. 如何识别倾斜的设备ID？
2. 加盐打散的具体实现（写出代码）。
3. 资源配置参数建议（executor数量、内存、分区数等）。
4. 如何验证优化效果？

作业四：拓展研究

1. 深入研究Spark SQL的AQE源码，了解CoalesceShufflePartitions和OptimizeSkewedJoin的实现原理。

2. 学习使用Spark的Stage和Task级别的Metrics（如执行时间、GC时间、Shuffle读写量），编写一个自动化分析脚本，识别慢任务并给出优化建议。

3. 对比Presto/Trino与Spark SQL在处理数据倾斜上的不同策略，写一份调研报告。

提交方式：本次作业要求提交代码、运行日志截图、Spark UI截图以及调优前后对比数据。

扩展阅读：

• Spark官方文档：Tuning Guide
• 《Spark性能优化实战》- 阿里巴巴技术团队
• Spark源码：sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive

通过本节课的学习，你已经掌握了PySpark性能调优的全套方法论，从资源到代码，从参数到架构。学完这节课，你应该能够从容应对生产环境中的各种性能问题。下一节课我们将进行综合项目实战，结合所有知识完成一个完整的大数据分析项目。我们下节课见！

🔗《20节课 PySpark 从入门到精通》系列课程导航

去订阅

🌟 感谢您耐心阅读到这里！
💡 如果本文对您有所启发欢迎：
👍 点赞📌 收藏 📤 分享给更多需要的伙伴。
🗣️ 期待在评论区看到您的想法, 共同进步。
🔔 关注我，持续获取更多干货内容～
🤗 我们下篇文章见～