Hive数据倾斜全解析：从Map端到Reduce端的调优实战

1. 数据倾斜的本质与判断方法

1.1 什么是数据倾斜

数据倾斜这个词听起来很学术，但说白了就一句话：本该大家一起干的活，全压到了少数几个人身上。在 Hive 里，这个“人”就是 task。一个 Hive SQL 提交后会被拆成多个 stage，每个 stage 里又有若干 task 并行跑，理想情况下每个 task 处理的数据量差不多，整个 job 的耗时取决于最慢的那个 task。可一旦某个 key 对应的数据量远超其他 key，处理这个 key 的 task 就会变成“拖油瓶”，其他 task 早就跑完了，它还在那儿慢慢磨，整个 job 就被它一个人拖住了。

我见过最夸张的一次，一个 group by 的 job，99 个 task 三分钟跑完，剩下 1 个 task 跑了两个多小时，最后还 OOM 挂了。这就是典型的数据倾斜。

从表现上看，数据倾斜通常有这几个信号：

• 长尾 task：大部分 task 秒级完成，个别 task 耗时是平均值的几十倍甚至上百倍
• Counter 异常：在 job 的 Counter 里看到某个 task 的输入记录数、输入字节数远超其他 task
• OOM 报错：reduce 端某个 task 内存爆掉，报java.lang.OutOfMemoryError
• 执行时间波动大：同一个 SQL 今天跑 5 分钟，明天跑 2 小时，数据分布变了倾斜就来了

1.2 倾斜的根源：key 分布不均

数据倾斜的根因永远是key 的分布不均匀。常见场景有这么几类：

第一类是空值或默认值过多。比如用户表里大量记录的 user_id 是 null 或者 -1，join 的时候这些记录全跑到同一个 reduce 上去了。

第二类是热点 key。比如电商大促期间某个爆款商品的 id 出现在几千万条订单里，按商品 id 聚合的时候这个 key 就是灾难。

第三类是数据类型不一致导致的隐式转换。这个坑特别隐蔽，join 的两张表关联字段一个是 string 一个是 bigint，Hive 会做隐式转换，转换后可能大量数据落到同一个值上，或者干脆转换失败变成 null。

第四类是group by 维度太少。比如按省份聚合，全国就 30 多个省，但广东一个省的数据量可能顶其他所有省加起来，那广东这个 reduce 就惨了。

1.3 怎么快速定位倾斜发生在哪个阶段

定位倾斜不能靠猜，得看证据。我的习惯是分三步走：

第一步，看 job 的 stage 划分。在 YARN 的 Application 页面或者 Hive 的日志里找到 job 的 DAG，看清楚有几个 stage，哪个 stage 慢。通常慢的 stage 就是倾斜发生的 stage。

第二步，看 task 级别的统计。在 job 详情页里按 task 的耗时排序，如果发现个别 task 的耗时、输入数据量、输出数据量是其他 task 的几十倍，基本就锁定了。

第三步，看 Counter。重点看HDFS Read、HDFS Write、Reduce input records、Reduce input groups这几个指标。如果某个 task 的Reduce input records特别大，说明这个 reduce 分到的数据多；如果Reduce input groups特别大，说明这个 reduce 分到的 key 种类多。

提示：定位倾斜时不要只看总耗时，一定要下钻到 task 级别。很多时候 job 总耗时看起来还行，是因为倾斜的 task 被其他 task 的平均值掩盖了，但下次数据量一涨就会暴露。

2. Map 端倾斜的成因与解决套路

2.1 Map 端也会倾斜吗

很多人以为数据倾斜只发生在 reduce 端，其实 map 端一样会倾斜，只是表现形式不同。Map 端的倾斜通常不是 key 分布问题，而是文件大小不均或者小文件太多导致的。

Hive 读数据的时候，一个 split 对应一个 map task。如果目录下有几个超大文件（比如 10GB）和一堆小文件（比如 1MB），那处理大文件的 map task 就会跑很久，处理小文件的 map task 秒完。这就是 map 端的“长尾”。

