一、倾斜现象识别1. Spark UI 页面最核心Stage 页面看 Task 列表同一个 Stage 下绝大部分 task 几秒跑完数据量几十 MB少数几个 task 跑几十分钟shuffle read 达到 GB 级别进度停滞。重点看Shuffle Read列数值差距巨大就是倾斜。看 Task 状态长时间 RUNNING或者反复失败 OOMGC 时间占比很高。看 SQL 标签页如果 SparkSQL看 DAG 图哪个 Shuffle 阶段出现长尾。2. 日志现象Executor 日志大量 Full GC、OutOfMemoryError堆内存溢出。部分 task 重试多次失败其他 task 全部完成。⚠️ Spark没有 MR 那种 Counter 直接看每个 reduce 输入记录数直接看 Spark UI Task 的Shuffle Read大小。二、完整排查步骤Spark UI 定位倾斜 Stage找到 Job展开 DAG找到发生 Shuffle 的 Stage有 Exchange 就是 shuffle。打开该 Stage 的 Task 列表观察 Shuffle Read。部分 task 读取的数据量远超其余 task确认发生数据倾斜。找到慢 Task 对应的 Executor ID点开下载 Executor 日志。在日志中查找热点 key。Spark 不会自动打印 key如果日志看不到可以开启采样或者写代码采样 key 分布找出热点 key。判断倾斜类型join 倾斜 / groupBy 聚合倾斜 / 空值 null 倾斜。区分 MR 和 Spark 排查MRYARN UI 看 CounterReduce input records→ 找慢 Reduce Container → 日志拿 key。SparkSpark UI 看 Stage 的 TaskShuffle Read大小 → 找慢 Task 对应的 Executor → Executor 日志定位热点 key。三、Spark 数据倾斜分类和解决方案类型 1groupBy / groupByKey 聚合倾斜某个 key 的记录数巨大shuffle read 单 task 爆炸。方案 1map 端预聚合combineByKey替代 groupByKey。使用reduceByKey / aggregateByKey自带 map 端本地预聚合减少 shuffle 数据量。❗groupByKey不会做 map 端聚合全部数据走 shuffle极易倾斜。方案 2SparkSQL 开启自适应执行 AQESpark3 重点set spark.sql.adaptive.enabledtrue; set spark.sql.adaptive.skewJoin.enabledtrue; -- 自动处理 join 倾斜 set spark.sql.adaptive.advisory-partition-size256MB;AQE 自动识别倾斜分区把大分区拆分成多个小 task不需要改业务代码。Spark3 强烈优先开启。类型 2Join 数据倾斜最高频两种场景大表 join 小表→ 使用广播 joinBroadcastHashJoin完全规避 shuffle。set spark.sql.autoBroadcastJoinThreshold104857600; -- 默认 100M小表自动广播小表广播到所有 Executor 内存没有 shuffle 阶段。大表 join 大表存在热点 key方案 AAQE 自动倾斜 join 优化Spark3优先。方案 B热点 key 打散加盐salting 加盐。对热点 key 加随机前缀打散到多个 task另一边表把热点 key 复制 N 份同样加上随机前缀join 之后去掉前缀合并结果。方案 C热点 key 单独抽出来单独 join正常 key 正常 join最后 union all 合并。类型 3null / 空 key 倾斜大量 null 的 key 全部落到同一个 shuffle 分区。业务允许直接过滤 null。不能过滤给 null key 拼接随机数打散。类型 4countDistinct 去重倾斜count(distinct col)容易倾斜。优化先用 group by 做去重再 countSpark3 AQE 也能缓解。类型 5分区数设置不合理分区太少每个 task 数据量大容易倾斜分区太多产生大量小 task。spark.sql.shuffle.partitions 默认 200不是越大越好结合数据量调整。四、AQE 核心要点开启 AQE 之后动态调整 shuffle 分区数。自动检测倾斜分区拆分倾斜分区解决 join 倾斜。自动合并小分区。局限AQE 只能处理 shuffle join 倾斜对于 RDD API 的 groupByKeyAQE 不生效需要代码层面优化。五、精简版Spark 排查数据倾斜打开 Spark UI定位存在 Exchangeshuffle的 Stage查看 task 列表的 Shuffle Read 字段如果少数 task 读取数据远大于其他 task说明出现倾斜。找到慢 task 对应的 Executor下载日志定位热点 key。倾斜分聚合倾斜、join 倾斜、null 值倾斜。优先开启 Spark3 AQE 自适应优化大表 join 小表使用广播 joinRDD 层面优先 reduceByKey 代替 groupByKey热点 key 可以加盐打散或者分离热点单独处理。MR 与 Spark 倾斜排查对比表项目MapReduceSpark观察指标Counter 中 Reduce input recordsSpark UI Task 的 Shuffle Read 大小定位慢任务慢 Reduce Task 的 Container慢 Shuffle task 对应的 Executorshuffle 是否排序MR 强制排序Spark 默认不强制排序sort-shuffle 会排序自动倾斜处理hive.groupby.skewindatatrue两轮 MRSpark3 AQE 自适应拆分倾斜分区map 端预聚合CombinerreduceByKey / combineByKey高频问题groupByKey 和 reduceByKey 区别reduceByKey 会在 map 端本地预聚合减少 shuffle 数据groupByKey 全部数据 shuffle 到下游没有本地聚合更容易发生数据倾斜。AQE 能解决所有倾斜吗不能只对 SparkSQL 的 shuffle join 生效原生 RDD API 的 groupByKey 无效需要代码优化。Spark 什么情况下会产生数据倾斜发生 shuffle 算子时join、groupBy、groupByKey、reduceByKey、distinct 等需要重分区的算子。
一、倾斜现象识别1. Spark UI 页面最核心Stage 页面看 Task 列表同一个 Stage 下绝大部分 task 几秒跑完数据量几十 MB少数几个 task 跑几十分钟shuffle read 达到 GB 级别进度停滞。重点看Shuffle Read列数值差距巨大就是倾斜。看 Task 状态长时间 RUNNING或者反复失败 OOMGC 时间占比很高。看 SQL 标签页如果 SparkSQL看 DAG 图哪个 Shuffle 阶段出现长尾。2. 日志现象Executor 日志大量 Full GC、OutOfMemoryError堆内存溢出。部分 task 重试多次失败其他 task 全部完成。⚠️ Spark没有 MR 那种 Counter 直接看每个 reduce 输入记录数直接看 Spark UI Task 的Shuffle Read大小。二、完整排查步骤Spark UI 定位倾斜 Stage找到 Job展开 DAG找到发生 Shuffle 的 Stage有 Exchange 就是 shuffle。打开该 Stage 的 Task 列表观察 Shuffle Read。部分 task 读取的数据量远超其余 task确认发生数据倾斜。找到慢 Task 对应的 Executor ID点开下载 Executor 日志。在日志中查找热点 key。Spark 不会自动打印 key如果日志看不到可以开启采样或者写代码采样 key 分布找出热点 key。判断倾斜类型join 倾斜 / groupBy 聚合倾斜 / 空值 null 倾斜。区分 MR 和 Spark 排查MRYARN UI 看 CounterReduce input records→ 找慢 Reduce Container → 日志拿 key。SparkSpark UI 看 Stage 的 TaskShuffle Read大小 → 找慢 Task 对应的 Executor → Executor 日志定位热点 key。三、Spark 数据倾斜分类和解决方案类型 1groupBy / groupByKey 聚合倾斜某个 key 的记录数巨大shuffle read 单 task 爆炸。方案 1map 端预聚合combineByKey替代 groupByKey。使用reduceByKey / aggregateByKey自带 map 端本地预聚合减少 shuffle 数据量。❗groupByKey不会做 map 端聚合全部数据走 shuffle极易倾斜。方案 2SparkSQL 开启自适应执行 AQESpark3 重点set spark.sql.adaptive.enabledtrue; set spark.sql.adaptive.skewJoin.enabledtrue; -- 自动处理 join 倾斜 set spark.sql.adaptive.advisory-partition-size256MB;AQE 自动识别倾斜分区把大分区拆分成多个小 task不需要改业务代码。Spark3 强烈优先开启。类型 2Join 数据倾斜最高频两种场景大表 join 小表→ 使用广播 joinBroadcastHashJoin完全规避 shuffle。set spark.sql.autoBroadcastJoinThreshold104857600; -- 默认 100M小表自动广播小表广播到所有 Executor 内存没有 shuffle 阶段。大表 join 大表存在热点 key方案 AAQE 自动倾斜 join 优化Spark3优先。方案 B热点 key 打散加盐salting 加盐。对热点 key 加随机前缀打散到多个 task另一边表把热点 key 复制 N 份同样加上随机前缀join 之后去掉前缀合并结果。方案 C热点 key 单独抽出来单独 join正常 key 正常 join最后 union all 合并。类型 3null / 空 key 倾斜大量 null 的 key 全部落到同一个 shuffle 分区。业务允许直接过滤 null。不能过滤给 null key 拼接随机数打散。类型 4countDistinct 去重倾斜count(distinct col)容易倾斜。优化先用 group by 做去重再 countSpark3 AQE 也能缓解。类型 5分区数设置不合理分区太少每个 task 数据量大容易倾斜分区太多产生大量小 task。spark.sql.shuffle.partitions 默认 200不是越大越好结合数据量调整。四、AQE 核心要点开启 AQE 之后动态调整 shuffle 分区数。自动检测倾斜分区拆分倾斜分区解决 join 倾斜。自动合并小分区。局限AQE 只能处理 shuffle join 倾斜对于 RDD API 的 groupByKeyAQE 不生效需要代码层面优化。五、精简版Spark 排查数据倾斜打开 Spark UI定位存在 Exchangeshuffle的 Stage查看 task 列表的 Shuffle Read 字段如果少数 task 读取数据远大于其他 task说明出现倾斜。找到慢 task 对应的 Executor下载日志定位热点 key。倾斜分聚合倾斜、join 倾斜、null 值倾斜。优先开启 Spark3 AQE 自适应优化大表 join 小表使用广播 joinRDD 层面优先 reduceByKey 代替 groupByKey热点 key 可以加盐打散或者分离热点单独处理。MR 与 Spark 倾斜排查对比表项目MapReduceSpark观察指标Counter 中 Reduce input recordsSpark UI Task 的 Shuffle Read 大小定位慢任务慢 Reduce Task 的 Container慢 Shuffle task 对应的 Executorshuffle 是否排序MR 强制排序Spark 默认不强制排序sort-shuffle 会排序自动倾斜处理hive.groupby.skewindatatrue两轮 MRSpark3 AQE 自适应拆分倾斜分区map 端预聚合CombinerreduceByKey / combineByKey高频问题groupByKey 和 reduceByKey 区别reduceByKey 会在 map 端本地预聚合减少 shuffle 数据groupByKey 全部数据 shuffle 到下游没有本地聚合更容易发生数据倾斜。AQE 能解决所有倾斜吗不能只对 SparkSQL 的 shuffle join 生效原生 RDD API 的 groupByKey 无效需要代码优化。Spark 什么情况下会产生数据倾斜发生 shuffle 算子时join、groupBy、groupByKey、reduceByKey、distinct 等需要重分区的算子。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
