1. 数据倾斜的本质与判断方法1.1 什么是数据倾斜数据倾斜这个词听起来很学术但说白了就一句话本该大家一起干的活全压到了少数几个人身上。在 Hive 里这个“人”就是 task。一个 Hive SQL 提交后会被拆成多个 stage每个 stage 里又有若干 task 并行跑理想情况下每个 task 处理的数据量差不多整个 job 的耗时取决于最慢的那个 task。可一旦某个 key 对应的数据量远超其他 key处理这个 key 的 task 就会变成“拖油瓶”其他 task 早就跑完了它还在那儿慢慢磨整个 job 就被它一个人拖住了。我见过最夸张的一次一个 group by 的 job99 个 task 三分钟跑完剩下 1 个 task 跑了两个多小时最后还 OOM 挂了。这就是典型的数据倾斜。从表现上看数据倾斜通常有这几个信号长尾 task大部分 task 秒级完成个别 task 耗时是平均值的几十倍甚至上百倍Counter 异常在 job 的 Counter 里看到某个 task 的输入记录数、输入字节数远超其他 taskOOM 报错reduce 端某个 task 内存爆掉报java.lang.OutOfMemoryError执行时间波动大同一个 SQL 今天跑 5 分钟明天跑 2 小时数据分布变了倾斜就来了1.2 倾斜的根源key 分布不均数据倾斜的根因永远是key 的分布不均匀。常见场景有这么几类第一类是空值或默认值过多。比如用户表里大量记录的 user_id 是 null 或者 -1join 的时候这些记录全跑到同一个 reduce 上去了。第二类是热点 key。比如电商大促期间某个爆款商品的 id 出现在几千万条订单里按商品 id 聚合的时候这个 key 就是灾难。第三类是数据类型不一致导致的隐式转换。这个坑特别隐蔽join 的两张表关联字段一个是 string 一个是 bigintHive 会做隐式转换转换后可能大量数据落到同一个值上或者干脆转换失败变成 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.aggrtruemap 端会先做一次局部聚合如果某个 map 读到的数据里某个 key 特别多这个 map 的聚合压力就大。2.2 小文件合并从源头减少 map 数小文件是 map 端倾斜的头号元凶。HDFS 上每个小文件都会对应一个 map task如果目录下有几千个小文件就会起几千个 map每个 map 处理一点点数据光启动和调度的开销就够呛。解决办法是在 map 输入前做小文件合并。Hive 提供了几个参数-- 设置 map 输入的最大 split 大小默认 256MB set mapreduce.input.fileinputformat.split.maxsize268435456; -- 设置 map 输入的最小 split 大小 set mapreduce.input.fileinputformat.split.minsize1; -- 每个 node 上 split 的大小影响并发度 set mapreduce.input.fileinputformat.split.size.per.node268435456; -- 每个 rack 上 split 的大小 set mapreduce.input.fileinputformat.split.size.per.rack268435456;这几个参数配合使用可以把多个小文件合并成一个 split减少 map 数。我一般会把maxsize设成 256MB 或 512MB具体看集群的 block size。如果是已经存在的小文件可以用concatenate命令或者重跑一个 insert overwrite 来合并-- 在 Hive 里合并小文件 alter table my_table partition(dt2024-01-01) concatenate;或者在写入时控制-- 设置 reduce 数控制输出文件数 set hive.exec.reducers.bytes.per.reducer268435456; set hive.exec.reducers.max1009;2.3 Map 端聚合参数调优hive.map.aggr这个参数默认是 true意思是 map 端会先做一次聚合把相同的 key 先合并减少传输到 reduce 的数据量。这个功能在大多数情况下是好事但如果某个 key 在单个 map 里就特别多map 端聚合反而会拖慢这个 map。这时候可以调hive.groupby.mapaggr.checkinterval控制 map 端聚合时检查的行数默认 100000。如果超过这个数还没聚合完就会把数据写到磁盘。适当调大可以减少磁盘溢写但会占更多内存。set hive.map.aggrtrue; set hive.groupby.mapaggr.checkinterval200000;还有一个参数hive.map.aggr.hash.percentmemory控制 map 端聚合的 hash 表占用的内存比例默认 0.5。如果 map 端聚合时 OOM可以调小这个值。2.4 实操心得map 端倾斜的排查顺序我处理 map 端倾斜的排查顺序一般是这样的先看输入目录的文件数和文件大小分布hadoop fs -count和hadoop fs -du两个命令就能看出来如果小文件多先合并小文件或者调整 split 参数如果文件大小均匀但 map 还是慢看是不是 map 端聚合导致的试着关掉hive.map.aggr对比一下如果关掉聚合变快了说明是聚合热点那就调聚合相关的参数注意调整 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 值拼上一个随机数让它们分散到不同的 reduceselect 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.reducer536870912; -- reduce 的最大个数 set hive.exec.reducers.max1009; -- 是否自动推断 reduce 数 set hive.exec.dynamic.partition.modenonstrict;如果倾斜不严重只是个别 reduce 稍微慢一点可以调大 reduce 的内存set mapreduce.reduce.memory.mb4096; set mapreduce.reduce.java.opts-Xmx3276m;还可以开启 reduce 端的推测执行让 Hadoop 为慢 task 启动备份任务set mapreduce.reduce.speculativetrue; set mapreduce.map.speculativetrue;不过推测执行对数据倾斜的效果有限因为慢 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 大小表 joinMapJoin 是首选如果 join 的两张表一大一小小表能完全放进内存那 MapJoin 是最优解。MapJoin 的原理是把小表加载到每个 map 的内存里map 读大表数据时直接在内存里做 join不需要 shuffle也就没有 reduce 端的倾斜问题。Hive 默认开启了 MapJoin 的自动转换set hive.auto.convert.jointrue; -- 小表的大小阈值默认 25MB set hive.mapjoin.smalltable.filesize25000000; -- 新版本用这个参数 set hive.auto.convert.join.noconditionaltask.size25000000;如果小表超过阈值可以手动指定 MapJoinselect /* mapjoin(b) */ a.*, b.* from big_table a join small_table b on a.id b.id;MapJoin 的注意事项小表必须能放进内存如果小表实际很大MapJoin 会导致 map OOM小表会被分发到每个 map如果 map 数很多网络传输和内存开销都不小小表里如果有重复 keyMapJoin 会做笛卡尔积结果可能膨胀4.2 大表 join 大表SMB Join 的适用条件如果两张表都很大但都是分桶表可以用 SMB JoinSort 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 Joinset hive.optimize.bucketmapjointrue; set hive.optimize.bucketmapjoin.sortedmergetrue; set hive.input.formatorg.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;SMB Join 的限制比较多两张表必须都是分桶表桶数要成倍数数据要按 join key 排序。实际生产中能满足这些条件的场景不多但一旦满足效果非常好。4.3 大表 join 大表且都有热点 key打散加膨胀如果两张表都很大又都有热点 keySMB 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()*20join 就会丢数据。一定要保证前缀的生成逻辑完全一致。坑三膨胀倍数过大导致数据爆炸。表 B 膨胀 100 倍如果表 B 本身有 1 亿条膨胀后就变成 100 亿条直接把集群干趴。膨胀倍数要根据热点 key 的数据量和集群容量来定一般 10 到 50 之间。坑四忘了处理 null。join 的 key 如果有 nullnull 之间不会 join 上但会全部落到同一个 reduce。如果 null 很多还是要先过滤或打散。5. 通用调优参数与实战速查5.1 倾斜相关的核心参数清单下面这张表是我平时调优时最常用的参数按场景分类整理参数默认值作用建议值hive.map.aggrtruemap 端聚合倾斜时试 falsehive.groupby.skewindatafalsegroup by 倾斜时负载均衡倾斜时设 truehive.optimize.skewjoinfalsejoin 倾斜优化倾斜时设 truehive.skewjoin.key100000判断倾斜 key 的阈值根据数据量调hive.skewjoin.mapjoin.map.tasks10000倾斜 join 的 map 数默认即可hive.exec.reducers.bytes.per.reducer256MB每个 reduce 的数据量调大可减少 reduce 数hive.exec.reducers.max1009reduce 最大个数默认即可mapreduce.reduce.memory.mb1024reduce 内存倾斜时调大mapreduce.map.memory.mb1024map 内存MapJoin 时调大其中hive.groupby.skewindata和hive.optimize.skewjoin是两个“一键倾斜优化”参数原理是 Hive 会自动做两阶段聚合或两阶段 join把倾斜的 key 打散。但这两个参数不是万能的有时候开了反而更慢需要实测。5.2 不同倾斜场景的解决方案对照场景表现首选方案备选方案空值倾斜大量 null 落到同一 reduce过滤 null 或加随机前缀调 reduce 内存热点 key 聚合个别 key 数据量巨大两阶段聚合加随机前缀调hive.groupby.skewindata大小表 join小表 join 大表MapJoin调大 MapJoin 阈值大表 join 大表两表都大且有热点打散加膨胀SMB Join小文件过多map 数过多且不均合并小文件调 split 参数数据类型不一致隐式转换导致倾斜统一数据类型显式 cast5.3 我的调优检查清单每次遇到倾斜我会按这个清单过一遍确认倾斜阶段是 map 端还是 reduce 端是 join 还是 group by确认倾斜 key找出具体是哪些 key 导致的数据量占比多少确认业务需求这些 key 的数据能不能过滤、能不能打散、能不能单独处理选择方案优先用参数解决参数解决不了再改写 SQL验证效果改完后对比 task 耗时分布确认长尾 task 消失记录方案把处理过程和参数记下来下次遇到类似问题直接复用提示调优不是一劳永逸的。数据分布会变今天不倾斜的 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%。第三步跟业务确认测试账号的数据可以过滤掉。于是改写 SQLselect /* 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.size3221225472;改完后表 B 被加载到内存做 MapJoin没有 reduce 阶段整个 job 10 分钟跑完。这个案例的关键点是先过滤掉导致倾斜的数据再考虑用 MapJoin 消除 shuffle。很多时候倾斜的根源就是少数几个异常 key把它们处理掉问题就解决了一大半。5.5 关于倾斜处理的一点个人体会处理数据倾斜这些年我最大的体会是不要一上来就调参数先搞清楚数据长什么样。很多倾斜问题看一眼 key 的分布就明白了是空值、是热点、还是类型不一致对症下药才有效。盲目调参数可能把不倾斜的 job 也调慢了。另外预防比治疗重要。建表的时候做好分桶、控制好小文件、统一关联字段的类型能避免很多后期的倾斜问题。数据开发规范里加上这几条能省下大量调优时间。最后倾斜处理没有银弹。MapJoin、打散、两阶段聚合每种方案都有适用场景和副作用。我的习惯是准备一个“倾斜处理工具箱”遇到问题先从工具箱里挑最合适的试了不行再换。时间长了常见的倾斜场景基本都能快速搞定。