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 joinfull 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.enabledtrue后会自动检测并拆分倾斜分区。但实际使用中 AQE 的倾斜判断阈值需要根据作业特征调整全默认配置未必最优。