这是我一篇关于 Spark 2 生产集群调优的系列文章中的第四节。前几节我们聊了任务提交参数、资源分配、小文件治理这些都是面上的问题而数据倾斜属于点上的问题——它只影响一个或几个 Task却能拖垮整个 Spark 作业是整个分布式计算里最典型的木桶短板。如果你在 Spark Web UI 里看到某个 Stage 下绝大多数 Task 几分钟跑完但有一两个 Task 跑了几个小时还在转圈不用怀疑这就是数据倾斜。这篇文章我会把 Spark2 里数据倾斜的成因、定位方法、常用解法和参数调优边界一次性讲透尤其会说明哪些手段在 Spark2 版本下是真正有效的哪些只是心理安慰。适合正在维护 Spark2 生产集群的开发者、数据工程师阅读。1. 数据倾斜为什么是分布式计算的木桶短板1.1 一眼识别倾斜症状长尾任务和异常耗时数据倾斜最直观的表现是同一个 Stage 里不同 Task 的执行时间出现极端分化。我见过最典型的场景某个 Stage 有 400 个 Task其中 398 个都在 2 分钟内完成但剩下 2 个 Task 跑了 1 小时 40 分钟还没结束。整个作业的运行时间因此等于这 2 个 Task 的运行时间其他资源全都在空转等待。在 Spark UI 的 Stage 详情页里倾斜的识别其实非常机械。你需要看几个指标的组合Task 执行时间的分布。如果中位数时间和最大时间差了一个数量级基本可以判定存在倾斜。例如中位数 30 秒最大 3600 秒这种长尾分布几乎就是倾斜的标志。Shuffle Read / Shuffle Write 的数据量。某个 Task 的 Shuffle Read 达到几十 GB而其他 Task 只有几百 MB说明下游阶段的某几个分区承接了远超正常水平的数据。溢写到磁盘Spill出现的位置。正常情况下如果内存资源给够Shuffle Read 的聚合过程不会溢写。一旦某个 Task 出现大量 Spill往往意味着它处理的数据量超过了内存承受能力而这个某个 Task通常就是倾斜 Task。GC 时间异常。倾斜 Task 的 GC 时间会比正常 Task 高出很多因为大量对象在某一个 Executor 堆内堆积频繁触发 Full GC。这些指标单独看可能各有原因但它们同时出现在同一批 Task 上时指向的结论几乎没有第二种可能数据分配不均。1.2 从案例看影响某电商平台 ETL 从 1 小时拖到 6 小时我在一个模拟项目场景里复现过一次很典型的倾斜问题。某电商平台的大宽表 ETL逻辑很简单把用户维表 Join 到行为日志表上产出用户标签表。日志表按天分区单日数据量约 5 亿条用户维表 8000 万条。开发同学把任务部署到 Spark2.3 集群后初始测试数据量小跑得很快。但全量数据上线后作业的运行时间从预期的 1 小时膨胀到 6 小时而且每天都不稳定。打开 Spark Web UI 发现问题出在 Join 节点那个 Stage。400 个 Shuffle 分区里有 3 个分区上的 Task 处理了总数据量的 40% 以上。这 3 个 Task 的 Shuffle Read 都超过 20GB而正常的 Task 只有 500MB 左右。进一步查数据发现行为日志表里有几个用户的曝光记录量级异常夸张——头部用户的单日行为记录占整日日志的 8%~15%。这种头部用户和用户维表 Join 时同一个用户 ID 的所有行为记录都会被 hash 到同一个分区于是这几个分区承接了海量数据。影响不仅是作业变慢。资源上这几个 Task 长时间占住 Executor其他作业的资源申请被挤压稳定性上倾斜 Task 频繁触发 Executor 内存溢出导致 Executor 被杀后任务重试又加剧了整体耗时的不可控下游调度上这个 ETL 的产出时间从凌晨 2 点推迟到上午 8 点直接影响次日的数据报表和分析任务。这类案例在生产里每天都会出现表现形式各有不同但底层机制是一致的也就是接下来要说的分区原理。2. Spark2中倾斜的根因shuffle分区与Key分布的不匹配2.1 分区器原理一条记录如何被分到某个Task要理解数据倾斜必须回到 Spark 的 shuffle 机制。Spark 在发生宽依赖时Join、GroupByKey、Distinct、ReduceByKey 等上游 Task 产出的数据会根据分区器被重新分配到下游的多个分区中。默认的分区器是 HashPartitioner它决定一条记录去哪个分区的逻辑很简单Hash 分区的核心规则是hash(partitionKey) % numPartitions。某个 Key 计算出哈希值后对分区数量取模结果就是这条记录要去的分区编号。当某个 Key 的出现次数远高于其他 Key 时这个 Key 对应的所有记录都会进入同一个分区。如果这个分区里恰好没有其他 Key 来稀释这个分区就会独自承接海量数据。还有一种情况是 Key 数量很少但每个 Key 的 Value 都很大。比如 GroupBy 的用户 ID 只有 10 万个但每个用户带了 10 万条明细所有用户的明细加总要 100 亿条记录但按 HashPartitioner 分布后10 万个分区不可能都承载均匀的数据量。Key 分布不均和 Value 体量不均本质上都会导致分区数据量失衡。2.2 倾斜怎么在 Join、GroupBy、Distinct 里传导不同的算子倾斜的传导路径不太一样但根子都是同一个。Join 倾斜。Spark 的 SortMergeJoin 在 shuffle 阶段会按 Join Key 对两张表分别做分区和排序。如果 A 表和 B 表的某个 Key 都是热点两张表在这个 Key 上的数据都会被 shuffle 到同一个分区然后这个分区的 Join 就要处理两个超大数据集的匹配。即使 B 表在这个 Key 上的数据量不大只要 A 表的热点 Key 数据量巨大这个任务也是灾难。GroupBy / Aggregate 倾斜。按某个维度做分组聚合时所有相同维度的记录会先被 shuffle 到同一个分区再做聚合。如果维度分布极不均衡比如按省份分组时某个省的数据量是其他省的几十倍这个分区的聚合操作就需要处理超大输入。Distinct 去重倾斜。Distinct 在底层也是先做分区再做去重热点 Key 同样会导致某个分区承接海量重复数据。笛卡尔积 / 广播场景下的倾斜。有时不是分区器的问题而是某一张表的某个 Key 数本身巨大广播后与另一张表所有分区进行连接形成局部热点。这类问题往往更隐蔽因为从 Web UI 上看每个 Task 都在工作但某个 Executor 的资源消耗明显超出其他 Executor。2.3 为什么 Spark2 时代这个问题比 Spark1 更容易被放大对比 Spark1 和 Spark2数据倾斜的底层机制没变但 Spark2 在某些场景下会把倾斜问题暴露得更明显。一个原因是 Spark2 默认采用 SortMergeJoin而不是 Spark1 时代常用的 HashJoin。SortMergeJoin 需要先对数据进行全量排序和分区这意味着参与 Join 的每个 Key 都必须经过完整的 shuffle 和排序流程。在 Spark1 中如果一张表足够小HashJoin 可以通过广播直接跳过 shuffle倾斜的触发面反而小一些。Spark2 虽然可以通过参数把自动广播阈值调大但一旦超过阈值SortMergeJoin 对倾斜的容忍度更低——数据量大的分区不仅要做匹配还要做大量排序和比较操作。另一个原因是 Spark2 在内存管理上引入了 Unified Memory堆内的执行内存和存储内存可以相互借用。这个设计整体提升了内存利用率但倾斜 Task 大量占用执行内存时可能导致同 Executor 上的其他 Task 内存被挤占出现相互拖累的现象。也就是说Spark2 里一个倾斜 Task 的影响范围可能不止自己而是整个 Executor 上的并发任务。还有一个不容忽视的原因是很多生产集群从 Spark1 迁移到 Spark2 时只改版本不改作业逻辑。旧作业里一些依靠 Spark1 行为特性恰好不倾斜的代码在 Spark2 的优化器规则和算子实现变化后shuffle 分区行为发生了变化倾斜问题就爆发了。3. 定位数据倾斜的完整链路从监控指标到元凶Key3.1 Spark Web UI 里最值得盯的三个数字定位数据倾斜第一步永远是看 Spark Web UI不要凭感觉猜。在 Stage 详情页我建议你聚焦三个数字抓住核心信息。第一个是每个 Task 的 Shuffle Read Size / Records。这个指标最直接。把所有 Task 的 Shuffle Read 拉出来排序如果最大值和最小值的差距超过 10 倍就已经有倾斜嫌疑。更精确的判断方式是看分位数如果 90% 的 Task 读取量都在 1GB 以下但有个别 Task 超过 10GB不用犹豫这就是倾斜。第二个是每个 Task 的 Duration 与 GC Time。Duration 的长尾分布是倾斜的外在表现而 GC Time 高往往说明这个 Task 处理的数据量大到内存装不下。如果一个 Task 的 GC Time 占总耗时的 30% 以上它的输入数据大概率是超量级的。第三个是整个 Job 的**Skipped StagesActive Stages 和 Failed Tasks 情况**。倾斜不一定导致失败但如果有 Task 因为内存溢出被反复重试Stage 的失败次数会明显异常。这种情况下倾斜和 Executor 资源不足的问题同时存在需要先解决倾斜再评估资源是否真的不够。除了这三个核心指标Stage 详情里还有一个表格列了每个 Task 的 ID、Executor 所在节点、输入/输出大小、Shuffle 读写、Shuffle Spill 等。把这些列全部展开按 Shuffle Read 排序一眼就能看到异常的那几个 Task。3.2 用 Spark Shell 和 DataFrame API 给倾斜程度打分Web UI 能告诉你哪个 Task 有问题但不能直接告诉你哪个 Key 是元凶。这一步需要通过代码来定位。最实用的做法是不直接跑完整作业而是先把涉及的数据读出来对疑似倾斜的字段做一次值频率统计。这里我给出一个用 Spark SQL 就能实现的打分方式-- 统计疑似热点字段的分布 SELECT user_id, COUNT(*) AS cnt FROM behavior_log GROUP BY user_id ORDER BY cnt DESC LIMIT 20如果前几个 Key 的数量级远远超过均值比如 top1 占了全表数据的 15%那么它就是这个 Stage 的倾斜元凶。但生产环境中的数据量很大直接对全量数据跑GROUP BY COUNT本身也要花时间。更高效的方式是分层采样先用df.sample(false, 0.01)采样 1% 的数据做频率统计验证热点 Key 是否在样本中同样突出。如果样本里就能看到明显的头部效应基本可以确认问题。如果用的是 Scala 或 PySpark 的 DataFrame API也可以用df.groupBy(user_id).count().sort($count.desc).show(20)直接看结果。排查完 Key 之后还需要确认这个 Key 在整条执行链路中的影响范围。建议打开 SQL 执行计划看目标 Stage 的上游是什么算子df.explain(true)通过explain(true)可以看到每个算子的输出分区数、输出表达式。如果某个算子输出的分区数等于你设置的spark.sql.shuffle.partitions且该算子之前有 Join 或 GroupBy那倾斜 Stage 就在这个位置。3.3 把问题锁定到具体算子和具体Key的排查顺序排查数据倾斜最容易犯的错误是一上来就调整参数比如把spark.sql.shuffle.partitions从 200 调到 2000。这个操作在部分场景下有效但在热点 Key 极度集中的场景下基本无效后面我会详细解释原因。我推荐的排查顺序是固定的按这个顺序走通常能在 10 分钟内定位问题打开 Job 列表找到运行时间最长的那个 Job进入它的 Stage 列表。在 Stage 列表里找到 Summary Metrics 中 Total Time Across All Tasks 和 Max Task Time 差距最大的 Stage点进去。按 Shuffle Read Size 对 Task 排序确认是不是个别 Task 的数据量惊人。对应打开explain(true)的执行计划确认这个 Stage 是哪个算子产生的。通常反向从叶子节点往上找第一个发生 Shuffle 的节点就是嫌疑位置。把这张表的关键字段做频率统计找到具体热点 Key。回到 Stage 详情页确认热点 Key 的哈希结果对应哪个分区号以及这个分区在哪台节点上运行。最后一步看你是否需要对热点 Key 做定向优化。如果热点 Key 只出现在某台节点上并且这台节点的资源本身较弱那么问题可能是资源不均匀叠加了倾斜而不是纯倾斜问题。4. 生产环境常用的五类倾斜解法与边界4.1 广播Join治小表倾斜的特效药别用在大表上如果倾斜发生在 Join 阶段而且其中一张表比较小通常小于 100MB或者经过过滤后小于 driver 内存可承受的范围最简单、最有效的手段是广播 Join。Spark2 里可以通过两种方式触发广播// 方式一显式 Hint df1.join(broadcast(df2), Seq(key), inner) // 方式二全局参数配置 // spark.sql.autoBroadcastJoinThreshold 104857600 // 默认 10MB可调大广播 Join 的原理是把小表的数据收集到 Driver再分发到每个 Executor 的内存中在 map 端直接完成匹配完全跳过 shuffle因此没有分区数据不均衡的问题。但这里有三个容易踩的坑第一个坑是阈值调整要谨慎。spark.sql.autoBroadcastJoinThreshold默认是 10MB如果你的维表是 50MB可以改到 100MB但不要随手改成 1GB。过大的广播阈值会导致 Driver 收集大量数据Driver 内存直接被打爆同时每批 Task 都要把广播变量从 Driver 拉取到 Executor网络和序列化开销也很大。第二个坑是广播后仍然可能出现单点瓶颈。如果小表广播后另一张大表里某个 Key 的数据集中在一个节点上BroadcastHashJoin 阶段虽然不 shuffle但匹配后的输出仍然可能落到同一个下游分区。第三个坑是广播的表必须是真正很小的维度表不能因为这张表看起来不大就广播。我在一个生产案例里见过同事把一张 3GB 的配置表设置成广播结果每个 Executor 都要缓存 3GB 数据直接引发大规模内存不足。4.2 热点Key加盐随机前缀与二次Join的配合当倾斜发生在 Join 且两边都是大表时广播 Join 不可用这是最常见的生产困境。标准解法是加盐Salting思路非常朴素让原本会被哈希到同一个分区的热点 Key通过加随机前缀的方式分散到多个分区。核心思路是三步识别出热点 Key比如某个 user_id9527数据量占 10%。把这个热点 Key 在大表侧随机拆成多个带前缀的 Key。比如把9527拆成9527_0、9527_1、9527_2、9527_3。在小表侧对同样的热点 Key 也用相同的前缀集合进行膨胀复制保证两边能匹配上。具体代码可以这么写// 1. 先找出热点 Key val hotKeys df1.groupBy(user_id).count().filter($count 1000000) .select(user_id) .as[String].collect().toSet // 2. 大表侧热点Key加随机前缀非热点Key保持原样 val df1Salted df1.withColumn(salted_key, when($user_id.isin(hotKeys.toSeq: _*), concat($user_id, lit(_), (rand() * 4).cast(int))) .otherwise($user_id)) // 3. 小表侧热点Key膨胀复制需要与加盐侧的前缀范围一致 val df2Salted df2.withColumn(salted_key, when($user_id.isin(hotKeys.toSeq: _*), explode(array(lit(0), lit(1), lit(2), lit(3)) .zipWithIndex.map(x concat($user_id, lit(_), lit(x._2))))) .otherwise($user_id)) // 4. 用加盐后的 Key 做 Join val result df1Salted.join(df2Salted, Seq(salted_key), inner)注意这个方案有个前提小表侧的膨胀复制数必须和大表侧的随机前缀扇出数一致。上例里面大表侧 rand()*4 生成了 0~3 共 4 个前缀小表侧也要膨胀复制出 4 份才能保证9527_0能匹配到9527_0。加盐的代价是数据量膨胀。热点 Key 拆成 N 份后小表侧的记录数会变为原来的 N 倍Shuffle 和最后再聚合的开销也相应增加。所以加盐的扇出数不是越大越好建议从 4 到 8 开始观察任务耗时变化再逐步调整。另外加盐方案主要解决 Join 本身的倾斜。如果倾斜发生在 GroupBy 聚合阶段一般采用下面说的两阶段聚合。4.3 两阶段聚合先局部汇总再全局汇总的取舍两阶段聚合的方案主要应对GroupBy 聚合倾斜。它的核心思路是数据先带着随机前缀做第一次局部聚合去掉前缀后做第二次全局聚合。// 第一步加盐分组局部聚合 val partial df.withColumn(salt, (rand() * 10).cast(int)) .groupBy(user_id, salt) .agg(sum(amount).as(partial_amount)) // 第二步去盐分组全局聚合 val final partial.groupBy(user_id) .agg(sum(partial_amount).as(total_amount))这个方案的收益很明显原本全部集中在同一个分区的热点 Key现在被拆到 10 个分区分别聚合每个分区的负载变为原来的 10%。代价是执行了两遍聚合额外的 Shuffle 开销大约在 10%~30% 之间但如果热点 Key 导致单个 Task 跑 2 小时这点开销完全值得。使用两阶段聚合要注意一个问题它在某些聚合函数下不适用。比如count(distinct ...)或者approx_count_distinct在局部聚合阶段做去重计数再在全局阶段合并结果不一定准确需要做专门的去重设计。另外如果聚合函数本身是有序敏感的比如 collecting list 时需要保序两阶段聚合会打乱顺序。对于生产环境我通常建议如果热点 Key 只占少数几个值不要对所有 Key 加盐只对热点 Key 加盐。方法是先识别热点 Key再在when条件里对热点 Key 加盐其他 Key 保持不变这样能减少不必要的局部聚合开销。4.4 空值与默认值的清洗多数倾斜的源头其实不是业务热点一个很容易被忽略但频繁出现的情况是大量记录的关键字段是空值或默认值。比如行为日志里的 user_id 在下游数仓里是空 String或者为-9999这类业务规定的默认值。这种情况下空值/默认值占据的数据量可能是所有热点业务 Key 的总和它们全部会被哈希到同一个分区制造出一个巨大的倾斜分区。这种倾斜的解法说起来最简单但生产里很多人会漏掉过滤不需要的空值记录。如果下游统计根本用不到未登录用户的行为直接在 Join 前把空 user_id 过滤掉。这一步能把数据量缩小 20% 甚至更多。如果空值记录必须保留可以给空值赋一个随机值或随机前缀这样它们会被均匀分散到所有分区。// 将空值替换为带随机后缀的Key避免集中在同一分区 df.withColumn(join_key, when($user_id.isNull || $user_id.equalTo(-9999), concat(lit(unknown_), lit((rand() * 100).cast(int)))) .otherwise($user_id))这里有一个生产上的教训某次我排查一个 Join 倾斜Web UI 显示同一个分区承接了 35% 的数据以为是头部用户热点。结果跑频率统计后发现排名第一的 Key 实际上是空字符串占了全部数据的 28%。业务热点 Key 只排在第二位。如果没做字段频率分析就盲目加盐不仅解决不了问题还会因为错误膨胀数据量让作业更慢。所以任何倾斜排查都要从字段频率统计开始而不是从业务直觉开始。4.5 全局排序/去重场景中的Reducer端重构倾斜不一定只出现在 Join 和 GroupBy 上。全局排序、全局 TopN、distinct也经常出现某个分区数据量异常。原因是这些操作在 reducer 端需要把全局数据集中到一个或少数几个分区上处理。以全局排序为例Spark 的sortBy最终会把数据分成多个排序分区但如果排序列的分布不均——比如大部分记录的排序 Key 都落在某个区间——这个区间对应的分区就会聚集大部分数据。这种情况通常需要配合range partitioning而不是默认的 hash 分区// 使用 repartitionAndSortWithinPartitions 自定义区间分区 df.repartitionByRange(col(score).desc)repartitionByRange会先采样预估一个均匀的分桶边界然后按范围分区这样可以让数据在多个分区之间更均衡。而distinct倾斜的解法本质上也是两阶段去重先加盐实现局部去重再去掉盐做最终去重。相比全局 distinct这个方案的输出结果完全一致但 Shuffle 负载会被分散。这里要特别提醒一个误区很多人遇到 Task 倾斜第一反应是增加spark.sql.shuffle.partitions以为把分区数调大就能让热点 Key 的数据更加分散。但热点 Key 的哈希结果在一个固定分区器下是确定的分区数从 200 调到 2000热点 Key 只会从压垮 1 个分区变成还是压垮 1 个分区并没有把数据量平分到 10 个分区。真正让热点分散的手段只有加盐、范围分区或者改变分区器逻辑。关于这一点下一节我会展开讲。5. Spark2参数调优的实效边界配置不是越多越好5.1 spark.sql.shuffle.partitions 的真实作用范围spark.sql.shuffle.partitions是 Spark SQL 里最著名的参数之一。它的作用是设置 SQL 作业在发生 Shuffle 时产出的分区数量默认是 200。很多排障手册遇到倾斜就建议调大它但实际上要看倾斜的成因。分两种情况来理解如果倾斜是因为分区数太少导致的比如集群有 500 个 Executor 核但 Shuffle 分区只有 200意味着 500 个核里只有 200 个在跑 Shuffle Task有 300 个核空转。此时调大分区数到 500 或者 1000能让任务在更多 Executor 上并行执行倾斜压力会被打散。这种情况调大分区是有效的。如果倾斜是个别 Key 数据量巨大导致的分区别管是 200 还是 2000热点 Key 的哈希结果仍然只会落在一个分区上。调大分区数只会让这个热点分区的数据量占比保持不变问题不会消失。更糟的是分区数调大后所有非热点分区都只处理很少的数据并发任务数变多调度开销和网络连接数反而上升作业总时间可能不降反升。所以调大spark.sql.shuffle.partitions的正确场景是并行度不足而不是数据倾斜。判断标准很简单看你 Web UI 上这个 Stage 的 Active Task 数和当前分区的对应关系。如果每个分区上只跑了一个 Task而且其他 Executor 核都在空转优先调大分区如果某个 Task 的输入大小是其他 Task 的几十倍调大分区只会把瓶颈转移到下一个 Stage。5.2 内存参数与溢写警惕倾斜场景下的伪内存不足倾斜 Task 处理的数据量远超常规时往往会伴随一个现象ExecutorLostFailure或者java.lang.OutOfMemoryError。很多同学会以为是 Executor 内存不够于是盲目调大spark.executor.memory。但这不是问题的根源。在 Spark2 的 Unified Memory 模型下真正的问题是倾斜 Task 在一个 Executor 内堆积了远超预期的数据占满了执行内存后还可能占用存储内存进而引发整个 Executor 的内存竞争。这时候加内存只是在缓解症状没有解决热点数据依然压在一个分区上的问题。实际操作中我更倾向于关注spark.shuffle.spill.compress和spark.shuffle.io.maxRetries这些与溢写行为相关的参数。倾斜 Task 的溢写是它的正常保护机制如果溢写发生在 shuffle read 阶段说明这个 Task 的输入就已经超过了内存。有个经验值可以参考如果倾斜 Task 的 Shuffle Read 数据量超过 Executor 执行内存的 2 倍这个 Task 大概率会被拖垮。此时要判断的是到底是先调整 Key 的分布策略而不是继续加内存。5.3 哪些调优是心理安慰一个配置组合的实测对比我见过有人在一次倾斜问题排查中同时调整了spark.executor.memory、spark.executor.cores、spark.sql.shuffle.partitions、spark.default.parallelism、spark.memory.fraction结果作业还是慢但配置已经改得妈都不认识。这里我想给你一个我非常确定的经验参数调优永远解决不了 Key 本身歪斜的问题。数据倾斜是数据分布问题不是资源问题。与其堆参数不如先解决热点数据的分布。做一个简单的验证你就明白了同一个带倾斜的作业在一组资源完全相同的集群上仅调整spark.sql.shuffle.partitions从 200 到 400你大概率会看到只有一个热点 Task 的耗时从 60 分钟变成 58 分钟作业总时长几乎不变。而把热点 Key 加盐拆成 4 份之后同一个热点的 Task 可能只需要 20 分钟就能跑完。所以我的实际建议是遇到倾斜问题参数的调整放在最后一步并且只关注三个spark.sql.shuffle.partitions的值是否与集群并行度匹配。spark.sql.autoBroadcastJoinThreshold是否需要针对小表场景放大。spark.sql.join.preferSortMergeJoin在 Spark2.4 版本里可以针对小表场景临时关闭让优化器选择广播 HashJoin。其他参数比如spark.executor.memory、spark.memory.storageFraction等等不要为了倾斜问题而轻易调整因为它们往往属于资源规划范畴而不是数据分布范畴。6. 工程侧的长效治理把数据倾斜关进笼子6.1 Key设计规范与数据模型评审数据倾斜是一次性的问题但它的产生根源往往是长期的。生产集群每天都在产生新任务如果不在工程侧建立规范每次遇到倾斜都要临时救火。我在这里想分享一个我所在团队抽象掉具体组织信息在实践中形成的三条规则第一关键业务字段要先做值域分析再进入模型评审。对即将作为 Join Key、GroupBy Key 的字段上线前先统计字段的值频率分布看是否有 top 值占比超过 1% 的情况。如果超过这个字段进入下游链路时就必须设计好倾斜预案。比如用户表与行为表 Join提前识别头部用户的比例评估是否需要分桶策略。第二空值和默认值必须在数仓开发规范里明确处理方式。约定主键字段留空时的填充规则建议保留原始值但不参与作为业务 Join Key。可以在基表层面建好清洗逻辑而不是每次在作业里临时处理。第三克制万能大宽表的冲动。宽表设计通常会引入大量 Join而且每个 Join 的 Key 都可能是热点。每多一次 Join就多了一次倾斜的可能。能通过维度退化处理的就不要引入额外的维度表关联。6.2 常态化巡检把倾斜排查变成例行体检既然数据倾斜的定位链路已经比较标准了我建议团队把排查过程沉淀成一个固定的巡检脚本每天任务跑完后自动输出疑似倾斜的 Stage 列表。我们的做法是从 Spark History Server 上拉取每个 Job 的执行信息计算每个 Stage 的 Task 最大耗时与中位耗时的比值超过阈值就自动告警。同时把 Shuffle Read 大小的极差也纳入告警指标。这个巡检脚本不需要很复杂本质上就是把人工看 Web UI 的逻辑固化下来。另外在数据模型发布流程上建议加一步倾斜影响评估对这个任务涉及的 Join 和 GroupBy 字段做一次抽样频率统计输出 Top Key 占比。如果某个 Key 占比超过 5%就要求开发同学说明是否有加盐、广播等应对预案。这个前置检查能减少大部分发布上线后的倾斜事故。6.3 个人经验与最后提醒数据倾斜这个事情处理方式并不复杂但很容易被带偏。我见过太多同事一上来就调参数或者直接换分区器最后把问题搞得更严重。我的经验是拿到任何一个倾斜现象先花 3 分钟理清楚数据集中在哪个 Key 上这个 Key 经过哪个算子传到了哪个分区然后再选择对策。从技术方案角度看广播 Join 适用小表加盐适用于单 Key 热点两阶段聚合适用于聚合场景过滤空值适用于脏数据场景四管齐下基本能覆盖生产环境中九成以上的倾斜问题。剩下的少量场景需要结合数据业务特征具体设计。最后再分享一个小技巧修复倾斜问题后不要只改作业代码建议把排查过程中得到的数据分布结论和解决方式记录在任务注释里。半年后再看同一个任务时你会感谢当时的自己。这个习惯在很多团队里都能大幅降低同类问题的返工率和排查时间。 
长河君 · 2026/10/13 10:55:42 · 阅读 1.2k Spark2生产集群数据倾斜实战：从长尾任务定位到调优方案
这是我一篇关于 Spark 2 生产集群调优的系列文章中的第四节。前几节我们聊了任务提交参数、资源分配、小文件治理这些都是面上的问题而数据倾斜属于点上的问题——它只影响一个或几个 Task却能拖垮整个 Spark 作业是整个分布式计算里最典型的木桶短板。如果你在 Spark Web UI 里看到某个 Stage 下绝大多数 Task 几分钟跑完但有一两个 Task 跑了几个小时还在转圈不用怀疑这就是数据倾斜。这篇文章我会把 Spark2 里数据倾斜的成因、定位方法、常用解法和参数调优边界一次性讲透尤其会说明哪些手段在 Spark2 版本下是真正有效的哪些只是心理安慰。适合正在维护 Spark2 生产集群的开发者、数据工程师阅读。1. 数据倾斜为什么是分布式计算的木桶短板1.1 一眼识别倾斜症状长尾任务和异常耗时数据倾斜最直观的表现是同一个 Stage 里不同 Task 的执行时间出现极端分化。我见过最典型的场景某个 Stage 有 400 个 Task其中 398 个都在 2 分钟内完成但剩下 2 个 Task 跑了 1 小时 40 分钟还没结束。整个作业的运行时间因此等于这 2 个 Task 的运行时间其他资源全都在空转等待。在 Spark UI 的 Stage 详情页里倾斜的识别其实非常机械。你需要看几个指标的组合Task 执行时间的分布。如果中位数时间和最大时间差了一个数量级基本可以判定存在倾斜。例如中位数 30 秒最大 3600 秒这种长尾分布几乎就是倾斜的标志。Shuffle Read / Shuffle Write 的数据量。某个 Task 的 Shuffle Read 达到几十 GB而其他 Task 只有几百 MB说明下游阶段的某几个分区承接了远超正常水平的数据。溢写到磁盘Spill出现的位置。正常情况下如果内存资源给够Shuffle Read 的聚合过程不会溢写。一旦某个 Task 出现大量 Spill往往意味着它处理的数据量超过了内存承受能力而这个某个 Task通常就是倾斜 Task。GC 时间异常。倾斜 Task 的 GC 时间会比正常 Task 高出很多因为大量对象在某一个 Executor 堆内堆积频繁触发 Full GC。这些指标单独看可能各有原因但它们同时出现在同一批 Task 上时指向的结论几乎没有第二种可能数据分配不均。1.2 从案例看影响某电商平台 ETL 从 1 小时拖到 6 小时我在一个模拟项目场景里复现过一次很典型的倾斜问题。某电商平台的大宽表 ETL逻辑很简单把用户维表 Join 到行为日志表上产出用户标签表。日志表按天分区单日数据量约 5 亿条用户维表 8000 万条。开发同学把任务部署到 Spark2.3 集群后初始测试数据量小跑得很快。但全量数据上线后作业的运行时间从预期的 1 小时膨胀到 6 小时而且每天都不稳定。打开 Spark Web UI 发现问题出在 Join 节点那个 Stage。400 个 Shuffle 分区里有 3 个分区上的 Task 处理了总数据量的 40% 以上。这 3 个 Task 的 Shuffle Read 都超过 20GB而正常的 Task 只有 500MB 左右。进一步查数据发现行为日志表里有几个用户的曝光记录量级异常夸张——头部用户的单日行为记录占整日日志的 8%~15%。这种头部用户和用户维表 Join 时同一个用户 ID 的所有行为记录都会被 hash 到同一个分区于是这几个分区承接了海量数据。影响不仅是作业变慢。资源上这几个 Task 长时间占住 Executor其他作业的资源申请被挤压稳定性上倾斜 Task 频繁触发 Executor 内存溢出导致 Executor 被杀后任务重试又加剧了整体耗时的不可控下游调度上这个 ETL 的产出时间从凌晨 2 点推迟到上午 8 点直接影响次日的数据报表和分析任务。这类案例在生产里每天都会出现表现形式各有不同但底层机制是一致的也就是接下来要说的分区原理。2. Spark2中倾斜的根因shuffle分区与Key分布的不匹配2.1 分区器原理一条记录如何被分到某个Task要理解数据倾斜必须回到 Spark 的 shuffle 机制。Spark 在发生宽依赖时Join、GroupByKey、Distinct、ReduceByKey 等上游 Task 产出的数据会根据分区器被重新分配到下游的多个分区中。默认的分区器是 HashPartitioner它决定一条记录去哪个分区的逻辑很简单Hash 分区的核心规则是hash(partitionKey) % numPartitions。某个 Key 计算出哈希值后对分区数量取模结果就是这条记录要去的分区编号。当某个 Key 的出现次数远高于其他 Key 时这个 Key 对应的所有记录都会进入同一个分区。如果这个分区里恰好没有其他 Key 来稀释这个分区就会独自承接海量数据。还有一种情况是 Key 数量很少但每个 Key 的 Value 都很大。比如 GroupBy 的用户 ID 只有 10 万个但每个用户带了 10 万条明细所有用户的明细加总要 100 亿条记录但按 HashPartitioner 分布后10 万个分区不可能都承载均匀的数据量。Key 分布不均和 Value 体量不均本质上都会导致分区数据量失衡。2.2 倾斜怎么在 Join、GroupBy、Distinct 里传导不同的算子倾斜的传导路径不太一样但根子都是同一个。Join 倾斜。Spark 的 SortMergeJoin 在 shuffle 阶段会按 Join Key 对两张表分别做分区和排序。如果 A 表和 B 表的某个 Key 都是热点两张表在这个 Key 上的数据都会被 shuffle 到同一个分区然后这个分区的 Join 就要处理两个超大数据集的匹配。即使 B 表在这个 Key 上的数据量不大只要 A 表的热点 Key 数据量巨大这个任务也是灾难。GroupBy / Aggregate 倾斜。按某个维度做分组聚合时所有相同维度的记录会先被 shuffle 到同一个分区再做聚合。如果维度分布极不均衡比如按省份分组时某个省的数据量是其他省的几十倍这个分区的聚合操作就需要处理超大输入。Distinct 去重倾斜。Distinct 在底层也是先做分区再做去重热点 Key 同样会导致某个分区承接海量重复数据。笛卡尔积 / 广播场景下的倾斜。有时不是分区器的问题而是某一张表的某个 Key 数本身巨大广播后与另一张表所有分区进行连接形成局部热点。这类问题往往更隐蔽因为从 Web UI 上看每个 Task 都在工作但某个 Executor 的资源消耗明显超出其他 Executor。2.3 为什么 Spark2 时代这个问题比 Spark1 更容易被放大对比 Spark1 和 Spark2数据倾斜的底层机制没变但 Spark2 在某些场景下会把倾斜问题暴露得更明显。一个原因是 Spark2 默认采用 SortMergeJoin而不是 Spark1 时代常用的 HashJoin。SortMergeJoin 需要先对数据进行全量排序和分区这意味着参与 Join 的每个 Key 都必须经过完整的 shuffle 和排序流程。在 Spark1 中如果一张表足够小HashJoin 可以通过广播直接跳过 shuffle倾斜的触发面反而小一些。Spark2 虽然可以通过参数把自动广播阈值调大但一旦超过阈值SortMergeJoin 对倾斜的容忍度更低——数据量大的分区不仅要做匹配还要做大量排序和比较操作。另一个原因是 Spark2 在内存管理上引入了 Unified Memory堆内的执行内存和存储内存可以相互借用。这个设计整体提升了内存利用率但倾斜 Task 大量占用执行内存时可能导致同 Executor 上的其他 Task 内存被挤占出现相互拖累的现象。也就是说Spark2 里一个倾斜 Task 的影响范围可能不止自己而是整个 Executor 上的并发任务。还有一个不容忽视的原因是很多生产集群从 Spark1 迁移到 Spark2 时只改版本不改作业逻辑。旧作业里一些依靠 Spark1 行为特性恰好不倾斜的代码在 Spark2 的优化器规则和算子实现变化后shuffle 分区行为发生了变化倾斜问题就爆发了。3. 定位数据倾斜的完整链路从监控指标到元凶Key3.1 Spark Web UI 里最值得盯的三个数字定位数据倾斜第一步永远是看 Spark Web UI不要凭感觉猜。在 Stage 详情页我建议你聚焦三个数字抓住核心信息。第一个是每个 Task 的 Shuffle Read Size / Records。这个指标最直接。把所有 Task 的 Shuffle Read 拉出来排序如果最大值和最小值的差距超过 10 倍就已经有倾斜嫌疑。更精确的判断方式是看分位数如果 90% 的 Task 读取量都在 1GB 以下但有个别 Task 超过 10GB不用犹豫这就是倾斜。第二个是每个 Task 的 Duration 与 GC Time。Duration 的长尾分布是倾斜的外在表现而 GC Time 高往往说明这个 Task 处理的数据量大到内存装不下。如果一个 Task 的 GC Time 占总耗时的 30% 以上它的输入数据大概率是超量级的。第三个是整个 Job 的**Skipped StagesActive Stages 和 Failed Tasks 情况**。倾斜不一定导致失败但如果有 Task 因为内存溢出被反复重试Stage 的失败次数会明显异常。这种情况下倾斜和 Executor 资源不足的问题同时存在需要先解决倾斜再评估资源是否真的不够。除了这三个核心指标Stage 详情里还有一个表格列了每个 Task 的 ID、Executor 所在节点、输入/输出大小、Shuffle 读写、Shuffle Spill 等。把这些列全部展开按 Shuffle Read 排序一眼就能看到异常的那几个 Task。3.2 用 Spark Shell 和 DataFrame API 给倾斜程度打分Web UI 能告诉你哪个 Task 有问题但不能直接告诉你哪个 Key 是元凶。这一步需要通过代码来定位。最实用的做法是不直接跑完整作业而是先把涉及的数据读出来对疑似倾斜的字段做一次值频率统计。这里我给出一个用 Spark SQL 就能实现的打分方式-- 统计疑似热点字段的分布 SELECT user_id, COUNT(*) AS cnt FROM behavior_log GROUP BY user_id ORDER BY cnt DESC LIMIT 20如果前几个 Key 的数量级远远超过均值比如 top1 占了全表数据的 15%那么它就是这个 Stage 的倾斜元凶。但生产环境中的数据量很大直接对全量数据跑GROUP BY COUNT本身也要花时间。更高效的方式是分层采样先用df.sample(false, 0.01)采样 1% 的数据做频率统计验证热点 Key 是否在样本中同样突出。如果样本里就能看到明显的头部效应基本可以确认问题。如果用的是 Scala 或 PySpark 的 DataFrame API也可以用df.groupBy(user_id).count().sort($count.desc).show(20)直接看结果。排查完 Key 之后还需要确认这个 Key 在整条执行链路中的影响范围。建议打开 SQL 执行计划看目标 Stage 的上游是什么算子df.explain(true)通过explain(true)可以看到每个算子的输出分区数、输出表达式。如果某个算子输出的分区数等于你设置的spark.sql.shuffle.partitions且该算子之前有 Join 或 GroupBy那倾斜 Stage 就在这个位置。3.3 把问题锁定到具体算子和具体Key的排查顺序排查数据倾斜最容易犯的错误是一上来就调整参数比如把spark.sql.shuffle.partitions从 200 调到 2000。这个操作在部分场景下有效但在热点 Key 极度集中的场景下基本无效后面我会详细解释原因。我推荐的排查顺序是固定的按这个顺序走通常能在 10 分钟内定位问题打开 Job 列表找到运行时间最长的那个 Job进入它的 Stage 列表。在 Stage 列表里找到 Summary Metrics 中 Total Time Across All Tasks 和 Max Task Time 差距最大的 Stage点进去。按 Shuffle Read Size 对 Task 排序确认是不是个别 Task 的数据量惊人。对应打开explain(true)的执行计划确认这个 Stage 是哪个算子产生的。通常反向从叶子节点往上找第一个发生 Shuffle 的节点就是嫌疑位置。把这张表的关键字段做频率统计找到具体热点 Key。回到 Stage 详情页确认热点 Key 的哈希结果对应哪个分区号以及这个分区在哪台节点上运行。最后一步看你是否需要对热点 Key 做定向优化。如果热点 Key 只出现在某台节点上并且这台节点的资源本身较弱那么问题可能是资源不均匀叠加了倾斜而不是纯倾斜问题。4. 生产环境常用的五类倾斜解法与边界4.1 广播Join治小表倾斜的特效药别用在大表上如果倾斜发生在 Join 阶段而且其中一张表比较小通常小于 100MB或者经过过滤后小于 driver 内存可承受的范围最简单、最有效的手段是广播 Join。Spark2 里可以通过两种方式触发广播// 方式一显式 Hint df1.join(broadcast(df2), Seq(key), inner) // 方式二全局参数配置 // spark.sql.autoBroadcastJoinThreshold 104857600 // 默认 10MB可调大广播 Join 的原理是把小表的数据收集到 Driver再分发到每个 Executor 的内存中在 map 端直接完成匹配完全跳过 shuffle因此没有分区数据不均衡的问题。但这里有三个容易踩的坑第一个坑是阈值调整要谨慎。spark.sql.autoBroadcastJoinThreshold默认是 10MB如果你的维表是 50MB可以改到 100MB但不要随手改成 1GB。过大的广播阈值会导致 Driver 收集大量数据Driver 内存直接被打爆同时每批 Task 都要把广播变量从 Driver 拉取到 Executor网络和序列化开销也很大。第二个坑是广播后仍然可能出现单点瓶颈。如果小表广播后另一张大表里某个 Key 的数据集中在一个节点上BroadcastHashJoin 阶段虽然不 shuffle但匹配后的输出仍然可能落到同一个下游分区。第三个坑是广播的表必须是真正很小的维度表不能因为这张表看起来不大就广播。我在一个生产案例里见过同事把一张 3GB 的配置表设置成广播结果每个 Executor 都要缓存 3GB 数据直接引发大规模内存不足。4.2 热点Key加盐随机前缀与二次Join的配合当倾斜发生在 Join 且两边都是大表时广播 Join 不可用这是最常见的生产困境。标准解法是加盐Salting思路非常朴素让原本会被哈希到同一个分区的热点 Key通过加随机前缀的方式分散到多个分区。核心思路是三步识别出热点 Key比如某个 user_id9527数据量占 10%。把这个热点 Key 在大表侧随机拆成多个带前缀的 Key。比如把9527拆成9527_0、9527_1、9527_2、9527_3。在小表侧对同样的热点 Key 也用相同的前缀集合进行膨胀复制保证两边能匹配上。具体代码可以这么写// 1. 先找出热点 Key val hotKeys df1.groupBy(user_id).count().filter($count 1000000) .select(user_id) .as[String].collect().toSet // 2. 大表侧热点Key加随机前缀非热点Key保持原样 val df1Salted df1.withColumn(salted_key, when($user_id.isin(hotKeys.toSeq: _*), concat($user_id, lit(_), (rand() * 4).cast(int))) .otherwise($user_id)) // 3. 小表侧热点Key膨胀复制需要与加盐侧的前缀范围一致 val df2Salted df2.withColumn(salted_key, when($user_id.isin(hotKeys.toSeq: _*), explode(array(lit(0), lit(1), lit(2), lit(3)) .zipWithIndex.map(x concat($user_id, lit(_), lit(x._2))))) .otherwise($user_id)) // 4. 用加盐后的 Key 做 Join val result df1Salted.join(df2Salted, Seq(salted_key), inner)注意这个方案有个前提小表侧的膨胀复制数必须和大表侧的随机前缀扇出数一致。上例里面大表侧 rand()*4 生成了 0~3 共 4 个前缀小表侧也要膨胀复制出 4 份才能保证9527_0能匹配到9527_0。加盐的代价是数据量膨胀。热点 Key 拆成 N 份后小表侧的记录数会变为原来的 N 倍Shuffle 和最后再聚合的开销也相应增加。所以加盐的扇出数不是越大越好建议从 4 到 8 开始观察任务耗时变化再逐步调整。另外加盐方案主要解决 Join 本身的倾斜。如果倾斜发生在 GroupBy 聚合阶段一般采用下面说的两阶段聚合。4.3 两阶段聚合先局部汇总再全局汇总的取舍两阶段聚合的方案主要应对GroupBy 聚合倾斜。它的核心思路是数据先带着随机前缀做第一次局部聚合去掉前缀后做第二次全局聚合。// 第一步加盐分组局部聚合 val partial df.withColumn(salt, (rand() * 10).cast(int)) .groupBy(user_id, salt) .agg(sum(amount).as(partial_amount)) // 第二步去盐分组全局聚合 val final partial.groupBy(user_id) .agg(sum(partial_amount).as(total_amount))这个方案的收益很明显原本全部集中在同一个分区的热点 Key现在被拆到 10 个分区分别聚合每个分区的负载变为原来的 10%。代价是执行了两遍聚合额外的 Shuffle 开销大约在 10%~30% 之间但如果热点 Key 导致单个 Task 跑 2 小时这点开销完全值得。使用两阶段聚合要注意一个问题它在某些聚合函数下不适用。比如count(distinct ...)或者approx_count_distinct在局部聚合阶段做去重计数再在全局阶段合并结果不一定准确需要做专门的去重设计。另外如果聚合函数本身是有序敏感的比如 collecting list 时需要保序两阶段聚合会打乱顺序。对于生产环境我通常建议如果热点 Key 只占少数几个值不要对所有 Key 加盐只对热点 Key 加盐。方法是先识别热点 Key再在when条件里对热点 Key 加盐其他 Key 保持不变这样能减少不必要的局部聚合开销。4.4 空值与默认值的清洗多数倾斜的源头其实不是业务热点一个很容易被忽略但频繁出现的情况是大量记录的关键字段是空值或默认值。比如行为日志里的 user_id 在下游数仓里是空 String或者为-9999这类业务规定的默认值。这种情况下空值/默认值占据的数据量可能是所有热点业务 Key 的总和它们全部会被哈希到同一个分区制造出一个巨大的倾斜分区。这种倾斜的解法说起来最简单但生产里很多人会漏掉过滤不需要的空值记录。如果下游统计根本用不到未登录用户的行为直接在 Join 前把空 user_id 过滤掉。这一步能把数据量缩小 20% 甚至更多。如果空值记录必须保留可以给空值赋一个随机值或随机前缀这样它们会被均匀分散到所有分区。// 将空值替换为带随机后缀的Key避免集中在同一分区 df.withColumn(join_key, when($user_id.isNull || $user_id.equalTo(-9999), concat(lit(unknown_), lit((rand() * 100).cast(int)))) .otherwise($user_id))这里有一个生产上的教训某次我排查一个 Join 倾斜Web UI 显示同一个分区承接了 35% 的数据以为是头部用户热点。结果跑频率统计后发现排名第一的 Key 实际上是空字符串占了全部数据的 28%。业务热点 Key 只排在第二位。如果没做字段频率分析就盲目加盐不仅解决不了问题还会因为错误膨胀数据量让作业更慢。所以任何倾斜排查都要从字段频率统计开始而不是从业务直觉开始。4.5 全局排序/去重场景中的Reducer端重构倾斜不一定只出现在 Join 和 GroupBy 上。全局排序、全局 TopN、distinct也经常出现某个分区数据量异常。原因是这些操作在 reducer 端需要把全局数据集中到一个或少数几个分区上处理。以全局排序为例Spark 的sortBy最终会把数据分成多个排序分区但如果排序列的分布不均——比如大部分记录的排序 Key 都落在某个区间——这个区间对应的分区就会聚集大部分数据。这种情况通常需要配合range partitioning而不是默认的 hash 分区// 使用 repartitionAndSortWithinPartitions 自定义区间分区 df.repartitionByRange(col(score).desc)repartitionByRange会先采样预估一个均匀的分桶边界然后按范围分区这样可以让数据在多个分区之间更均衡。而distinct倾斜的解法本质上也是两阶段去重先加盐实现局部去重再去掉盐做最终去重。相比全局 distinct这个方案的输出结果完全一致但 Shuffle 负载会被分散。这里要特别提醒一个误区很多人遇到 Task 倾斜第一反应是增加spark.sql.shuffle.partitions以为把分区数调大就能让热点 Key 的数据更加分散。但热点 Key 的哈希结果在一个固定分区器下是确定的分区数从 200 调到 2000热点 Key 只会从压垮 1 个分区变成还是压垮 1 个分区并没有把数据量平分到 10 个分区。真正让热点分散的手段只有加盐、范围分区或者改变分区器逻辑。关于这一点下一节我会展开讲。5. Spark2参数调优的实效边界配置不是越多越好5.1 spark.sql.shuffle.partitions 的真实作用范围spark.sql.shuffle.partitions是 Spark SQL 里最著名的参数之一。它的作用是设置 SQL 作业在发生 Shuffle 时产出的分区数量默认是 200。很多排障手册遇到倾斜就建议调大它但实际上要看倾斜的成因。分两种情况来理解如果倾斜是因为分区数太少导致的比如集群有 500 个 Executor 核但 Shuffle 分区只有 200意味着 500 个核里只有 200 个在跑 Shuffle Task有 300 个核空转。此时调大分区数到 500 或者 1000能让任务在更多 Executor 上并行执行倾斜压力会被打散。这种情况调大分区是有效的。如果倾斜是个别 Key 数据量巨大导致的分区别管是 200 还是 2000热点 Key 的哈希结果仍然只会落在一个分区上。调大分区数只会让这个热点分区的数据量占比保持不变问题不会消失。更糟的是分区数调大后所有非热点分区都只处理很少的数据并发任务数变多调度开销和网络连接数反而上升作业总时间可能不降反升。所以调大spark.sql.shuffle.partitions的正确场景是并行度不足而不是数据倾斜。判断标准很简单看你 Web UI 上这个 Stage 的 Active Task 数和当前分区的对应关系。如果每个分区上只跑了一个 Task而且其他 Executor 核都在空转优先调大分区如果某个 Task 的输入大小是其他 Task 的几十倍调大分区只会把瓶颈转移到下一个 Stage。5.2 内存参数与溢写警惕倾斜场景下的伪内存不足倾斜 Task 处理的数据量远超常规时往往会伴随一个现象ExecutorLostFailure或者java.lang.OutOfMemoryError。很多同学会以为是 Executor 内存不够于是盲目调大spark.executor.memory。但这不是问题的根源。在 Spark2 的 Unified Memory 模型下真正的问题是倾斜 Task 在一个 Executor 内堆积了远超预期的数据占满了执行内存后还可能占用存储内存进而引发整个 Executor 的内存竞争。这时候加内存只是在缓解症状没有解决热点数据依然压在一个分区上的问题。实际操作中我更倾向于关注spark.shuffle.spill.compress和spark.shuffle.io.maxRetries这些与溢写行为相关的参数。倾斜 Task 的溢写是它的正常保护机制如果溢写发生在 shuffle read 阶段说明这个 Task 的输入就已经超过了内存。有个经验值可以参考如果倾斜 Task 的 Shuffle Read 数据量超过 Executor 执行内存的 2 倍这个 Task 大概率会被拖垮。此时要判断的是到底是先调整 Key 的分布策略而不是继续加内存。5.3 哪些调优是心理安慰一个配置组合的实测对比我见过有人在一次倾斜问题排查中同时调整了spark.executor.memory、spark.executor.cores、spark.sql.shuffle.partitions、spark.default.parallelism、spark.memory.fraction结果作业还是慢但配置已经改得妈都不认识。这里我想给你一个我非常确定的经验参数调优永远解决不了 Key 本身歪斜的问题。数据倾斜是数据分布问题不是资源问题。与其堆参数不如先解决热点数据的分布。做一个简单的验证你就明白了同一个带倾斜的作业在一组资源完全相同的集群上仅调整spark.sql.shuffle.partitions从 200 到 400你大概率会看到只有一个热点 Task 的耗时从 60 分钟变成 58 分钟作业总时长几乎不变。而把热点 Key 加盐拆成 4 份之后同一个热点的 Task 可能只需要 20 分钟就能跑完。所以我的实际建议是遇到倾斜问题参数的调整放在最后一步并且只关注三个spark.sql.shuffle.partitions的值是否与集群并行度匹配。spark.sql.autoBroadcastJoinThreshold是否需要针对小表场景放大。spark.sql.join.preferSortMergeJoin在 Spark2.4 版本里可以针对小表场景临时关闭让优化器选择广播 HashJoin。其他参数比如spark.executor.memory、spark.memory.storageFraction等等不要为了倾斜问题而轻易调整因为它们往往属于资源规划范畴而不是数据分布范畴。6. 工程侧的长效治理把数据倾斜关进笼子6.1 Key设计规范与数据模型评审数据倾斜是一次性的问题但它的产生根源往往是长期的。生产集群每天都在产生新任务如果不在工程侧建立规范每次遇到倾斜都要临时救火。我在这里想分享一个我所在团队抽象掉具体组织信息在实践中形成的三条规则第一关键业务字段要先做值域分析再进入模型评审。对即将作为 Join Key、GroupBy Key 的字段上线前先统计字段的值频率分布看是否有 top 值占比超过 1% 的情况。如果超过这个字段进入下游链路时就必须设计好倾斜预案。比如用户表与行为表 Join提前识别头部用户的比例评估是否需要分桶策略。第二空值和默认值必须在数仓开发规范里明确处理方式。约定主键字段留空时的填充规则建议保留原始值但不参与作为业务 Join Key。可以在基表层面建好清洗逻辑而不是每次在作业里临时处理。第三克制万能大宽表的冲动。宽表设计通常会引入大量 Join而且每个 Join 的 Key 都可能是热点。每多一次 Join就多了一次倾斜的可能。能通过维度退化处理的就不要引入额外的维度表关联。6.2 常态化巡检把倾斜排查变成例行体检既然数据倾斜的定位链路已经比较标准了我建议团队把排查过程沉淀成一个固定的巡检脚本每天任务跑完后自动输出疑似倾斜的 Stage 列表。我们的做法是从 Spark History Server 上拉取每个 Job 的执行信息计算每个 Stage 的 Task 最大耗时与中位耗时的比值超过阈值就自动告警。同时把 Shuffle Read 大小的极差也纳入告警指标。这个巡检脚本不需要很复杂本质上就是把人工看 Web UI 的逻辑固化下来。另外在数据模型发布流程上建议加一步倾斜影响评估对这个任务涉及的 Join 和 GroupBy 字段做一次抽样频率统计输出 Top Key 占比。如果某个 Key 占比超过 5%就要求开发同学说明是否有加盐、广播等应对预案。这个前置检查能减少大部分发布上线后的倾斜事故。6.3 个人经验与最后提醒数据倾斜这个事情处理方式并不复杂但很容易被带偏。我见过太多同事一上来就调参数或者直接换分区器最后把问题搞得更严重。我的经验是拿到任何一个倾斜现象先花 3 分钟理清楚数据集中在哪个 Key 上这个 Key 经过哪个算子传到了哪个分区然后再选择对策。从技术方案角度看广播 Join 适用小表加盐适用于单 Key 热点两阶段聚合适用于聚合场景过滤空值适用于脏数据场景四管齐下基本能覆盖生产环境中九成以上的倾斜问题。剩下的少量场景需要结合数据业务特征具体设计。最后再分享一个小技巧修复倾斜问题后不要只改作业代码建议把排查过程中得到的数据分布结论和解决方式记录在任务注释里。半年后再看同一个任务时你会感谢当时的自己。这个习惯在很多团队里都能大幅降低同类问题的返工率和排查时间。