还有一种情况是map 端做聚合时 key 分布不均。比如开了hive.map.aggr=true，map 端会先做一次局部聚合，如果某个 map 读到的数据里某个 key 特别多，这个 map 的聚合压力就大。

2.2 小文件合并：从源头减少 map 数

小文件是 map 端倾斜的头号元凶。HDFS 上每个小文件都会对应一个 map task，如果目录下有几千个小文件，就会起几千个 map，每个 map 处理一点点数据，光启动和调度的开销就够呛。

解决办法是在 map 输入前做小文件合并。Hive 提供了几个参数：

-- 设置 map 输入的最大 split 大小，默认 256MB set mapreduce.input.fileinputformat.split.maxsize=268435456; -- 设置 map 输入的最小 split 大小 set mapreduce.input.fileinputformat.split.minsize=1; -- 每个 node 上 split 的大小，影响并发度 set mapreduce.input.fileinputformat.split.size.per.node=268435456; -- 每个 rack 上 split 的大小 set mapreduce.input.fileinputformat.split.size.per.rack=268435456;

这几个参数配合使用，可以把多个小文件合并成一个 split，减少 map 数。我一般会把maxsize设成 256MB 或 512MB，具体看集群的 block size。

如果是已经存在的小文件，可以用concatenate命令或者重跑一个 insert overwrite 来合并：

-- 在 Hive 里合并小文件 alter table my_table partition(dt='2024-01-01') concatenate;

或者在写入时控制：

-- 设置 reduce 数，控制输出文件数 set hive.exec.reducers.bytes.per.reducer=268435456; set hive.exec.reducers.max=1009;

2.3 Map 端聚合参数调优

hive.map.aggr这个参数默认是 true，意思是 map 端会先做一次聚合，把相同的 key 先合并，减少传输到 reduce 的数据量。这个功能在大多数情况下是好事，但如果某个 key 在单个 map 里就特别多，map 端聚合反而会拖慢这个 map。

这时候可以调hive.groupby.mapaggr.checkinterval，控制 map 端聚合时检查的行数，默认 100000。如果超过这个数还没聚合完，就会把数据写到磁盘。适当调大可以减少磁盘溢写，但会占更多内存。

set hive.map.aggr=true; set hive.groupby.mapaggr.checkinterval=200000;

还有一个参数hive.map.aggr.hash.percentmemory，控制 map 端聚合的 hash 表占用的内存比例，默认 0.5。如果 map 端聚合时 OOM，可以调小这个值。

2.4 实操心得：map 端倾斜的排查顺序

我处理 map 端倾斜的排查顺序一般是这样的：

1. 先看输入目录的文件数和文件大小分布，hadoop fs -count和hadoop fs -du两个命令就能看出来
2. 如果小文件多，先合并小文件，或者调整 split 参数
3. 如果文件大小均匀但 map 还是慢，看是不是 map 端聚合导致的，试着关掉hive.map.aggr对比一下
4. 如果关掉聚合变快了，说明是聚合热点，那就调聚合相关的参数

注意：调整 split 参数时不要盲目调大，split 太大会导致单个 map 处理时间过长，并发度不够。一般建议 split 大小和 HDFS block size 保持一致，或者略大一点。

3. Reduce 端倾斜的解决手段

3.1 空值引发的倾斜怎么处理

空值导致的倾斜是最常见也最好解决的。比如两张表 join，一张表的关联字段有大量 null，这些 null 在 join 时会被当成同一个 key 分到同一个 reduce，如果 null 的记录有几千万条，这个 reduce 就炸了。

处理办法有两种：

第一种，提前过滤空值。如果业务上 null 的记录本来就不需要，直接在子查询里过滤掉：

select a.*, b.* from ( select * from table_a where user_id is not null ) a join table_b b on a.user_id = b.user_id;

