Hive作业跑了6小时还没完——Hive里的数据倾斜和Spark有什么不同？

-- 开启GROUP BY两阶段聚合（解决GROUP BY倾斜）
SET hive.groupby.skewindata = true;
 
-- 生效原理：
-- 第一个MR：先按key+随机数做局部聚合（reduce到所有reducers）
-- 第二个MR：再按真实key做全局聚合
-- 相当于两个MapReduce Job，比一个Job多一次写磁盘/读磁盘
 
-- 注意：这个参数对COUNT(DISTINCT)无效！
-- COUNT(DISTINCT)必须手动改写子查询（和Spark一样）
 
-- 示例：原始有倾斜的GROUP BY
SELECT
  channel_id,
  SUM(order_amount) AS total_amount,
  COUNT(*) AS order_cnt
FROM orders
GROUP BY channel_id;
-- 开启skewindata后，Hive自动加了两阶段处理
-- 但COUNT(DISTINCT)不行：必须手动改

-- COUNT(DISTINCT)必须手动改写：
-- 原始（会倾斜）：
-- SELECT channel_id, COUNT(DISTINCT user_id) AS uv FROM orders GROUP BY channel_id;
 
-- 改写后（子查询去重）：
SELECT channel_id, COUNT(user_id) AS uv
FROM (
  SELECT channel_id, user_id
  FROM orders
  GROUP BY channel_id, user_id  -- 先去重
) t
GROUP BY channel_id;
-- 这个写法在Hive和Spark SQL里都有效

-- 自动MapJoin（类似Spark的autoBroadcastJoinThreshold）
SET hive.auto.convert.join = true;          -- 开启自动MapJoin
SET hive.auto.convert.join.noconditionaltask = true;  -- 无条件转换
SET hive.auto.convert.join.noconditionaltask.size = 10000000;  -- 10MB阈值
-- 小于10MB的表自动转为MapJoin
 
-- 或者手动指定MapJoin（加/*+ MAPJOIN(small_table) */）：
SELECT /*+ MAPJOIN(p) */
  o.order_id,
  o.amount,
  p.product_name
FROM orders o
JOIN products p ON o.product_id = p.product_id;
 
-- Hive MapJoin vs Spark Broadcast Join的区别：
-- Hive：在Map阶段完成，没有Reduce，没有Shuffle
-- Spark：Executor本地Join，没有Shuffle
-- 本质相同，实现层不同

-- 开启JOIN倾斜处理
SET hive.optimize.skewjoin = true;
SET hive.skewjoin.key = 100000;  -- key超过10万行才认为倾斜
 
-- 生效原理：
-- 1. Hive统计每个key的行数（采样统计）
-- 2. 超过hive.skewjoin.key阈值的key被标记为倾斜key
-- 3. 倾斜key单独走一个MapReduce Job处理（两侧数据都缓存到内存/磁盘）
-- 4. 非倾斜key走正常MapReduce Job
-- 5. 最后UNION结果
 
-- 局限性：
-- 对LEFT JOIN的倾斜处理效果有限
-- 统计key行数有采样误差，可能遗漏倾斜key
-- 比手动拆分方案效果差，但不需要改SQL

Hive倾斜参数全景图

参数	解决的问题	生效方式	局限性
hive.groupby.skewindata	GROUP BY聚合倾斜	自动两阶段MR	对COUNT(DISTINCT)无效
hive.auto.convert.join	大小表JOIN倾斜	自动MapJoin广播	小表有大小限制
hive.optimize.skewjoin	JOIN热点key倾斜	热点key单独MR处理	采样统计可能不准
hive.exec.reducers.bytes.per.reducer	整体数据量大导致慢	调整Reducer数量	不针对倾斜，是整体调优
mapred.reduce.tasks	手动设置Reducer数量	直接设定	设太少加剧倾斜

-- 生产环境常用的Hive倾斜处理参数组合：
SET hive.groupby.skewindata = true;           -- 开GROUP BY两阶段
SET hive.auto.convert.join = true;             -- 开自动MapJoin
SET hive.auto.convert.join.noconditionaltask.size = 25000000;  -- 25MB
SET hive.optimize.skewjoin = true;             -- 开JOIN倾斜处理
SET hive.skewjoin.key = 50000;                 -- 5万行以上算倾斜
SET mapred.reduce.tasks = 500;                 -- 增加Reducer数量，分散压力
 
-- 以上参数适合Hive on MapReduce
-- 如果是Hive on Spark，这些参数大多数失效！
-- Hive on Spark要用Spark的参数：
SET spark.sql.adaptive.enabled = true;
SET spark.sql.autoBroadcastJoinThreshold = 209715200;

面试高频追问

追问一：hive.groupby.skewindata开了之后，作业从1个MR变成2个MR，会不会总体更慢？

对于倾斜严重的作业，总体不会更慢，因为倾斜Task本来就是瓶颈，拆成两个MR后两个MR都很快，总时间反而短。但对于数据分布均匀的作业，开这个参数会多一个MR，增加10-20%的开销。所以这个参数不是无脑全局开的，建议只对已知倾斜的作业单独设置，而不是在Hive-site.xml里全局配置。我在早期短视频平台时，就因为DBA把这个参数全局开了，导致大量没倾斜的小作业时间从30秒变成45秒，后来改成按作业单独设置，效果好多了。

