Spark 3.5 DataFrame 优化：避免 Shuffle 与提升 SQL 执行效率的技巧

在 Spark 3.5 中，通过避免 Shuffle 和优化 SQL 执行可显著提升性能。以下是关键技巧：

---

一、避免 Shuffle 的核心方法

1. 广播 Join（Broadcast Hash Join）
   当小表（<100MB）与大表关联时，将小表广播到所有 Executor：

   spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100MB")  # 自动广播
   # 或手动广播
   small_df = spark.table("small_table")
   large_df.join(broadcast(small_df), "key")  # 避免 Shuffle
   

2. 分区优化

   • 使用 coalesce() 而非 repartition() 减少分区（无 Shuffle）：

     df.coalesce(4)  # 合并到 4 个分区
     

   • 提前按 Join Key 分区：

     df.repartitionByRange(100, "key")  # 避免 Join 时的 Shuffle
     

3. 使用 Bucketing
   将数据预分桶存储，相同 Bucket ID 的数据在相同分区：

   df.write.bucketBy(32, "key").saveAsTable("bucketed_table")  # 写入时分桶
   spark.table("bucketed_table").join(spark.table("bucketed_table2"), "key")  # 无 Shuffle Join
   

---

二、SQL 执行效率优化

1. 启用 AQE（Adaptive Query Execution）
   Spark 3.0+ 默认开启，动态优化执行计划：

   spark.conf.set("spark.sql.adaptive.enabled", "true")  # 合并小分区、优化 Join 策略
   

2. 列裁剪与谓词下推

   • 仅选择必要列：SELECT col1, col2 FROM ...
   • 提前过滤数据：

     SELECT * FROM table WHERE date > '2023-01-01'  -- 谓词下推减少扫描量
     

3. 避免 UDF，使用内置函数
   内置函数（如 regexp_extract()）比 Python UDF 快 10 倍以上：

   -- 优于自定义 UDF
   SELECT regexp_extract(description, '\\d+') AS id FROM table
   

4. 优化聚合操作
   使用 approx_count_distinct() 替代 count(distinct)：

   SELECT approx_count_distinct(user_id) FROM logs  -- 减少 Shuffle 数据量
   

---

三、配置与监控技巧

1. 关键参数调整

   spark.conf.set("spark.sql.shuffle.partitions", "200")  # 根据数据量调整（默认 200）
   spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")  # 自动合并小分区
   

2. 监控 Shuffle 行为
   通过 Spark UI 检查：

   • Shuffle Read/Write：避免 >1GB 的 Shuffle 数据
   • 任务执行时间：长尾任务需优化分区

3. 数据格式优化
   使用列式存储（Parquet/ORC）并启用压缩：

   df.write.parquet("path", compression="snappy")  # 减少 I/O 开销
   

---

示例：避免 Shuffle 的 Join

# 场景：大表 orders 关联小表 users
users = spark.table("users").filter("country='US'")  # 先过滤小表
orders = spark.table("orders")

# 广播 Join（无 Shuffle）
result = orders.join(broadcast(users), "user_id")

# 检查执行计划（应显示 BroadcastHashJoin）
result.explain()

---

最佳实践总结

场景	优化方案
大小表 Join	广播小表
大表 Join 大表	分桶或预分区
聚合操作	启用 AQE + 调整分区数
重复使用 DataFrame	缓存（cache()）
复杂计算	优先用内置函数，避免 UDF

通过组合使用这些技巧，可显著降低 Shuffle 开销并提升 SQL 执行效率。