数据倾斜导致标签计算任务跑了4小时——怎么优化到30分钟？

-- 倾斜SQL：用户行为表 JOIN 用户基础表
-- 问题：user_id有15%是NULL（未登录的匿名行为）
SELECT
  b.user_id,
  COUNT(*) AS event_cnt_7d,
  COUNT(DISTINCT event_name) AS event_types
FROM dwd_user_event a
JOIN dim_user_base b ON a.user_id = b.user_id
WHERE a.dt BETWEEN DATE_SUB('${bizdate}', 7) AND '${bizdate}'
GROUP BY b.user_id;
 
-- Spark UI表现：
-- Stage 3: 199/200 tasks completed in 2min
-- Stage 3: 1/200 tasks running for 3h 47min  <-- 这个Task处理了所有NULL

-- 方案一：过滤NULL值
SELECT
  b.user_id,
  COUNT(*) AS event_cnt_7d,
  COUNT(DISTINCT event_name) AS event_types
FROM dwd_user_event a
JOIN dim_user_base b ON a.user_id = b.user_id
WHERE a.dt BETWEEN DATE_SUB('${bizdate}', 7) AND '${bizdate}'
  AND a.user_id IS NOT NULL    -- 过滤NULL，消除倾斜
  AND b.user_id IS NOT NULL
GROUP BY b.user_id;
 
-- 效果：过滤掉15%的NULL数据后
-- 执行时间从4小时降到40分钟（NULL占了大量数据且全堆在一个Task）

-- 方案二：加盐打散
-- 场景：某测试设备device_id占了30%的数据（每天产生千万条事件）
 
-- Step 1: 识别热门key
SELECT device_id, COUNT(*) AS cnt
FROM dwd_device_event WHERE dt = '${bizdate}'
GROUP BY device_id ORDER BY cnt DESC LIMIT 20;
-- 发现device_id='test_device_001'占了30%数据
 
-- Step 2: 大表加盐（给热门key加随机后缀）
SELECT
  CASE
    WHEN device_id = 'test_device_001'
      THEN CONCAT(device_id, '_', CAST(FLOOR(RAND() * 10) AS STRING))
    ELSE device_id
  END AS device_id_salted,
  event_name, event_time
FROM dwd_device_event WHERE dt = '${bizdate}';
 
-- Step 3: 小表膨胀（维度表复制10份，和盐值对应）
SELECT CONCAT(device_id, '_', salt) AS device_id_salted, device_name, device_type
FROM dim_device
CROSS JOIN (SELECT EXPLODE(SEQUENCE(0, 9)) AS salt) t;
 
-- Step 4: 用salted key做JOIN
-- 原来1个Task处理30%数据 -> 现在10个Task各处理3%

适用场景：能明确识别出哪些key是热门key时。如果热门key数量少（几十个），加盐效果好；如果热门key数量多或者不确定，用方案三更方便。

方案三：AQE自动处理——Spark 3.0+的福音

Spark 3.0引入了Adaptive Query Execution（AQE），可以自动检测和处理数据倾斜。开启后不需要手动加盐，Spark会自动把倾斜的分区拆分成多个小分区。

-- 方案三：开启AQE自动处理倾斜
SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.skewJoin.enabled = true;
 
-- AQE倾斜检测参数
SET spark.sql.adaptive.skewJoin.skewedPartitionFactor = 5;
  -- 分区数据量超过中位数的5倍就认为是倾斜
SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 256m;
  -- 分区数据量超过256MB才触发倾斜处理
 
-- AQE还能自动合并小分区、自动选择JOIN策略
SET spark.sql.adaptive.coalescePartitions.enabled = true;
SET spark.sql.adaptive.localShuffleReader.enabled = true;
 
-- 开启AQE后，直接跑原始SQL即可，不需要手动加盐
SELECT b.user_id, COUNT(*) AS event_cnt_7d
FROM dwd_user_event a
JOIN dim_user_base b ON a.user_id = b.user_id
WHERE a.dt BETWEEN DATE_SUB('${bizdate}', 7) AND '${bizdate}'
GROUP BY b.user_id;

我在出海硬件品牌的实测效果：开启AQE后，那个4小时的倾斜任务自动降到了40分钟左右。再加上过滤NULL值，最终降到12分钟。AQE是Spark 3.0以上版本的标配，强烈建议生产环境默认开启。

三种方案对比

