1. 从一次深夜告警说起：数据倾斜的“威力”

凌晨两点，手机突然震动，告警信息显示线上一个关键的Spark数据处理任务已经卡在最后几个Stage超过两个小时。登录集群监控一看，一个Reduce阶段的进度条在99%的位置纹丝不动，而对应的Executor日志里，某个Task的GC时间异常地长，堆内存几乎打满。其他几百个Task早已完成，资源闲置，唯独这一个Task还在苦苦挣扎。这就是典型的数据倾斜（Data Skew）现场——少量Key承载了海量数据，导致单个计算节点成为整个作业的瓶颈，拖垮了整个任务的执行效率，甚至直接导致OOM（内存溢出）失败。

数据倾斜不是Spark的专利，但却是Spark开发者和数据工程师们最常遇到、也最头疼的性能问题之一。它本质上是数据分布不均的问题：在Shuffle（数据混洗）过程中，大量数据被分配到了同一个或少数几个分区（Partition），导致这些分区对应的Task处理的数据量远大于其他Task。想象一下，100个人分1000个包裹，理想情况是每人10个。但如果其中一个人分到了900个，其他人每人只分到1个，那么整个分拣工作的完成时间就取决于那个拿了900个包裹的人。在分布式计算中，这个“倒霉”的Task就成了木桶的最短木板。

处理数据倾斜，远不止是调优几个参数那么简单。它要求我们深入理解Spark的Shuffle机制、数据本身的业务特性，并掌握一系列从预防、诊断到修复的“组合拳”。接下来，我将结合多年实战中踩过的坑和总结的经验，系统性地拆解数据倾斜的成因、定位方法和解决方案。

2. 数据倾斜的根因剖析：不只是Key分布不均

很多人认为数据倾斜就是某些Key的数据量过大。这没错，但只看到了表面。我们需要更深入地理解，在Spark的哪些操作下，这种分布不均会被放大，以及除了业务数据本身，还有哪些技术因素会加剧倾斜。

2.1 哪些Spark操作最容易引发倾斜？

数据倾斜主要发生在需要进行Shuffle的操作中，因为Shuffle决定了数据如何跨节点重新分布。

1. 聚合类操作（GroupByKey, ReduceByKey, AggregateByKey） ：这是倾斜的“重灾区”。如果某个Key对应的记录数异常多，那么所有这个Key的数据都会被发送到同一个Reduce Task进行处理。例如，在统计用户行为日志时，如果存在一个“默认用户”或“测试用户”ID，其日志量可能占全量的一半以上。
2. 连接操作（Join） ：特别是大表与小表的Join。如果小表（广播Join除外）中某个Key的数据量很大，或者两张表都存在某个热点Key，那么在Shuffle过程中，这些Key对应的分区就会负载过重。更隐蔽的一种情况是，参与Join的字段存在大量空值（NULL），这些空值在Shuffle时可能被分配到同一个分区。
3. 去重操作（Distinct） ：底层通常通过 ReduceByKey 或 GroupBy 实现，因此同样受Key分布影响。
4. 重分区操作（Repartition, Coalesce） ：如果直接使用 repartition 而不指定分区字段，或者指定的字段本身分布不均，就会人为制造出倾斜的分区。

2.2 倾斜的“放大器”：资源分配与数据本地性

单纯的数据分布不均，如果量级不大，可能不会造成严重问题。但以下几个因素会像放大器一样，让问题急剧恶化：

• 不合理的分区数 ：如果设置的分区数（ spark.sql.shuffle.partitions 或 RDD的 partition 数）过少，那么每个分区承载的数据量本身就很大，热点Key的负面影响会更显著。反之，分区数过多，则管理开销增大，但可能让数据分布更均匀一些（尽管不能根治倾斜）。
• Executor内存配置 ：处理热点分区的Task需要将大量数据拉取到内存中进行计算或聚合。如果Executor的堆内存（ spark.executor.memory ）设置过小，极易引发频繁的Full GC甚至OOM。而Spark的机制是，一个Stage中只要有一个Task失败数次，整个作业就可能失败。
• 数据序列化与压缩 ：如果Shuffle数据没有压缩（ spark.shuffle.compress ），网络传输和磁盘I/O的压力会倍增，加剧热点Task的延迟。不高效的序列化方式（如Java序列化）也会增加CPU和内存开销。

