← 返回文章列表

Spark SQL 性能调优实战指南

2025-04-07 Spark

Spark SQL 是大数据处理中最常用的计算引擎之一，但在实际生产环境中，我们经常会遇到性能瓶颈。本文将从多个维度深入探讨 Spark SQL 的性能优化技巧，帮助你打造高效的数据处理 pipeline。

一、数据分区策略优化

1.1 合理的分区数设置

分区数直接影响并行度和任务调度效率。分区过少会导致资源利用率低，分区过多则会增加调度开销。

// 推荐的分区数计算公式
val optimalPartitionNum = math.max(2, totalDataSize / 128MB)

// 重新分区
spark.sql.shuffle.partitions = 200
spark.conf.set("spark.sql.shuffle.partitions", "400")

1.2 分区字段选择

选择合适的分区字段可以大幅减少数据倾斜和 shuffle 开销：

  • 优先选择基数适中的字段（100-10000个不同值）
  • 避免使用日期字段作为唯一分区键（会导致小文件问题）
  • 组合分区：日期 + 业务类型

二、缓存与持久化策略

2.1 何时使用缓存

缓存适用于以下场景：

  • DataFrame 被多次重复使用
  • 迭代算法中的中间结果
  • 昂贵的计算结果（如复杂的 JOIN 操作）

2.2 存储级别选择

// 内存缓存
val cachedDF = expensiveDF.cache()

// 内存+磁盘序列化
expensiveDF.persist(StorageLevel.MEMORY_AND_DISK_SER)

// 使用完毕后释放缓存
cachedDF.unpersist()

三、广播变量优化

广播变量是解决小表 JOIN 大表性能问题的利器。当小表数据量小于 spark.sql.autoBroadcastJoinThreshold（默认10MB）时，Spark 会自动使用广播 JOIN。

// 手动广播小表
import org.apache.spark.sql.functions.broadcast
val result = largeDF.join(broadcast(smallDF), "join_key")

// 调整自动广播阈值
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "50MB")

四、数据倾斜处理

4.1 识别数据倾斜

通过 Spark UI 观察任务执行情况，如果发现某些 task 执行时间远长于其他 task，则可能存在数据倾斜。

4.2 加盐处理方案

// 给热点 key 添加随机前缀
val saltedDF = df.withColumn("salt_key", 
    concat(col("key"), lit("_"), (rand() * 10).cast("int")))

// 扩容小表，与加盐后的表 JOIN
val expandedSmallDF = smallDF
    .withColumn("salt", explode(lit((0 to 9).toArray)))
    .withColumn("salt_key", concat(col("key"), lit("_"), col("salt")))

五、文件格式与压缩

5.1 推荐文件格式

格式 优点 适用场景
Parquet 列式存储、高效压缩、谓词下推 分析型查询
ORC Hive 原生支持、轻量级索引 Hive 生态
Delta Lake ACID、版本控制、Time Travel 数据湖

5.2 压缩算法选择

// Parquet 使用 Snappy 压缩（推荐）
spark.conf.set("spark.sql.parquet.compression.codec", "snappy")

// 高压缩比场景使用 GZIP
spark.conf.set("spark.sql.parquet.compression.codec", "gzip")

六、SQL 优化技巧

6.1 谓词下推

确保过滤条件能够下推到数据源，减少数据读取量：

// 好的写法：先过滤再 JOIN
val result = largeDF.filter(col("date") >= "2025-01-01")
    .join(smallDF, "key")

// 避免：先 JOIN 再过滤
val result = largeDF.join(smallDF, "key")
    .filter(col("date") >= "2025-01-01")

6.2 列裁剪

// 好的写法：只选择需要的列
val result = df.select("user_id", "amount", "status")

// 避免：select * 后再过滤列
val result = df.selectExpr("*").select("user_id", "amount")

6.3 避免使用 UDF

UDF 会阻止 Catalyst 优化器进行优化，尽量使用 Spark SQL 内置函数：

// 避免使用 UDF
val upperUDF = udf((s: String) => s.toUpperCase)
df.withColumn("upper_name", upperUDF(col("name")))

// 使用内置函数
df.withColumn("upper_name", upper(col("name")))

七、Spark 配置优化

7.1 内存配置

// Executor 内存配置
spark.executor.memory = 8g
spark.executor.memoryOverhead = 2g

// Driver 内存配置
spark.driver.memory = 4g
spark.driver.memoryOverhead = 1g

// 堆外内存
spark.memory.offHeap.enabled = true
spark.memory.offHeap.size = 4g

7.2 GC 优化

// 使用 G1GC
spark.executor.extraJavaOptions = "-XX:+UseG1GC -XX:MaxGCPauseMillis=200"

// 调整 GC 参数
spark.executor.extraJavaOptions = "-XX:InitiatingHeapOccupancyPercent=35"

八、监控与诊断

8.1 Spark UI 关键指标

  • Jobs：查看任务执行情况
  • Stages：查看 stage 详情，识别数据倾斜
  • Storage：查看缓存数据
  • SQL：查看 SQL 执行计划

8.2 执行计划分析

// 查看逻辑执行计划
df.explain()

// 查看物理执行计划
df.explain(true)

// 查看 Spark SQL 执行计划
spark.sql("EXPLAIN EXTENDED SELECT * FROM table").show(false)

九、最佳实践总结

  1. 合理设置分区数：根据数据量调整 shuffle 分区数
  2. 善用广播 JOIN：小表广播，避免 shuffle
  3. 处理数据倾斜：加盐、拆分热点 key
  4. 选择合适文件格式：Parquet + Snappy 是通用选择
  5. 及时释放缓存：避免内存溢出
  6. 使用内置函数：避免 UDF 阻止优化
  7. 关注执行计划：定期分析执行计划，发现优化点

Spark SQL 性能调优是一个持续的过程，需要在实际生产环境中不断积累经验。希望本文能帮助你更好地优化 Spark SQL 作业！