数据倾斜导致 Spark 作业慢？从原理到优化的完整攻略

摘要：

Spark 作业执行缓慢，数据倾斜是常见原因。本文介绍数据倾斜的产生原理、识别方法，以及提高并行度、两阶段聚合、广播连接、 AQE 自适应执行和 RSS 远程 Shuffle 服务等优化方案。

一、什么是数据倾斜

在 Spark 作业执行过程中，当某些分区（ Partition ）的数据量显著高于其他分区时，就会出现数据倾斜（ Data Skew ）现象。这种不均衡通常发生在 Shuffle 阶段，尤其是 groupByKey、reduceByKey、join 等操作中。由于 Key 分布不均，导致部分 Task 处理的数据量远超其他任务，成为整个作业的瓶颈。

数据倾斜的典型表现包括：

• 个别 Task 执行时间远超其他 Task ，例如 90% 的 Task 在 1 分钟内完成，剩余 10% 却耗时超过 1 小时。
• Executor 频繁发生 GC （垃圾回收）或内存溢出（ OOM ）。
• Stage 长时间处于"Running"状态但无进度推进。
• 资源利用率极不均衡， CPU 和内存集中在少数节点，其他节点资源闲置。

二、数据倾斜是如何产生的

数据倾斜的本质原因是数据分布不均。在进行 Shuffle 操作时， Spark 必须将各个节点上相同的 Key 拉取到某个节点上的一个 Task 来进行处理。此时如果某个 Key 对应的数据量特别大，该 Key 所对应的 Task 就会承担绝大部分计算负担。

以下是几类常见的倾斜诱因：

2.1 业务数据本身存在热点 Key

某些业务场景天然会产生数据分布不均。例如，电商平台的订单数据中，少数热门商品的订单量远超其他商品；日志数据中，某个特定用户 ID 或设备 ID 产生了大量记录；社交数据中，某些账号的互动数据占比极高。

2.2 空值或默认值集中

数据中的空值（null）、空字符串或默认值（如 0、-1）如果大量存在，会在 Shuffle 过程中全部聚集到同一个分区，导致严重倾斜。例如，某张订单表中 user_id = 0 的记录占了整张表的相当比例，通常是因为历史兼容逻辑导致某些记录的 user_id 未能正确写入。

2.3 Join 操作中的大表与大表关联

当两张大表进行 Join 操作时，如果 Join Key 的分布极度不均匀，就容易出现倾斜。尤其是当某一边存在大量重复的主键，或者某些热门 Key 在两边表中都占据较大比例时， Shuffle 过程会产生严重的负载不均。

2.4 分区策略不合理

使用自定义 Partitioner 时，如果没有充分考虑负载均衡，也会导致数据倾斜。默认的分区器（HashPartitioner）虽然能覆盖大多数场景，但在 Key 分布本身不均的情况下，仍无法避免倾斜问题。

三、如何识别数据倾斜

准确识别数据倾斜是解决它的前提。以下是几种常用的诊断手段：

3.1 使用 Spark Web UI 监控

Spark Web UI 是识别数据倾斜的首选工具。在作业执行期间或通过 History Server 查看已完成作业时，重点关注"Stages"标签页。每个 Stage 的 Task 执行时间分布直方图能够直观反映是否存在数据倾斜——如果大部分 Task 快速完成，而少数 Task 执行时间显著拉长，通常意味着数据分配不均。

具体可以关注以下指标：

• Task Duration：各 Task 的执行时间差异。
• Shuffle Read Size / Records：各 Task 读取的 Shuffle 数据量差异。
• GC Time：某些 Task 的 GC 时间异常偏高，可能是数据量过大导致。

3.2 分析 Driver 和 Executor 日志

通过查看 Driver 和 Executor 的日志输出，可以发现有价值的线索。例如， OOM 异常栈信息、 GC 频繁触发的日志、长延迟 Task 的记录等，都能帮助定位倾斜发生的位置。

3.3 采样统计 Key 分布

在开发阶段，可以通过对数据集进行采样统计，分析 Key 的频率分布，发现高频 Key 。例如，使用 df.groupBy("key").count().show() 可以查看各 Key 的出现次数，判断是否存在热点 Key 。

3.4 使用 AQE 的运行时统计信息

在 Spark 3.0 及以上版本中， AQE （ Adaptive Query Execution ，自适应查询执行）会在运行时收集 Shuffle 中间结果的统计信息。如果启用了 AQE ，可以通过其运行时统计数据来辅助判断是否存在数据倾斜。

四、数据倾斜的解决方案

针对数据倾斜问题，可以从多个层面入手，选择合适的解决方案。

4.1 提高 Shuffle 操作的并行度

这是最简单直接的方法。通过增加 Shuffle 分区数，可以让原本集中到少数分区的数据分散到更多分区中，从而减轻单个 Task 的负载。

在 Spark SQL 中，可以通过设置 spark.sql.shuffle.partitions 参数来调整 Shuffle 分区数，默认值为 200 ，在实际生产环境中通常需要调大。

需要注意的是，提高并行度并不能从根本上解决单个 Key 数据量过大的问题，但对于轻度倾斜的场景，往往能带来明显的改善。

4.2 两阶段聚合（局部聚合 + 全局聚合）

两阶段聚合是一种有效的倾斜缓解方案。其核心思路是：通过添加随机前缀，将原本集中在同一个 Task 的数据打散，先进行一次局部聚合，然后再去掉随机前缀进行全局聚合。

具体步骤如下：

• 第一阶段：对热点 Key 添加随机前缀（例如，将 key 变为 key_0、key_1……key_9），然后进行局部聚合。
• 第二阶段：去掉随机前缀，对局部聚合的结果进行全局聚合。

