Spark 动态分区倾斜诊断与非典型 Skew 排查指南

Spark 的 AQE（Adaptive Query Execution）自 3.2 版本起默认开启 skewJoin，能自动检测并拆分倾斜分区。但实践中存在大量 AQE 无法覆盖的非典型倾斜：动态分区导致的数据分布不均、数据类型隐式转换引发的 key 收敛、以及因 join 类型限制导致 skewJoin 未触发。这些场景需要从 Spark UI 的指标体系出发，构建系统化的诊断流程。

非典型倾斜的三种典型模式

1. 动态分区倾斜（Dynamic Partition Skew）

当写入端使用动态分区（spark.sql.sources.partitionOverwriteMode=dynamic）时，不同分区的数据量差异可能极大。典型场景是按天分区的增量表中，某一天的数据量是其他日期的数十倍。

这种倾斜的特征是：Shuffle Write 阶段各 Task 输出量差异不大，但写入阶段部分 Partition Writer 的输出文件体积远大于其他分区。在 Spark UI 的 Stage 详情页中，表现为 Task 的 Shuffle Write Size 均匀，但写入 HDFS 的 File Size 呈现明显长尾。

诊断步骤：在 Spark UI 的 SQL 页面找到涉及动态分区写入的 Stage；展开 Stage 查看 Shuffle Write Size by Task 分布；如果写入侧均匀但最终文件不均，则进入目录层检查各分区大小。修复手段包括：预聚合分区数据量、调整分区键粒度、或使用 repartition() 在写入前手动均衡数据。

2. 数据类型隐式倾斜（Implicit Type Skew）

Spark SQL 在 join 时会对不同数据类型进行隐式转换。当一侧 key 为 String 类型（如 "123"），另一侧为 Long 类型（如 123）时，Spark 会将 String 侧全部转为 Double 再比对。这个转换过程中，所有形如 "abc"、"" 或无法解析为数字的字符串会被统一映射为 NaN，导致大量非数字 key 收敛到同一个分区。

隐式倾斜的诊断难点在于：Spark UI 的 Stage 层面无法直接看到 key 分布，需要通过以下方式定位——在 Stage 的 Task 列表中观察 Shuffle Read Size，如果某个 Task 的读取量是中位数的 5 倍以上（满足 skewedPartitionFactor 默认值 5），但 skewedPartitionThresholdInBytes 未超过 256MB 的阈值，AQE 不会触发 skewJoin。此时需要在 SQL 层面检查 join key 的类型定义，确认是否存在隐式类型转换。

修复方式是统一 join key 的数据类型。例如将 CAST(order_id AS STRING) = product_id 改为 CAST(product_id AS STRING) = order_id，确保两侧类型一致，消除 NaN 收敛。

3. AQE skewJoin 未触发的场景

AQE skewJoin 有两个触发条件同时满足：分区大小超过 skewedPartitionFactor × 中位数，且超过 skewedPartitionThresholdInBytes（默认 256MB）。以下场景会导致 skewJoin 无法触发：

• Broadcast Join：当一侧数据量小于 spark.sql.autoBroadcastJoinThreshold（默认 10MB）时，Spark 选择 broadcast join，不产生 shuffle，AQE skewJoin 不适用。
• Shuffle Hash Join：AQE skewJoin 仅在 Sort Merge Join 中生效。如果 Spark 选择了 Shuffle Hash Join，倾斜分区不会被自动拆分。
• FULL OUTER JOIN：AQE 不支持 FULL OUTER JOIN 的 skew 拆分，这是已知限制。
• 倾斜分区低于阈值：当中位数为 10MB、factor 为 5 时，阈值为 50MB，但绝对阈值 256MB 未达到，即使相对倾斜严重也不会触发。

基于 Spark UI 的系统化诊断流程

第一步：定位慢 Stage

打开 Spark UI 的 SQL 页面，找到执行时间最长的 Stage。点击进入 Stage 详情页，重点关注 Task 耗时分布。如果 Task 列表中存在明显的长尾（部分 Task 耗时是中位数的 3-5 倍），则存在倾斜嫌疑。

第二步：读取 Shuffle Read/Write 指标

在 Stage 详情页的 Summary 栏查看 Shuffle Read 和 Shuffle Write 的统计分布。关注以下几个指标：

• Shuffle Read (min/median/max)：如果 max/median > 5，说明存在显著的 shuffle 数据倾斜。
• Shuffle Write (min/median/max)：如果 Write 侧均匀但 Read 侧不均，问题可能出在上游 Stage 的分区逻辑。
• Task Duration (min/median/max)：直接反映处理时间的不均衡程度。

第三步：判断 AQE 是否介入

在 SQL 页面查看 Physical Plan。如果看到 AdaptiveSparkPlan 节点展开后有 Optimized Skewed Join 的子节点，说明 AQE 已经检测到倾斜并进行了拆分。如果看不到此节点，说明 skewJoin 未被触发，需要排查上述四种未触发场景。

第四步：Stage Skew Ratio 计算

手动计算 Stage 的 skew ratio：max_task_shuffle_read / median_task_shuffle_read。这个比值是判断倾斜严重程度的核心指标。当 skew ratio > 5 且分区绝对大小 > 256MB 时，AQE 应该已经介入；如果未介入，检查 skewJoin.enabled 配置和 join 类型。

修复手段与权衡

Key Salting：在倾斜 key 上拼接随机前缀，分散到多个分区处理后再聚合。适用于 GROUP BY 和 JOIN 场景。代价是需要两次 shuffle（加盐聚合 + 去盐聚合），增加计算开销。

Broadcast Join 强制：对于倾斜的维度表，如果其数据量在广播阈值内，可以强制 broadcast 避免 shuffle。使用 /*+ BROADCAST(table) */ hint。

两阶段聚合：先按 (key, salt) 分组聚合，再去掉 salt 按 key 做最终聚合。将一次大聚合拆成多次小聚合，每阶段都不倾斜。

显式类型转换：在 join 之前对 key 做显式 CAST，避免 Spark 的隐式转换导致 key 收敛。

调整 AQE 参数：如果倾斜分区的绝对大小在 100-256MB 之间，可以适当降低 skewedPartitionThresholdInBytes 到 128MB，使 AQE 能够介入。但阈值过低会导致 AQE 对正常分区过度拆分，增加任务数和调度开销。

非典型倾斜的排查需要同时关注数据层（key 分布、类型）、SQL 层（join 类型、hint）和引擎层（AQE 配置、调度策略），单一层面的调整往往无法根治问题。