第二种，给空值加随机前缀。如果 null 的记录必须保留（比如要统计 null 的数量），可以给 null 值拼上一个随机数，让它们分散到不同的 reduce：

select a.*, b.* from ( select *, case when user_id is null then concat('null_', cast(rand() * 100 as int)) else user_id end as join_key from table_a ) a join table_b b on a.join_key = b.user_id;

这样 null 值就被打散成null_0到null_99一百个 key，分散到不同 reduce 上。等 join 完了如果需要再聚合，可以再去掉前缀。

3.2 热点 key 的打散方案

热点 key 比空值麻烦，因为热点 key 是有业务含义的，不能随便过滤也不能随便改。常见的打散方案是加随机前缀做两阶段聚合。

假设要按商品 id 聚合，某个爆款商品 id 的数据量特别大。可以这样做：

第一阶段，给商品 id 加上随机前缀，先做一次局部聚合：

select concat(cast(rand() * 10 as int), '_', product_id) as tmp_key, sum(amount) as partial_sum from order_table group by concat(cast(rand() * 10 as int), '_', product_id);

第二阶段，去掉前缀再做一次全局聚合：

select substr(tmp_key, instr(tmp_key, '_') + 1) as product_id, sum(partial_sum) as total_amount from ( select concat(cast(rand() * 10 as int), '_', product_id) as tmp_key, sum(amount) as partial_sum from order_table group by concat(cast(rand() * 10 as int), '_', product_id) ) t group by substr(tmp_key, instr(tmp_key, '_') + 1);

这样原来一个 reduce 要处理的数据被分散到 10 个 reduce 上，每个 reduce 处理十分之一，最后再汇总一次。随机前缀的数量根据热点 key 的数据量来定，一般 10 到 100 之间。

3.3 参数调优：让 reduce 更抗压

除了改写 SQL，调整参数也能缓解 reduce 端倾斜。几个关键参数：

-- 每个 reduce 处理的数据量，默认 256MB，调大可以减少 reduce 数 set hive.exec.reducers.bytes.per.reducer=536870912; -- reduce 的最大个数 set hive.exec.reducers.max=1009; -- 是否自动推断 reduce 数 set hive.exec.dynamic.partition.mode=nonstrict;

如果倾斜不严重，只是个别 reduce 稍微慢一点，可以调大 reduce 的内存：

set mapreduce.reduce.memory.mb=4096; set mapreduce.reduce.java.opts=-Xmx3276m;

还可以开启 reduce 端的推测执行，让 Hadoop 为慢 task 启动备份任务：

set mapreduce.reduce.speculative=true; set mapreduce.map.speculative=true;

不过推测执行对数据倾斜的效果有限，因为慢 task 是真的数据多，备份 task 一样要处理这么多数据，只是可能因为机器负载不同而快一点。

3.4 实操案例：一个 group by 倾斜的完整处理过程

之前处理过一个 case，按用户 id 聚合订单金额，用户表里有几个测试账号，每个账号有几百万条订单，其他正常用户平均只有几十条。跑 group by 的时候，处理这几个测试账号的 reduce 要跑一个多小时。

处理过程是这样的：

第一步，先确认是这几个账号的问题。查了一下订单表，发现user_id in ('test_001', 'test_002', 'test_003')的记录占了总记录的 60%。

第二步，跟业务确认这几个账号的数据要不要保留。业务说测试账号的数据要保留，但可以单独处理。

第三步，改写 SQL，把测试账号的数据单独拎出来处理，正常账号走正常逻辑：

-- 正常用户 select user_id, sum(amount) as total from order_table where user_id not in ('test_001', 'test_002', 'test_003') group by user_id union all -- 测试账号，加随机前缀打散 select user_id, sum(partial) as total from ( select user_id, sum(amount) as partial from order_table where user_id in ('test_001', 'test_002', 'test_003') group by user_id, cast(rand() * 20 as int) ) t group by user_id;

这样测试账号的数据被分散到 20 个 reduce 上，每个 reduce 处理二十分之一，整体耗时从一小时降到了十分钟。