这种方案能够有效分散热点 Key 的负载，但也会增加一次额外的 Shuffle 操作。适用于热点 Key 明确且数量可控的场景。

4.3 将 Reduce Join 转为 Map Join （广播连接）

当 Join 操作涉及一张大表和一张小表时，可以考虑使用广播连接（ Broadcast Join ）。广播连接会将小表的数据分发到每个 Executor 的内存中，从而在 Map 端完成 Join 操作，避免 Shuffle 过程。

Spark 提供了 spark.sql.autoBroadcastJoinThreshold 参数来控制自动广播的阈值。当小表的估算大小低于该阈值时， Spark 会自动选择广播连接。

在 Spark SQL 中，也可以使用 Hint 来强制指定广播连接，例如 /*+ BROADCAST(dim) */。

4.4 过滤导致倾斜的 Key

如果业务上可以接受舍弃少量数据，可以直接过滤掉导致倾斜的 Key 。例如，某些默认值（如 user_id = 0 或 key = null）在业务中并不重要，可以在数据处理前将其过滤掉，从而避免倾斜。

4.5 开启 AQE 自适应查询执行

Spark 3.0 引入了 AQE （ Adaptive Query Execution ），能够在查询执行过程中，根据准确的运行时统计数据重新优化查询计划。 AQE 在以下场景中能够自动缓解数据倾斜：

• 动态切换 Join 策略：当 AQE 检测到 Join 的一方运行时数据量小于广播阈值时，会自动将 Sort-Merge Join 转换为 Broadcast Hash Join 。
• 动态合并 Shuffle 分区： AQE 会根据目标大小（spark.sql.adaptive.advisoryPartitionSizeInBytes）合并连续的 Shuffle 分区，避免过多小任务带来的调度开销。
• 动态优化倾斜 Join： AQE 能够自动识别倾斜的分区，并对其进行拆分（ Split ），将倾斜的数据重新分配到多个 Task 中执行。

要启用 AQE ，需要将 spark.sql.adaptive.enabled 设置为 true。在 Spark 3.0 及以上版本中， AQE 默认处于开启状态。

相关的重要参数包括：

4.6 使用随机前缀和扩容 RDD 进行 Join

对于两张大表 Join 且存在热点 Key 的场景，可以使用随机前缀方案：对大表中倾斜的 Key 添加随机前缀，将其分散到多个分区；同时对小表（或维度表）进行扩容，将每条记录复制多份，分别与大表中带有随机前缀的 Key 进行 Join 。这种方式能够将热点 Key 的负载分散到多个 Task 中。

这种方案适用于热点 Key 数量有限、且可以明确识别的场景。缺点是会增加数据量和代码复杂度。

五、腾讯云 EMR 提供的优化能力

除了上述通用的 Spark 优化方案，腾讯云弹性 MapReduce （ EMR ）还提供了一些额外的能力，帮助更好地应对数据倾斜和 Shuffle 性能问题。

5.1 RSS Uniffle 远程 Shuffle 服务

在大规模的 Spark 作业中， Shuffle 操作往往是性能瓶颈和稳定性风险的来源。尤其是在节点回收（如云原生场景下的 Pod 驱逐）的情况下， Shuffle 数据丢失会导致作业失败，需要重新计算，浪费大量资源。

腾讯云 EMR 新增了 RSS 集群类型，支持 Uniffle 组件，为 Spark 集群提供远程 Shuffle 服务（ Remote Shuffle Service ）。通过将 Shuffle 数据写到远程的 RSS 集群中，而非 Executor 本地磁盘，可以带来以下好处：

• 提升 Shuffle 任务的性能，尤其是在大数据量场景下。
• 增强任务稳定性，避免节点回收场景下的 Shuffle 数据丢失问题。
• 支持 EMR on CVM 和 EMR on TKE 两种集群类型。

使用 RSS Uniffle 时，需要在 Spark 配置中设置 spark.shuffle.manager 为 org.apache.spark.shuffle.RssShuffleManager，并配置相关的 ClassPath 参数。建议在任务级别启用，而非全局启用，以便灵活控制哪些作业使用 RSS 服务。

5.2 EMR 集群的 Spark 参数配置管理

腾讯云 EMR 提供了便捷的 Spark 参数配置管理界面，可以在控制台中直接修改 spark-defaults.conf 中的参数，无需逐台登录服务器进行配置。这对于批量调优 spark.sql.shuffle.partitions、spark.sql.adaptive.* 等参数非常有帮助。

在完成参数修改后， EMR 支持将配置变更应用到集群中，使参数调整能够快速生效。

六、 Spark 数据倾斜调优的实践建议

综合以上内容，在实际工作中遇到 Spark 作业因数据倾斜而变慢时，可以参考以下调优思路：

• 先诊断，后优化：通过 Spark Web UI 和日志定位倾斜发生的具体 Stage 和 Task ，明确是哪些 Key 或操作导致了倾斜。
• 从简单方案开始：优先尝试提高 Shuffle 并行度，或者开启 AQE 让 Spark 自动优化。
• 针对性处理热点 Key：如果能够明确识别热点 Key ，可以考虑使用两阶段聚合或随机前缀方案。
• 优化 Join 操作：对于大小表 Join ，优先使用广播连接；对于大表 Join 大表，可以考虑拆分热点 Key 或使用 RSS 服务提升 Shuffle 稳定性。
• 从数据源头治理：在数据采集和预处理阶段，尽量避免产生大量空值或默认值集中现象，可以从根本上减少数据倾斜的发生。

了解更多产品详情：腾讯云 EMR 产品页