追问二：Hive和Spark SQL的COUNT(DISTINCT)倾斜，处理方法有区别吗？

处理方法完全相同——都是手动改写子查询：先GROUP BY (原始group key, distinct key)去重，再COUNT。Hive的hive.groupby.skewindata对COUNT(DISTINCT)无效，Spark的AQE对COUNT(DISTINCT)也无效。这个限制来自SQL语义本身，不是引擎的问题——COUNT(DISTINCT)的结果不满足结合律，无法自动分阶段优化。所以不管用Hive还是Spark，COUNT(DISTINCT)倾斜都得手动改SQL。

追问三：Hive SQL和Spark SQL语法有哪些关键差异，迁移时容易踩什么坑？

我在某出海硬件品牌做EMR到Databricks迁移时，大约20%的HiveSQL需要微调。常见差异：1）NULL处理——Hive的NULL排序默认在最前，Spark默认在最后，ORDER BY xxx ASC NULLS LAST需要显式写；2）窗口函数——Hive对某些窗口函数的帧语义实现和Spark略有不同，特别是ROWS BETWEEN；3）lateral view explode——语法基本相同但部分边界情况不同；4）UDF——Hive的GenericUDF需要重写为Spark的UDF；5）分区剪裁——Hive有些场景需要显式写分区条件，Spark的谓词下推更智能。建议迁移前先跑回归测试，对比结果。

实战踩坑

我在早期的短视频平台工作时，有一个每日跑批的用户标签作业，用Hive on MapReduce处理，在我加入之前已经跑了2年，每天要跑6-8小时。业务反映这个作业越来越慢，从最早的3小时涨到了6小时。

-- 原始作业（简化版），每日跑批统计用户多维度标签
SELECT
  u.user_id,
  COUNT(DISTINCT o.category) AS category_cnt,  -- 购买品类数
  COUNT(DISTINCT o.brand) AS brand_cnt,         -- 购买品牌数
  SUM(o.amount) AS total_spend
FROM user_orders o
JOIN user_base u ON o.user_id = u.user_id
WHERE o.dt BETWEEN '2020-01-01' AND '2020-01-31'
GROUP BY u.user_id;
 
-- 问题1：COUNT(DISTINCT)两个，hive.groupby.skewindata无效
-- 问题2：某些超级用户（刷单/活跃达人）的订单数是普通用户的10000倍
-- 这两个问题叠加，某些Reducer跑了7小时还没完

-- 优化后的SQL（分解COUNT(DISTINCT)，过滤超级用户异常数据）
 
-- 先过滤掉异常订单量的用户（超过5000单/月按5000处理，避免倾斜）
WITH capped_orders AS (
  SELECT user_id, category, brand, amount
  FROM (
    SELECT
      user_id, category, brand, amount,
      ROW_NUMBER() OVER(PARTITION BY user_id ORDER BY order_time) AS rn
    FROM user_orders
    WHERE dt BETWEEN '2020-01-01' AND '2020-01-31'
  ) t
  WHERE rn <= 5000  -- 每个用户最多取5000单，消除超级用户倾斜
),
-- 分解COUNT(DISTINCT)为子查询
category_stats AS (
  SELECT user_id, COUNT(DISTINCT category) AS category_cnt
  FROM (SELECT DISTINCT user_id, category FROM capped_orders) t
  GROUP BY user_id
),
brand_stats AS (
  SELECT user_id, COUNT(DISTINCT brand) AS brand_cnt
  FROM (SELECT DISTINCT user_id, brand FROM capped_orders) t
  GROUP BY user_id
),
spend_stats AS (
  SELECT user_id, SUM(amount) AS total_spend
  FROM capped_orders
  GROUP BY user_id
)
SELECT u.user_id, c.category_cnt, b.brand_cnt, s.total_spend
FROM user_base u
LEFT JOIN /*+ MAPJOIN(c) */ category_stats c ON u.user_id = c.user_id
LEFT JOIN /*+ MAPJOIN(b) */ brand_stats b ON u.user_id = b.user_id
LEFT JOIN /*+ MAPJOIN(s) */ spend_stats s ON u.user_id = s.user_id;
-- 优化后：作业从6小时降到1.5小时

教训：老作业变慢，不一定是代码变差了，更可能是数据分布变化了——新加入的超级用户、异常账号让原来能跑的代码开始倾斜。建议对核心作业建立「数据画像」，定期监控key的分布，发现分布变化及时告警。

总结

Hive倾斜有三个核心参数：skewindata处理GROUP BY、auto.convert.join处理小表JOIN、optimize.skewjoin处理热点key JOIN。原理和Spark相同，但实现是MapReduce的多Stage，每次优化都要额外写磁盘。COUNT(DISTINCT)不管Hive还是Spark都要手动改子查询。Hive on Spark时要切换为Spark参数体系。面试时能区分Hive on MR和Hive on Spark，能说出各参数的生效原理和局限性，就超过了大多数候选人。

下一篇聊倾斜的「治本」方法——数据预处理和分桶优化：从数据写入阶段就消除倾斜的根源，而不是每次查询都处理倾斜。

你们还有在用Hive on MR的作业吗？有没有考虑迁到Spark？最大的顾虑是什么？评论区聊聊～

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