理解这些根因和放大器，是我们制定应对策略的基础。接下来，我们需要一套方法来精准定位倾斜点。

3. 定位倾斜：从监控大盘到代码行级排查

当作业变慢或失败时，如何快速确定是数据倾斜，并找到那个“罪魁祸首”的Key？盲目猜测和修改代码是低效的。一套清晰的排查链路至关重要。

3.1 第一步：集群监控与Spark UI诊断

这是最直观的入口。以开头提到的场景为例：

1. 查看Stage时间线 ：在Spark UI的Stages页，找到执行时间异常长的Stage。观察其“Summary Metrics”，重点看“Duration”的分布。如果中位数（Median）很小，但最大值（Max）极大，例如中位数10秒，最大值2小时，这强烈暗示了数据倾斜。
2. 分析Task指标 ：点进那个异常的Stage，查看Task的“Duration”、“GC Time”、“Shuffle Read Size”、“Records Read”等指标。排序“Shuffle Read Size”或“Records Read”，通常排名第一的Task其读取量会比其他Task高出几个数量级（比如其他Task读100MB，它读10GB）。这个Task所在的分区就是热点分区。
3. 检查Executor日志 ：如果Task失败，去对应的Executor日志中查找OOM或StackOverflow错误堆栈。通常错误信息会指向具体的Shuffle读取或聚合代码行。

3.2 第二步：数据采样与热点Key识别

通过UI我们知道了有倾斜，但还不知道是哪个Key导致的。这时需要在代码中引入数据采样分析。

方法一：使用 sample 进行抽样统计

val skewedRDD = ... // 你的RDD或DataFrame转换成的RDD
// 采样10%的数据
val sampleRDD = skewedRDD.sample(false, 0.1)
// 统计每个Key的出现次数，并排序
val sampleKeyCount = sampleRDD.map((_, 1)).reduceByKey(_ + _).map{case (key, count) => (count, key)}.sortByKey(false)
// 取Top N的热点Key
val topNhotKeys = sampleKeyCount.take(10).map(_._2)
topNhotKeys.foreach(println)

这个方法适合数据量大的情况，通过采样快速定位热点Key。但要注意，采样可能漏掉一些非常集中但总量不大的Key。

方法二：使用 countByKey （仅适用于小规模RDD） countByKey 会将结果收集到Driver端，因此如果Key空间很大或数据量大，会导致Driver OOM。仅在你确信Key数量不多时使用。

方法三：SQL方式探查（针对DataFrame）

df.groupBy(“your_key_column”).count().orderBy(desc(“count”)).limit(10).show()

这是最常用、最直观的方式，直接对DataFrame操作，快速看到热点Key及其数量。

定位到热点Key后，我们就可以针对性地“下药”了。解决方案分为几个层次，从治标到治本。

4. 解决方案一：参数调优与资源扩容（治标不治本）

对于倾斜程度不特别严重，或者只是临时应急的场景，可以尝试调整Spark配置和资源。这通常不能根治问题，但可能让作业先跑起来。

