Spark 数据倾斜排查实战：广播变量的正确用法与那些不该踩的坑

Spark 数据倾斜排查实战：广播变量的正确用法与那些不该踩的坑

一、一个 Task 跑了 40 分钟，其他 3 秒就结束了

Spark 作业中最让人抓狂的不是 OOM，不是数据丢失，而是数据倾斜。19 个 Task 在两分钟内跑完，剩下那一个跑了 40 分钟还没结束。打开 Spark UI 一看，某个分区的 Shuffle Write 量是其他分区的 200 倍。这就是典型的数据倾斜，几乎每个用过 Spark 的数据工程师都会遇到。

数据倾斜的核心原因是key 分布极度不均匀。在数据分析场景下，最常见的倾斜 key 就是null值、默认值（如'unknown'、'0'），以及少数"热点"维度值（如一线城市的用户量是三四线城市的百倍）。

flowchart TD A[发现数据倾斜] --> B{倾斜发生在哪个阶段?} B -->|Join阶段| C{倾斜Key特征?} B -->|GroupBy/Agg阶段| D{倾斜Key特征?} C -->|少量热点Key| E[加盐打散: 给Key加随机前缀] C -->|大量null值| F[过滤或分离null: 单独处理] C -->|大表Join小表| G[广播Join: 小表塞进内存] D -->|少量热点Key| H[两阶段聚合: 先局部加盐 再去盐聚合] D -->|所有Key都倾斜| I[增加并行度或调整分区策略] G --> J{广播变量大小检查} J -->|小于spark.sql.autoBroadcastJoinThreshold| K[自动广播: 默认10MB] J -->|大于阈值| L[显式broadcast hint + 调大阈值]

二、广播变量不是万能药：三条件缺一不可

广播 Join 是解决数据倾斜最优雅的方案：把小表数据完整发送到每个 Executor 的内存中，大表的分区数据直接在本地做 Map-side Join，完全避免了 Shuffle。

但广播变量生效需要同时满足三个条件，缺一个就会翻车：

条件一：小表必须"真的小"。spark.sql.autoBroadcastJoinThreshold默认 10MB，但这不是说表文件 10MB 就行——Spark 判断的是执行计划中预估的"广播后内存占用"，包含了反序列化的开销，通常比文件大小大 2-3 倍。一个磁盘上 8MB 的 Parquet 文件，广播到内存可能是 20MB。

条件二：不能用full outer join。广播 Join 只支持等值连接中的 inner/left/right join，full outer join必须走 SortMergeJoin，再怎么 hint 都没用。如果业务上确实需要全外连接，得拆成 left + anti join 的组合拳。

条件三：广播不能解决"大表 Join 大表"。两个都超过 1GB 的表，广播任何一方都会导致 Executor OOM。这时候必须回到加盐打散的思路上来。

实战中最常见的翻车场景：开发环境小表只有几千行，自动广播一切正常；上了生产环境小表变成了几十万行，自动广播悄悄失效，作业突然慢 10 倍。排查半天才发现是被自动广播的阈值卡住了。

from pyspark.sql import SparkSession spark = SparkSession.builder \ .config("spark.sql.autoBroadcastJoinThreshold", "104857600") # 100MB .getOrCreate() # 显式使用 broadcast hint from pyspark.sql.functions import broadcast result = large_df.join(broadcast(small_df), "key", "left")

三、加盐打散的工程化实践：盐值不是随便选的

对于无法用广播解决的大表 Join 大表倾斜，加盐打散是主力方案。但"加随机数"说起来简单，做起来细节很多：

盐值范围的选择直接影响效果和开销。盐值太小（如 0-4），散列不够均匀；盐值太大（如 0-99），膨胀后的数据量可能让本来不倾斜的 key 也变慢。一般取 10-50 之间的值，具体要根据倾斜程度调参：

import pyspark.sql.functions as F # 只对倾斜key加盐，正常key不动 SALT_RANGE = 20 hot_keys = ['北京', '上海', '广州'] # 从Spark UI中找出的倾斜key # Step 1: 给倾斜key加随机盐 salted_large = large_df.withColumn( 'salted_key', F.when(F.col('city').isin(hot_keys), F.concat(F.col('city'), F.lit('_'), (F.rand() * SALT_RANGE).cast('int').cast('string'))) .otherwise(F.col('city')) ) # Step 2: 小表每条膨胀SALT_RANGE倍 expanded_small = small_df.crossJoin( spark.range(0, SALT_RANGE).withColumnRenamed('id', 'salt_val') ).withColumn( 'salted_key', F.concat(F.col('city'), F.lit('_'), F.col('salt_val').cast('string')) ).unionByName(small_df.withColumn('salted_key', F.col('city'))) # 注意：此处简化了union的列对齐细节，实际需要处理列名一致性

关键注意：小表的膨胀可能导致小表也变"大"。如果小表原本 50 万行、膨胀 20 倍变成 1000 万行，就要重新评估广播的可行性。这时可能需要将膨胀后的小表也做一次显式广播。

四、倾斜不只是 Join 的专利：GroupBy 聚合也要防

很多人只在 Join 场景下关注倾斜，其实groupBy+ 聚合函数的倾斜同样致命。当某个分组的数量级远超其他分组时，Shuffle 过程中该分组的数据全部进入同一个 Task，造成单点瓶颈。

两阶段聚合是标准解法：第一阶段给 key 加随机前缀做局部聚合（每个 Task 内先聚合一轮），第二阶段去掉随机前缀做最终聚合：

# 两阶段聚合 SALT_RANGE = 50 # Phase 1: 加盐局部聚合 phase1 = df.withColumn( 'salted_key', F.concat(F.col('group_key'), F.lit('_'), (F.rand() * SALT_RANGE).cast('int')) ).groupBy('salted_key').agg(F.sum('value').alias('partial_sum')) # Phase 2: 去盐全局聚合 phase2 = phase1.withColumn( 'original_key', F.split(F.col('salted_key'), '_')[0] ).groupBy('original_key').agg(F.sum('partial_sum').alias('total_sum'))

两阶段聚合的前提是聚合函数可分拆。sum、count、min、max都支持分拆后合并；但avg、count(distinct)、median不能直接分拆——avg需要在第一阶段同时计算sum和count，第二阶段再做除法。

五、总结

数据倾斜的应对策略可以按场景归类：

• 大表 Join 小表：调大广播阈值 + 显式 broadcast hint，避免走 Shuffle Join。
• 大表 Join 大表，少量热点 key：加盐打散热点 key，正常 key 不动。
• 大表 Join 大表，全是倾斜 key：调整分区策略或接受 Shuffle 优化（使用 AQE 的spark.sql.adaptive.skewJoin.enabled）。
• GroupBy 聚合倾斜：两阶段聚合，注意聚合函数是否可分拆。
• null 值倾斜：先过滤或单独处理 null 值，不要让 null 参与 Shuffle。

Spark 3.x 的 AQE（自适应查询执行）内置了倾斜 Join 优化，开启spark.sql.adaptive.skewJoin.enabled=true后会自动检测并拆分倾斜分区。但实际使用中 AQE 的倾斜判断阈值需要根据作业特征调整，全默认配置未必最优。