一、什么是数据倾斜

咱们先来聊聊什么是数据倾斜。简单来说，就是在处理大数据的时候，某些任务分配到的数据量远远超过其他任务，导致这些任务运行特别慢，甚至直接卡死。比如你有一堆数据要处理，结果其中某个 key 的数据特别多，负责处理这个 key 的节点就累得半死，而其他节点却闲得发慌。

举个生活中的例子，就像你去超市结账，本来有 10 个收银台，结果 90% 的人都挤在一个收银台排队，其他 9 个收银台却没人用，这效率能高吗？

二、数据倾斜的常见场景

数据倾斜不是随便发生的，它通常出现在以下几种情况：

  1. Join 操作：比如两张表关联时，某个关联字段的值特别多。
  2. Group By：分组统计时，某个分组的记录数远超其他分组。
  3. Distinct 去重：某些值重复率极高，导致去重计算压力大。
  4. Partition 不均匀：数据分布不均，某些分区数据量爆炸。

示例 1：Spark SQL 中的 Join 倾斜（技术栈：Apache Spark）

# 假设有两张表：orders（订单表）和 users（用户表）
# 其中，某些用户下了大量订单，导致 join 时数据倾斜

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SkewJoinExample").getOrCreate()

# 模拟数据：user_id=1 的用户有 10000 条订单，其他用户只有 10 条
orders_data = [(1, f"order_{i}") for i in range(10000)] + [(i, f"order_{i}") for i in range(2, 100)]
users_data = [(1, "VIP用户"), (2, "普通用户"), (3, "普通用户")]

orders_df = spark.createDataFrame(orders_data, ["user_id", "order_id"])
users_df = spark.createDataFrame(users_data, ["user_id", "user_type"])

# 直接 join 会导致数据倾斜
joined_df = orders_df.join(users_df, "user_id")
joined_df.explain()  # 查看执行计划，会发现某个 task 处理的数据量极大

这个例子中，user_id=1 的数据量极大，直接 join 会导致该 task 处理时间过长。

三、数据倾斜的解决方案

既然知道问题出在哪儿，那咱们就得想办法解决。下面介绍几种常见的解决方案。

1. 加盐（Salting）

加盐的核心思想是把倾斜的 key 打散，让数据均匀分布。比如原本 user_id=1 的数据特别多，我们可以给它加上随机前缀，让它变成 1_0, 1_1, 1_2...，然后再做聚合。

示例 2：Spark 中使用加盐解决 Group By 倾斜

from pyspark.sql.functions import col, concat, lit, randint

# 假设我们要按 user_id 分组统计订单数，但 user_id=1 的数据量极大
# 先给 user_id=1 的数据加上随机前缀（0-9）
salted_orders_df = orders_df.withColumn(
    "salted_user_id",
    concat(col("user_id"), lit("_"), (randint(0, 9)))  # 加盐：user_id -> user_id_random
)

# 现在按 salted_user_id 分组统计
grouped_df = salted_orders_df.groupBy("salted_user_id").count()

# 如果需要还原原始 user_id，可以去掉后缀
final_result = grouped_df.withColumn(
    "user_id",
    col("salted_user_id").substr(1, 1)  # 提取原始 user_id
)

2. 两阶段聚合

两阶段聚合的思路是先局部聚合，再全局聚合。比如先对倾斜的 key 做一次小范围聚合，然后再整体聚合。

示例 3：Spark 中的两阶段聚合

# 第一阶段：局部聚合（比如按 user_id 和随机数分组）
stage1_df = orders_df.withColumn("random", (randint(0, 9)))  # 添加随机列
stage1_result = stage1_df.groupBy("user_id", "random").count()

# 第二阶段：全局聚合（去掉随机列，再次聚合）
final_result = stage1_result.groupBy("user_id").sum("count")

3. 倾斜数据单独处理

如果某个 key 的数据量特别大，可以把它单独拎出来处理，再和其他数据合并。

示例 4：Spark 中分离倾斜数据

# 找出倾斜的 key（比如 user_id=1）
skewed_data = orders_df.filter(col("user_id") == 1)
normal_data = orders_df.filter(col("user_id") != 1)

# 分别处理
skewed_result = skewed_data.groupBy("user_id").count()
normal_result = normal_data.groupBy("user_id").count()

# 合并结果
final_result = skewed_result.union(normal_result)

4. 调整分区策略

如果数据倾斜是因为分区不均，可以手动调整分区策略，比如使用 repartition 或自定义分区器。

示例 5：Spark 中调整分区

# 默认分区可能导致数据倾斜
orders_df.rdd.getNumPartitions()  # 查看当前分区数

# 手动调整分区
repartitioned_df = orders_df.repartition(100, "user_id")  # 按 user_id 重新分区

四、不同场景下的优化策略

不同的数据处理框架（如 Spark、Flink、Hadoop）对数据倾斜的处理方式略有不同，但核心思路是相通的。

1. Spark 优化建议

  • 使用 broadcast join 避免 shuffle。
  • 调整 spark.sql.shuffle.partitions 增加分区数。
  • 使用 AQE（自适应查询执行）（Spark 3.0+）。

2. Flink 优化建议

  • 使用 rebalance 算子均衡数据分布。
  • 对倾斜 key 使用 localKeyBy 预处理。

3. Hadoop MapReduce 优化建议

  • 自定义 Partitioner 让数据分布更均匀。
  • 使用 Combiner 减少 shuffle 数据量。

五、总结

数据倾斜是大数据处理中的常见问题，但只要我们掌握核心思路（加盐、两阶段聚合、倾斜数据分离、调整分区），就能有效解决。不同的框架可能有不同的优化手段，但万变不离其宗。

最后提醒几点：

  1. 监控任务运行情况，及时发现倾斜问题。
  2. 合理设置资源，避免因倾斜导致集群资源浪费。
  3. 测试优化效果，确保方案真正有效。