1. 增加Shuffle分区数 ：通过 spark.sql.shuffle.partitions （默认200）或 spark.default.parallelism 调大。这相当于把原来承载大量数据的一个分区，拆分成更多的小分区，让热点Key的数据分散到更多Task中处理。但注意，如果某个Key的数据量实在太大（比如几十亿条），仅仅增加分区数，这个Key的数据还是会集中在与其哈希值对应的那几个分区里，无法打散。 公式不总是有效，但可以尝试将其设置为 core总数 * 2 ~ 4 倍 。
2. 启用Shuffle压缩并选择高效序列化 ：设置 spark.shuffle.compress=true （默认true）并使用 spark.io.compression.codec=snappy （或lz4）来减少Shuffle数据量。设置 spark.serializer=org.apache.spark.serializer.KryoSerializer 并注册类，以降低序列化开销。
3. 增加Executor内存与核数 ：直接给处理热点分区的Task“喂”更多资源。调整 spark.executor.memory , spark.executor.memoryOverhead , spark.executor.cores 。这是最直接的“土豪”做法，成本高，且对于极端倾斜（单个Key数据量超过Executor内存）依然无效。
4. 提高Shuffle操作的并行度与超时 ：对于Broadcast Hash Join，可以调大 spark.sql.autoBroadcastJoinThreshold 让小表更容易被广播，避免Shuffle。对于不可避免的Shuffle Join，可以设置 spark.sql.adaptive.enabled=true （Spark 3.x后推荐开启），让Spark AQE（自适应查询执行）动态调整执行计划。同时，适当调大 spark.sql.broadcastTimeout 和 spark.network.timeout ，防止因数据量大、传输慢导致的误报失败。

注意 ：参数调优是“麻醉剂”，不是“手术刀”。它缓解了症状，但没有解决数据分布不均的根本问题。长期来看，我们需要从数据和处理逻辑层面入手。

5. 解决方案二：业务逻辑与数据处理层面的优化（核心手段）

这才是解决数据倾斜的根本之道，需要结合具体的业务场景和数据处理逻辑。

5.1 过滤异常数据

很多时候，热点Key是无效的“脏数据”，比如：

• 日志中的测试账号、默认用户（如 user_id=0 或 ‘null’ ）。
• 爬虫或机器产生的垃圾流量。
• 由于程序BUG产生的重复或无效记录。

操作 ：直接在产品逻辑上过滤掉这些数据。例如：

val cleanDF = originalDF.filter(col(“user_id”) =!= 0 && col(“user_id”).isNotNull)

在过滤前，最好先评估这些异常数据是否还有分析价值（比如单独分析测试行为），如果没有，果断过滤。

5.2 热点Key单独处理（两阶段聚合）

这是处理聚合操作倾斜的经典方法，尤其适用于 count 、 sum 、 avg 等可分解的聚合函数。其核心思想是：将聚合分成局部聚合和全局聚合两步。

原理 ：先在每个分区内对Key进行打散（加盐）做一次预聚合，减少Shuffle数据量；然后对打散后的结果进行第二次聚合，得到最终结果。

场景 ：统计每个商品的销售额，但某几个“爆款”商品的记录量巨大。

步骤 ：

1. 局部聚合（加盐） ：给每个Key加上一个随机前缀（盐），比如 商品A 变成 商品A_1 , 商品A_2 , … 商品A_n 。这样，原来 商品A 的海量数据就被随机分散到多个不同的新Key中，在第一个Shuffle阶段被送到不同的Task进行局部聚合。

   import org.apache.spark.sql.functions._
   val saltNum = 10 // 假设我们打散成10份
   val saltedDF = df.withColumn(“salted_key”, concat(col(“product_id”), lit(“_”), (rand() * saltNum).cast(“int”)))
   val firstAggDF = saltedDF.groupBy(“salted_key”).agg(sum(“amount”).as(“partial_sum”))
   

2. 还原Key并全局聚合 ：将加盐的Key还原回原始Key，然后进行第二次聚合。

   val originalKeyDF = firstAggDF.withColumn(“original_key”, split(col(“salted_key”), “_”).getItem(0))
   val finalResultDF = originalKeyDF.groupBy(“original_key”).agg(sum(“partial_sum”).as(“total_amount”))
   

为什么有效 ：第一次Shuffle，数据被随机打散，负载相对均衡。第二次Shuffle，虽然Key还原了，但经过第一次聚合后，每个Key的数据量已经大大减少（从原始记录数变成了 盐值个数 条中间结果），因此倾斜程度被极大缓解。