提示：打散随机前缀的数量不是越多越好。前缀太多会导致第二阶段聚合的 key 变多，增加 shuffle 数据量。一般从 10 开始试，不够再加。

4. Join 场景下的倾斜处理

4.1 大小表 join：MapJoin 是首选

如果 join 的两张表一大一小，小表能完全放进内存，那 MapJoin 是最优解。MapJoin 的原理是把小表加载到每个 map 的内存里，map 读大表数据时直接在内存里做 join，不需要 shuffle，也就没有 reduce 端的倾斜问题。

Hive 默认开启了 MapJoin 的自动转换：

set hive.auto.convert.join=true; -- 小表的大小阈值，默认 25MB set hive.mapjoin.smalltable.filesize=25000000; -- 新版本用这个参数 set hive.auto.convert.join.noconditionaltask.size=25000000;

如果小表超过阈值，可以手动指定 MapJoin：

select /*+ mapjoin(b) */ a.*, b.* from big_table a join small_table b on a.id = b.id;

MapJoin 的注意事项：

• 小表必须能放进内存，如果小表实际很大，MapJoin 会导致 map OOM
• 小表会被分发到每个 map，如果 map 数很多，网络传输和内存开销都不小
• 小表里如果有重复 key，MapJoin 会做笛卡尔积，结果可能膨胀

4.2 大表 join 大表：SMB Join 的适用条件

如果两张表都很大，但都是分桶表，可以用 SMB Join（Sort Merge Bucket Join）。SMB Join 要求两张表按 join key 分桶且排序，桶的数量成倍数关系。这样 join 的时候只需要对应的桶做 join，不需要全表 shuffle。

建表时指定分桶：

create table table_a ( id bigint, name string ) clustered by (id) sorted by (id) into 256 buckets; create table table_b ( id bigint, value string ) clustered by (id) sorted by (id) into 256 buckets;

然后开启 SMB Join：

set hive.optimize.bucketmapjoin=true; set hive.optimize.bucketmapjoin.sortedmerge=true; set hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;

SMB Join 的限制比较多：两张表必须都是分桶表，桶数要成倍数，数据要按 join key 排序。实际生产中能满足这些条件的场景不多，但一旦满足，效果非常好。

4.3 大表 join 大表且都有热点 key：打散加膨胀

如果两张表都很大，又都有热点 key，SMB Join 也用不了，那就只能打散。思路是给热点 key 加随机前缀，同时把小表对应的热点 key 膨胀成多份。

假设表 A 有热点 keyhot_001，表 B 也有hot_001。处理步骤：

第一步，把表 B 中hot_001的记录膨胀成 N 份，每份带不同的随机前缀：

select concat(cast(rand() * 10 as int), '_', id) as join_key, value from table_b where id = 'hot_001' union all select id as join_key, value from table_b where id != 'hot_001';

第二步，表 A 中hot_001的记录也加上随机前缀：

select concat(cast(rand() * 10 as int), '_', id) as join_key, name from table_a where id = 'hot_001' union all select id as join_key, name from table_a where id != 'hot_001';

第三步，两张表按join_key做 join。这样hot_001的数据被分散到 10 个 reduce 上，每个 reduce 处理十分之一。

这个方案的关键是膨胀的倍数要一致，表 A 加 10 个前缀，表 B 也要膨胀成 10 份，否则 join 会丢数据。

4.4 实操避坑：join 倾斜处理的几个坑

坑一：MapJoin 的小表判断失误。Hive 判断小表是按表的大小，但有时候表本身不大，经过过滤后更小，Hive 却因为原始表大小超阈值而不走 MapJoin。这时候可以手动加 hint，或者调大阈值。

坑二：打散后数据重复。给两张表都加随机前缀时，如果前缀的生成方式不一致（比如一张表用rand()*10，另一张用rand()*20），join 就会丢数据。一定要保证前缀的生成逻辑完全一致。