维度	过滤NULL	加盐打散	AQE自动处理
适用场景	NULL导致的倾斜	热门key导致的倾斜	通用倾斜场景
实现复杂度	低（加WHERE条件）	高（改SQL逻辑）	低（加配置参数）
对业务影响	丢弃NULL数据	无影响	无影响
Spark版本要求	无	无	Spark 3.0+
效果	非常好（消除根因）	好（手动拆分）	好（自动拆分）
我的选择	能过滤就过滤	AQE解决不了时用	默认开启

面试高频追问

追问一：怎么判断一个任务是不是数据倾斜导致的慢？

看Spark UI的Stage详情页。如果某个Stage的Task时间分布极不均匀——比如Median是10秒但Max是2小时——那大概率是倾斜。再看Shuffle Read Size，如果Max远大于Median（比如Max是10GB，Median是50MB），倾斜就确定了。然后找到那个最慢的Task，看它处理的Key是什么，就能定位倾斜的原因。

追问二：GROUP BY也会倾斜吗？不只是JOIN？

会的。GROUP BY的倾斜和JOIN类似——某些key的数据量特别大，导致对应的Reducer任务特别重。解决方案是两阶段聚合：先加随机前缀做局部聚合（第一阶段），再去掉前缀做全局聚合（第二阶段）。AQE对GROUP BY的倾斜也有效，它会自动拆分大分区。

-- GROUP BY两阶段聚合
-- 第一阶段：加盐做局部聚合
WITH stage1 AS (
  SELECT
    CONCAT(country, '_', CAST(FLOOR(RAND()*10) AS STRING)) AS country_salted,
    SUM(amount) AS partial_sum, COUNT(*) AS partial_cnt
  FROM dwd_order_detail WHERE dt = '${bizdate}'
  GROUP BY CONCAT(country, '_', CAST(FLOOR(RAND()*10) AS STRING))
)
-- 第二阶段：去盐做全局聚合
SELECT
  SUBSTRING(country_salted, 1, LENGTH(country_salted)-2) AS country,
  SUM(partial_sum) AS total_amount, SUM(partial_cnt) AS total_cnt
FROM stage1
GROUP BY SUBSTRING(country_salted, 1, LENGTH(country_salted)-2);

追问三：MapJoin能解决倾斜吗？

如果倾斜发生在大表JOIN小表的场景，MapJoin（Broadcast Join）是最好的解决方案——直接把小表广播到所有Executor，大表不需要Shuffle，自然就没有倾斜了。我在出海硬件品牌把autoBroadcastJoinThreshold从默认10MB调到200MB后，很多之前倾斜的JOIN自动变成了Broadcast Join，作业时间减少了40%。但MapJoin有个前提：小表要足够小（一般<200MB），否则广播本身就很慢。

实战踩坑

我在某出海硬件品牌遇到的那个4小时的任务，排查过程是这样的：

第一步，看Spark UI发现Stage 3有200个Task，199个在2分钟内完成，1个跑了3小时47分钟。

第二步，点进那个慢Task，发现Shuffle Read Size是12GB，而其他Task的平均值只有60MB。

第三步，怀疑是某个user_id的数据量特别大，跑了一个分布SQL，发现user_id IS NULL的数据占了15%——大约3000万条全部分到了同一个Task。

第四步，和业务确认后，NULL是未登录的匿名行为，不需要建画像。加了WHERE user_id IS NOT NULL后，任务从4小时降到40分钟。再开启AQE后，进一步降到12分钟。

教训：排查倾斜的标准流程是：看Spark UI → 找到慢Task → 查Shuffle数据分布 → 定位倾斜Key → 选择对应方案。不要一上来就加盐，先搞清楚倾斜的原因。很多时候一个WHERE条件就能解决问题，不需要复杂的加盐逻辑。

总结

画像场景的数据倾斜常见于NULL值聚集和热门用户数据量大。解决三板斧：过滤NULL（最简单）、加盐打散（通用）、AQE自动处理（最推荐）。排查流程固定四步：Spark UI看Task分布→找慢Task→查数据分布→定位根因。面试时能讲清楚排查过程和优化效果（4小时→12分钟），比说「我用了AQE」有说服力得多。

下一篇聊一个画像和数仓结合的话题：用户生命周期怎么定义？从新客到流失的状态机设计。

你们遇到过最严重的数据倾斜是什么场景？怎么解决的？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长