实操心得 ：盐值个数（ saltNum ）的选择很重要。太小，打散效果有限；太大，会增加额外的Shuffle开销。通常可以根据热点Key的数据量是平均值的多少倍来估算，比如100倍的热点，可以尝试用50-100的盐值。可以通过采样数据来测试不同盐值下的数据分布。

5.3 倾斜Join的优化

对于Join操作，如果有一张表很小，首选 广播Join（Broadcast Hash Join） ，完全避免Shuffle。但如果两张表都很大，且存在倾斜，就需要特殊处理。

方法一：拆分热点Key，非热点正常Join 这是最有效的方案之一。思路是将存在热点Key的数据和正常数据分开处理。

1. 识别热点Key ：通过采样或历史知识，找出维表（或事实表）中的热点Key列表。
2. 数据拆分 ：
   • 将事实表中与热点Key关联的数据拆分出来（ fact_hot ）。
   • 将维表中热点Key的数据拆分出来（ dim_hot ）。
   • 剩余的非热点数据分别为 fact_normal 和 dim_normal 。
3. 分别Join ：
   • fact_hot 与 dim_hot 进行Join。因为 dim_hot 数据量小，可以将其广播，实现高效的 Broadcast Join 。
   • fact_normal 与 dim_normal 进行普通的 Shuffle Hash Join 或 Sort Merge Join 。
4. 合并结果 ：将两部分Join的结果用 union 合并。

// 假设hotKeys是一个已知的热点Key集合
val hotKeysBroadcast = spark.sparkContext.broadcast(hotKeys)

val factDF = …
val dimDF = …

val factHot = factDF.filter(col(“join_key”).isin(hotKeysBroadcast.value: _*))
val factNormal = factDF.filter(!col(“join_key”).isin(hotKeysBroadcast.value: _*))

val dimHot = dimDF.filter(col(“key”).isin(hotKeysBroadcast.value: _*))
val dimNormal = dimDF.filter(!col(“key”).isin(hotKeysBroadcast.value: _*))

// 热点部分使用广播Join
val joinedHot = factHot.join(broadcast(dimHot), factHot(“join_key”) === dimHot(“key”))
// 正常部分使用普通Shuffle Join
val joinedNormal = factNormal.join(dimNormal, factNormal(“join_key”) === dimNormal(“key”))

val finalResult = joinedHot.union(joinedNormal)

方法二：使用随机前缀扩容维表 当热点Key在维表中，且维表无法被广播（大小超过300MB默认阈值）时，可以将维表中的热点Key复制多份（加随机前缀），同时将事实表中的对应Key也加上相同范围的前缀，从而将一次倾斜的Join变成多次负载均衡的Join。

步骤 ：

1. 对维表中的热点Key，复制成N份（如10份），每条数据加上前缀 [0-N)_ 。
2. 对事实表中的热点Key，在Join Key字段上也加上一个 [0-N) 的随机前缀。
3. 进行Join，此时一个热点Key的数据会被分散到N个不同的Join任务中。
4. 对结果进行去前缀处理，得到最终数据。

这个方法实现起来比方法一更复杂，需要确保事实表和维表的“加盐”规则能正确匹配。

5.4 使用Spark 3.x AQE的倾斜Join优化

如果你使用的是Spark 3.0及以上版本，并且开启了AQE（ spark.sql.adaptive.enabled=true ），那么恭喜你，Spark提供了一种原生的倾斜Join处理能力。

原理 ：AQE会在运行时统计每个Shuffle分区的数据大小，如果发现某个分区远远大于其他分区（通过 spark.sql.adaptive.skewJoin.skewedPartitionFactor 和 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes 参数判断），它会自动将这个倾斜的分区 拆分 成多个更小的子分区，然后分别与另一张表的对应分区进行Join。

配置 ：

spark.sql.adaptive.enabled true
spark.sql.adaptive.skewJoin.enabled true
spark.sql.adaptive.skewJoin.skewedPartitionFactor 5 # 倾斜因子，默认5。分区大小 > 中位数 * 5 则判定为倾斜
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes 256MB # 倾斜分区最小阈值，默认256MB