坑三：膨胀倍数过大导致数据爆炸。表 B 膨胀 100 倍，如果表 B 本身有 1 亿条，膨胀后就变成 100 亿条，直接把集群干趴。膨胀倍数要根据热点 key 的数据量和集群容量来定，一般 10 到 50 之间。

坑四：忘了处理 null。join 的 key 如果有 null，null 之间不会 join 上，但会全部落到同一个 reduce。如果 null 很多，还是要先过滤或打散。

5. 通用调优参数与实战速查

5.1 倾斜相关的核心参数清单

下面这张表是我平时调优时最常用的参数，按场景分类整理：

其中hive.groupby.skewindata和hive.optimize.skewjoin是两个“一键倾斜优化”参数，原理是 Hive 会自动做两阶段聚合或两阶段 join，把倾斜的 key 打散。但这两个参数不是万能的，有时候开了反而更慢，需要实测。

5.2 不同倾斜场景的解决方案对照

5.3 我的调优检查清单

每次遇到倾斜，我会按这个清单过一遍：

1. 确认倾斜阶段：是 map 端还是 reduce 端，是 join 还是 group by
2. 确认倾斜 key：找出具体是哪些 key 导致的，数据量占比多少
3. 确认业务需求：这些 key 的数据能不能过滤、能不能打散、能不能单独处理
4. 选择方案：优先用参数解决，参数解决不了再改写 SQL
5. 验证效果：改完后对比 task 耗时分布，确认长尾 task 消失
6. 记录方案：把处理过程和参数记下来，下次遇到类似问题直接复用

提示：调优不是一劳永逸的。数据分布会变，今天不倾斜的 SQL 明天可能就倾斜了。建议对重要的 job 做监控，发现耗时异常波动时及时排查。

5.4 一个完整的调优实例

最后分享一个完整的调优实例。有个 job 是两张表 join，表 A 10 亿条，表 B 5000 万条，join key 是 user_id。跑起来之后 reduce 端严重倾斜，个别 reduce 处理了 2 亿条数据，跑了 3 小时还没完。

排查过程：

第一步，看表 B 的大小，5000 万条压缩后大概 2GB，超过 MapJoin 默认阈值 25MB，所以没走 MapJoin。

第二步，看 user_id 的分布，发现表 A 里有 5 个 user_id 是测试账号，每个有几千万条记录，占了表 A 的 30%。

第三步，跟业务确认，测试账号的数据可以过滤掉。于是改写 SQL：

select /*+ mapjoin(b) */ a.*, b.* from ( select * from table_a where user_id not in ('test_001','test_002','test_003','test_004','test_005') ) a join table_b b on a.user_id = b.user_id;

同时调大 MapJoin 阈值：

set hive.auto.convert.join.noconditionaltask.size=3221225472;

改完后，表 B 被加载到内存做 MapJoin，没有 reduce 阶段，整个 job 10 分钟跑完。

这个案例的关键点是：先过滤掉导致倾斜的数据，再考虑用 MapJoin 消除 shuffle。很多时候倾斜的根源就是少数几个异常 key，把它们处理掉，问题就解决了一大半。

5.5 关于倾斜处理的一点个人体会

处理数据倾斜这些年，我最大的体会是：不要一上来就调参数，先搞清楚数据长什么样。很多倾斜问题，看一眼 key 的分布就明白了，是空值、是热点、还是类型不一致，对症下药才有效。盲目调参数可能把不倾斜的 job 也调慢了。

另外，预防比治疗重要。建表的时候做好分桶、控制好小文件、统一关联字段的类型，能避免很多后期的倾斜问题。数据开发规范里加上这几条，能省下大量调优时间。

最后，倾斜处理没有银弹。MapJoin、打散、两阶段聚合，每种方案都有适用场景和副作用。我的习惯是准备一个“倾斜处理工具箱”，遇到问题先从工具箱里挑最合适的，试了不行再换。时间长了，常见的倾斜场景基本都能快速搞定。