Spark数据倾斜优化求助：大表小表Join任务阻塞问题

1. 精准排查数据倾斜根因 #

• 定位热点分区：通过Spark UI的Stage页面查看Task的输入数据量，聚焦执行时间最长、数据量最大的Task，确认倾斜源头。同时检查Join键的分布，重点关注占比极高的Key或Null值（Null会被统一分配到同一分区）。
• 验证分区均匀性：执行以下代码确认加盐/重分区后的实际数据分布，若单个分区数据量远超平均水平（如5倍以上），说明现有策略未解决倾斜：

  # 检查大表分区数据量分布
  large_df.groupBy(spark_partition_id()).count().orderBy(col("count").desc()).show(20)
  # 检查加盐后的Join键+盐值组合分布
  large_df.groupBy("join_key", "salt").count().orderBy(col("count").desc()).show(20)
  

2. 优化Join策略细节 #

热点Key拆分处理 #

针对极端高频的Join Key，单独拆分后打散处理，避免热点集中：

from pyspark.sql.functions import col, floor, rand

# 筛选高频Key（阈值根据实际数据调整）
hot_keys = small_df.groupBy("join_key").count().filter("count > 10000").select("join_key").collect()
hot_keys_list = [row.join_key for row in hot_keys]

# 拆分热点/冷点数据集
small_hot = small_df.filter(col("join_key").isin(hot_keys_list))
large_hot = large_df.filter(col("join_key").isin(hot_keys_list))
small_cold = small_df.filter(~col("join_key").isin(hot_keys_list))
large_cold = large_df.filter(~col("join_key").isin(hot_keys_list))

# 热点数据集加盐打散后Join
salt_num = 10  # 盐值范围可根据热点数据量调整
small_hot_salted = small_hot.withColumn("salt", floor(rand() * salt_num))
large_hot_salted = large_hot.withColumn("salt", floor(rand() * salt_num))
joined_hot = small_hot_salted.join(large_hot_salted, on=["join_key", "salt"]).drop("salt")

# 冷点数据集采用广播Join（若冷点小表数据量适合广播）
joined_cold = small_cold.join(large_cold, on="join_key")

# 合并最终结果
final_df = joined_hot.unionByName(joined_cold)

精细化广播Join #

若小表单条数据体积较大（含大字段/冗余列），仅广播Join必需字段，减少内存占用：

# 仅保留Join键和业务必要字段
small_df_trimmed = small_df.select("join_key", "required_col1", "required_col2")
joined_df = small_df_trimmed.join(large_df, on="join_key")

同时关闭自动广播阈值，手动控制Join策略：

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)

重分区策略优化 #

• 结合Join键与盐值重分区，确保数据均匀分布：

  # 分区数建议为Worker总核数的2-3倍（如40核设置80-120）
  large_df_repartitioned = large_df.withColumn("salt", floor(rand() * 10)).repartition(120, "join_key", "salt")
  

• 避免盲目设置过多分区，分区数过大会增加Shuffle开销。

3. 调整Spark内存与Shuffle参数 #

• 优化Shuffle分区数：针对2000万行数据，建议设置为400-800：

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

• 调整内存分配比例，增加执行内存占比以减少磁盘溢写：

  spark.conf.set("spark.memory.fraction", 0.7)
  spark.conf.set("spark.memory.storageFraction", 0.2)
  

• 优先启用Sort-Merge Join：

  spark.conf.set("spark.sql.join.preferSortMergeJoin", "true")
  

4. 数据预处理与Databricks专属优化 #

• 清理无效数据：过滤Join键为Null的行，避免Null值导致的分区倾斜：

  small_df_clean = small_df.filter(col("join_key").isNotNull())
  large_df_clean = large_df.filter(col("join_key").isNotNull())
  

• Delta表布局优化：若大表为Delta格式，执行Optimize+ZORDER减少Join时的IO开销：

  OPTIMIZE large_delta_table ZORDER BY join_key;
  

• 启用Photon引擎：若集群版本支持，开启Photon加速计算：

  spark.conf.set("spark.databricks.photon.enabled", "true")
  

内容的提问来源于stack exchange，提问作者MOKRANI