优点 ：无需修改业务代码，由Spark引擎自动完成，对用户透明。这是处理未知倾斜或临时性倾斜的利器。

局限 ：AQE的倾斜优化目前主要针对Sort Merge Join。对于Shuffle Hash Join的支持可能有限。它也无法解决因某个Key数据量过大导致单个分区无论如何拆分都超过Executor内存的极端情况。

6. 解决方案三：从数据源头与模型设计根治

最高级的解决方案，是在数据产生的源头和数仓模型设计阶段，就避免倾斜的发生。

1. 设计合理的业务键 ：避免使用像 user_id=0 、 device_id=‘unknown’ 这样的默认值作为Key。可以考虑使用更均匀分布的代理键，或者在日志埋点时，为这些特殊值生成随机的、符合分布的ID。
2. ETL过程引入随机因子 ：在数据清洗和入库的早期阶段，如果预见到某个字段未来可能成为倾斜的Key（比如按城市分组，但“其他”或“未知”城市占比很高），可以提前进行打散或分类。
3. 分层建模时考虑数据分布 ：在构建维度表和事实表时，评估连接键的基数（Cardinality）和分布。对于极高基数的字段（如用户ID），Join成本天然就高，需要考虑是否采用其他查询模式。对于低基数但分布不均的字段，可以在汇总层（DWS层）提前进行聚合，减少下游查询时的数据量。
4. 选择合适的分区键 ：对于需要持久化存储的表（如Hive表），选择分区字段时，不仅要考虑查询过滤条件，还要考虑该字段值的分布是否均匀。避免使用值分布极度不均的字段作为唯一的分区键。

7. 实战案例：一个真实的数据倾斜排查与修复全流程

最后，分享一个我处理过的真实案例，串联起诊断和解决的全过程。

背景 ：一个每日运行的用户行为漏斗分析作业，突然从30分钟延长到3小时。作业主要是一个包含多个 groupBy 和 join 的复杂SQL。

排查过程 ：

1. Spark UI定位 ：发现一个以 groupBy session_id 为核心的Stage耗时占整体的85%。该Stage的Task读数据量中，最大值为120GB，中位数仅为1.2GB，倾斜比例高达100倍。
2. 热点Key识别 ：在代码中添加采样分析，发现 session_id 为 ‘-’ （表示无法获取或异常）的记录占总量的70%以上。原因是某次前端SDK升级导致错误，产生了大量无效会话。
3. 解决方案制定与实施 ：
   • 短期修复（治标） ：为了不影响当日报表产出，我们首先尝试了参数调优。将 spark.sql.shuffle.partitions 从200增加到800，并为该作业单独申请了内存更大的Executor（从8G增加到16G）。作业时间从3小时缩短到1.5小时，但仍不理想。
   • 业务逻辑修复（治本） ：与数据产品经理和前端团队确认， session_id=‘-’ 的记录无任何分析价值。立即修改ETL脚本，在数据接入层（ODS）就过滤掉所有 session_id 为无效值的记录。 where session_id != ‘-’ and session_id is not null 。
   • 长期优化 ：推动前端团队修复SDK的BUG，从源头杜绝无效数据的产生。同时在数仓设计文档中，明确此类默认值的处理规范。
4. 效果 ：经过业务逻辑过滤后，该作业次日运行时间恢复至25分钟，资源消耗降低60%。

这个案例告诉我们，参数调优能救急，但找到数据本身的脏数据根源并清洗，才是性价比最高的解决方案。同时，建立有效的数据质量监控，能在倾斜发生前就预警。

处理数据倾斜没有银弹，它是一个需要结合监控、分析、实验和业务理解的综合工程。从被动救火到主动预防，关键在于建立起对数据分布的敏感度，并在系统设计和开发初期就将“均匀分布”作为一个重要的非功能性需求来考虑。每一次对倾斜的深入排查，都是对业务数据和计算框架的一次再认识。