1. 数据倾斜的本质与判断方法
1.1 什么是数据倾斜，为什么它总在大数据任务里阴魂不散
数据倾斜说白了就是数据分布不均匀，导致某个或某几个计算节点干了绝大部分的活，其他节点早早完工在旁边干瞪眼。这种情况在Hive里特别常见，因为Hive底层跑的是MapReduce或Tez/Spark引擎，只要涉及到shuffle操作，key的分布就直接决定了每个reducer的负载。
我举个生活化的例子：一个小组十个人搬砖，本来应该每人搬100块，结果有一个人分到了800块，其他九个人加起来才200块。那个搬800块的人累死也搬不完，其他人早就下班了。这就是数据倾斜——不是总量大，而是分配不均。
在Hive中，数据倾斜通常发生在三个环节：Map端读取、Reduce端聚合、Join关联。每个环节的成因和表现都不一样，解决手段也完全不同。很多人一遇到任务跑得慢就盲目加资源，结果内存翻倍了任务还是卡在99%不动，这就是没找到根因。
1.2 怎么快速判断任务是否发生了数据倾斜
判断数据倾斜有几个非常实用的信号，我在实际工作中主要看这几个：
- 任务卡在reduce阶段99%或100%长时间不动：这是最典型的症状。因为大部分reducer已经跑完了，就剩一两个还在处理超大key。
- Counter中某个reducer的输入记录数远超其他：在Job History或者YARN的日志里能看到每个reduce task处理的记录数，如果最大值和平均值差了几十倍甚至上百倍，基本可以确诊。
- Map阶段就出现长尾：如果Map端读取时就倾斜，通常是因为文件大小极不均匀，比如某些文件几百MB，某些只有几KB。
- OOM频繁出现在某个特定task：某个key对应的数据量太大，reduce端内存扛不住直接崩了。
实操心得：我一般会先跑一遍任务，然后去YARN的ResourceManager页面看每个task的duration和输入数据量。如果发现某个task的处理时间是其他的10倍以上，不用犹豫，就是倾斜了。
1.3 倾斜的根因分类：key分布不均、文件大小不均、null值集中
数据倾斜的根因大致可以归为三类：
第一类：key分布严重不均。比如用户行为日志里，某些热门商品或热门页面的访问量是普通商品的几万倍。做group by或者join的时候，这些热点key全部涌向同一个reducer。
第二类：文件大小严重不均。HDFS上某些文件特别大，某些特别小。Map任务按文件切分，大文件的map task自然跑得慢。这种情况在日志采集不规律或者上游任务输出不均匀时特别常见。
第三类：null值或默认值集中。这个坑很多人踩过——join的关联键里有大量null，或者group by的字段有大量空值。null在Hive里会被当成一个特殊的key，所有的null数据全部跑到同一个reducer，瞬间把那个节点压垮。
理解了这三类根因，后面的解决方案就有了方向。接下来我按Map端、Reduce端、Join三个场景分别展开，把每种场景下的解决手段讲透。
2. Map端数据倾斜的解决手段
2.1 小文件合并：从源头减少Map任务的不均衡
Map端倾斜最常见的原因就是输入文件大小不均。HDFS上如果有大量小文件，每个小文件至少对应一个Map任务，启动开销大不说，如果某些文件特别大而某些特别小，Map阶段的耗时就完全不可控。
解决办法是在Map读取之前做小文件合并。Hive提供了几个参数来控制：
-- 在Map端输入时合并小文件 SET hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; -- 每个Map处理的最大输入量（字节），默认256MB SET mapred.max.split.size=256000000; -- 一个节点上split的最小大小 SET mapred.min.split.size.per.node=100000000; -- 一个机架上split的最小大小 SET mapred.min.split.size.per.rack=100000000;这几个参数配合使用的逻辑是：CombineHiveInputFormat会把多个小文件合并成一个split，合并的上限由mapred.max.split.size控制。如果小文件加起来还不够mapred.min.split.size.per.node，就继续合并。
注意事项：
hive.input.format这个参数在较新版本的Hive中默认就是CombineHiveInputFormat，但如果你用的是老版本或者某些发行版，需要手动确认。另外，合并小文件会增加单个Map任务的输入量，如果Map端逻辑很重（比如复杂的UDF），合并太多可能导致单个Map跑太久，需要权衡。
2.2 文件大小不均的预处理：提前做一次均衡
如果上游任务输出的文件大小差异极大，光靠参数调优可能不够。我通常会在ETL流程里加一个均衡步骤，把数据重新分布一遍。
具体做法是：创建一个中间表，用distribute by rand()或者distribute by cast(rand()*N as int)把数据随机打散，然后重新写入。这样每个文件的大小就基本均匀了。
-- 把倾斜的数据重新均衡写入中间表 INSERT OVERWRITE TABLE mid_table PARTITION(dt='2024-01-01') SELECT /*+ REPARTITION(100) */ col1, col2, col3 FROM source_table DISTRIBUTE BY CAST(RAND() * 100 AS INT);这里REPARTITION(100)是Spark引擎的写法，如果是MR引擎，用DISTRIBUTE BY就够了。数字100是根据数据总量和目标文件大小算出来的——假设数据总量100GB，希望每个文件1GB左右，那就分成100份。
实操心得：这个预处理步骤会多跑一个Job，看起来浪费了时间，但实际上它能让后续所有依赖这张表的任务都受益。我在一个日均处理TB级数据的项目里加了这个步骤后，下游任务的稳定运行时间从平均2小时降到了40分钟。
2.3 Map端聚合：把combiner用起来
Map端的另一个优化点是开启Map端聚合。当你的SQL里有group by或者count distinct这类聚合操作时，在Map端先做一次局部聚合，可以大幅减少shuffle到Reduce端的数据量。
-- 开启Map端聚合 SET hive.map.aggr=true; -- 控制Map端聚合的内存使用 SET hive.map.aggr.hash.percentmemory=0.5; -- 聚合的条目数上限 SET hive.groupby.mapaggr.checkinterval=100000;hive.map.aggr=true会让Hive在Map阶段就做一次类似Combiner的操作。比如你要统计每个key的count，Map端可以先在本地把相同key的计数加起来，只把局部结果传给Reduce，这样网络传输和Reduce处理的数据量都能大幅减少。
但这里有个坑：如果group by的key本身就很少（比如只有几十个），Map端聚合反而会消耗大量内存去维护哈希表，甚至OOM。所以hive.map.aggr.hash.percentmemory这个参数要控制好，一般设0.5左右比较稳妥。
常见问题：有人开了
hive.map.aggr=true之后发现任务反而更慢了，原因就是key太少，Map端维护哈希表的开销超过了收益。这种情况下应该关掉Map端聚合，或者改用hive.groupby.skewindata参数来处理。
3. Reduce端数据倾斜的解决手段
3.1 负载均衡：hive.groupby.skewindata的妙用
Reduce端倾斜最经典的解决方案就是开启负载均衡。Hive提供了一个参数hive.groupby.skewindata，开启后会启动两个MR Job：
SET hive.groupby.skewindata=true;第一个Job会先把数据随机分发到各个Reducer做预聚合，第二个Job再按照真正的key做最终聚合。这样即使某个key的数据量特别大，在第一个Job里也被随机打散了，不会集中到一个Reducer上。
这个方案的原理其实很简单：用一次额外的shuffle换取负载均衡。第一个Job的Reducer输出结果会按照key重新分区，但因为已经做过预聚合，数据量已经小了很多，第二个Job的Reducer压力就小多了。
注意事项：这个参数只对group by有效，对join无效。而且它会多启动一轮MR，对于本来就不倾斜的任务来说反而增加了开销。所以我的建议是——只在确认有倾斜的时候才开，不要默认开启。
3.2 两阶段聚合：手动实现更精细的控制
如果hive.groupby.skewindata的效果不够理想，或者你需要更精细的控制，可以手动实现两阶段聚合。思路和上面类似，但你可以自己控制打散的粒度和聚合的逻辑。
-- 第一阶段：加随机前缀打散 CREATE TABLE tmp_stage1 AS SELECT CONCAT(CAST(CAST(RAND()*10 AS INT) AS STRING), '_', user_id) AS tmp_key, user_id, amount FROM source_table; -- 第二阶段：去掉前缀做最终聚合 SELECT SPLIT(tmp_key, '_')[1] AS user_id, SUM(amount) AS total_amount FROM tmp_stage1 GROUP BY SPLIT(tmp_key, '_')[1];这个方案的好处是完全可控。你可以根据数据量调整随机前缀的范围——数据量越大，前缀范围越大，打散效果越好。缺点是写起来麻烦，需要多建一张中间表。
我在处理一个日增数据量超过500GB的任务时用过这个方案，把原来需要3小时的聚合操作压缩到了40分钟。关键就是把随机前缀设成了20，让每个热点key被分散到20个Reducer上。
3.3 参数调优：reduce个数与内存的合理配置
Reduce端倾斜有时候不是数据分布的问题，而是Reducer数量设置不合理。默认情况下Hive会根据输入数据量自动推算Reducer个数，但这个推算不一定准确。
-- 手动指定Reducer个数 SET mapred.reduce.tasks=200; -- 或者让Hive自动推算，但控制上下限 SET hive.exec.reducers.bytes.per.reducer=256000000; SET hive.exec.reducers.max=500;hive.exec.reducers.bytes.per.reducer控制每个Reducer处理的数据量，默认256MB。如果你的数据总量是100GB，那就会启动约400个Reducer。但如果数据倾斜严重，即使Reducer数量够多，热点key还是会把某个Reducer压垮。
这时候可以考虑增加Reducer的内存：
SET mapred.reduce.child.java.opts=-Xmx4096m;把Reducer的堆内存从默认的1GB或2GB调到4GB甚至8GB，让单个Reducer能扛住更大的key。但这只是治标不治本，根本解决还是要靠打散。
实操心得：我一般会先看数据总量和key的分布情况，如果热点key的数据量是平均值的100倍以上，光加内存是没用的，必须打散。如果只是10倍左右，加内存配合适度的Reducer数量调整就能解决。
4. Join场景下的数据倾斜解决
4.1 Map Join：小表驱动大表的最佳实践
Join倾斜最有效的解决方案是Map Join。当join的两张表中有一张足够小（默认25MB以下），Hive可以把小表加载到每个Map任务的内存里，直接在Map端完成join，完全避免shuffle。
-- 开启Map Join自动转换 SET hive.auto.convert.join=true; -- 小表的大小阈值，默认25MB SET hive.mapjoin.smalltable.filesize=25000000; -- 新版本用这个参数 SET hive.auto.convert.join.noconditionaltask.size=25000000;Map Join的原理是：小表被分发到每个Map任务，Map任务读取大表的split时，直接用内存中的小表做关联。这样就没有Reduce阶段了，自然也不存在Reduce端倾斜。
但这里有个限制——小表必须真的小。如果小表有几百MB甚至几GB，分发到每个Map任务会消耗大量内存和网络带宽，反而拖慢任务。我一般把阈值控制在100MB以内，超过这个大小就考虑其他方案。
常见问题：有人把小表阈值调到500MB，结果任务跑得比以前还慢。原因是小表分发到几百个Map任务，每个都要加载500MB数据，网络传输和内存开销巨大。所以这个阈值不能盲目调大。
4.2 大表Join大表：SMB Join与分桶优化
如果两张表都很大，Map Join用不了，那就需要考虑SMB Join（Sort-Merge-Bucket Join）。这个方案要求两张表都按照join key分桶且排序，然后Hive可以按桶做Map端join。
-- 建表时指定分桶 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.auto.convert.sortmerge.join=true; SET hive.optimize.bucketmapjoin=true; SET hive.optimize.bucketmapjoin.sortedmerge=true;SMB Join的核心思想是：桶对桶关联。table_a的第1个桶只和table_b的第1个桶做join，第2个桶只和第2个桶join，以此类推。这样每个Map任务只需要处理对应桶的数据，数据量大幅减少。
但这个方案的前提是两张表的分桶数和分桶字段完全一致，而且数据必须已经排好序。如果表没有分桶，需要先重建表并导入数据，成本比较高。我一般只在核心的、频繁join的大表上做这个优化。
4.3 空值引发的倾斜：null值处理与随机化
Join倾斜里有一个特别隐蔽的坑——null值集中。当join key里有大量null时，这些null会被当成同一个key，全部涌向一个Reducer。
解决办法很简单：给null值加随机前缀，让它们分散到不同的Reducer。
-- 处理左表的null值 SELECT a.id, b.name FROM ( SELECT CASE WHEN id IS NULL THEN CONCAT('null_', CAST(RAND()*100 AS INT)) ELSE id END AS id, other_col FROM table_a ) a LEFT JOIN table_b b ON a.id = b.id;这样null值就被随机打散成了null_0到null_99，分散到100个Reducer上。因为null值本来就不参与实际的业务关联，打散后不会影响最终结果。
注意事项：这个方法只适用于左表null值不需要参与关联的场景。如果null值有业务含义，比如表示"未知用户"，那打散后需要确保业务逻辑能正确处理。另外，右表的null值不需要处理，因为join时右表的null本来就不会匹配到任何左表记录。
4.4 热点key单独处理：从数据层面拆分
如果某个热点key的数据量实在太大，上面这些方法都搞不定，那就只能从数据层面把热点key拆出来单独处理。
具体做法是：先把热点key的数据单独抽出来，做一次专门的join，然后把结果和其余数据的join结果union起来。
-- 热点key单独处理 CREATE TABLE hot_result AS SELECT a.id, a.col1, b.col2 FROM table_a a JOIN table_b b ON a.id = b.id WHERE a.id = 'hot_key_001'; -- 非热点key正常处理 CREATE TABLE normal_result AS SELECT a.id, a.col1, b.col2 FROM table_a a JOIN table_b b ON a.id = b.id WHERE a.id != 'hot_key_001'; -- 合并结果 INSERT OVERWRITE TABLE final_result SELECT * FROM hot_result UNION ALL SELECT * FROM normal_result;这个方案看起来笨，但效果最确定。热点key单独跑一个任务，可以给它分配更多的资源；非热点key的数据分布均匀，正常跑就行。我在一个电商项目中用这个方法处理过几个超级热门商品的关联查询，效果立竿见影。
实操心得：热点key的识别需要先跑一次统计，看看哪些key的记录数远超平均值。我一般会先跑一个
SELECT id, COUNT(*) FROM table GROUP BY id ORDER BY COUNT(*) DESC LIMIT 20，把Top 20的key拿出来看看。如果Top 1的key占了总数据量的10%以上，就值得单独处理。
5. 通用调优参数与实战排查清单
5.1 一套可以直接抄的倾斜调优参数模板
经过多个项目的积累，我整理了一套通用的倾斜调优参数模板。这些参数不是万能的，但覆盖了80%的倾斜场景，可以作为调优的起点。
-- Map端优化 SET hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; SET mapred.max.split.size=256000000; SET mapred.min.split.size.per.node=100000000; SET mapred.min.split.size.per.rack=100000000; -- Map端聚合 SET hive.map.aggr=true; SET hive.map.aggr.hash.percentmemory=0.5; -- Reduce端负载均衡 SET hive.groupby.skewindata=true; -- Join优化 SET hive.auto.convert.join=true; SET hive.mapjoin.smalltable.filesize=25000000; SET hive.auto.convert.sortmerge.join=true; SET hive.optimize.bucketmapjoin=true; SET hive.optimize.bucketmapjoin.sortedmerge=true; -- Reduce个数控制 SET hive.exec.reducers.bytes.per.reducer=256000000; SET hive.exec.reducers.max=500; -- 内存控制 SET mapred.reduce.child.java.opts=-Xmx4096m; SET mapred.map.child.java.opts=-Xmx2048m;这套参数的使用逻辑是：先开Map端聚合和负载均衡，再看Join是否能用Map Join，最后调Reducer数量和内存。不要一次性全开，而是根据具体任务的症状逐步调整。
注意事项：
hive.groupby.skewindata=true和hive.map.aggr=true同时开启时，Hive会优先使用skewindata的两阶段聚合，map.aggr的效果会被覆盖。所以如果你的任务主要是group by倾斜，开skewindata就够了；如果是join倾斜，重点在Map Join和SMB Join上。
5.2 排查倾斜的完整操作流程
遇到任务跑得慢，我一般按这个流程排查：
第一步：确认是否真的倾斜。去YARN页面看每个task的duration和输入数据量，如果最大值和平均值差距在10倍以内，可能只是资源不足，不是倾斜。
第二步：定位倾斜发生的阶段。看任务卡在Map阶段还是Reduce阶段。Map阶段卡住通常是文件大小不均；Reduce阶段卡住通常是key分布不均。
第三步：分析key的分布。对疑似倾斜的key做一次count统计，看看Top key的占比。如果某个key占了总数据量的5%以上，基本可以确认是热点key。
第四步：选择合适的解决方案。Map端倾斜用文件合并和均衡预处理；Reduce端group by倾斜用skewindata或两阶段聚合；Join倾斜用Map Join、SMB Join或热点key拆分。
第五步：验证效果。调优后重新跑任务，对比调优前后的运行时间和资源消耗。如果效果不明显，回到第三步重新分析。
5.3 常见问题速查表
| 问题现象 | 可能原因 | 解决方案 |
|---|---|---|
| 任务卡在Reduce 99% | 热点key导致单个Reducer处理慢 | 开启skewindata或两阶段聚合 |
| Map阶段长尾严重 | 输入文件大小不均 | 合并小文件或做均衡预处理 |
| Join任务OOM | 大表关联导致单个Reducer内存不足 | 改用Map Join或SMB Join |
| null值导致倾斜 | join key或group by字段有大量null | 给null加随机前缀打散 |
| 开了skewindata反而更慢 | 任务本身不倾斜，多了一轮MR开销 | 关闭skewindata，检查是否真的倾斜 |
| Map Join分发小表超时 | 小表阈值设得太大 | 降低hive.mapjoin.smalltable.filesize |
| 分桶表join仍然慢 | 分桶数不一致或数据未排序 | 检查分桶定义，重建表 |
5.4 几个我踩过的坑和对应的经验
坑一：盲目调大Reducer内存。有一次任务OOM，我直接把Reducer内存从2GB调到8GB，结果任务还是崩。后来发现是热点key的数据量太大，单个Reducer处理的数据量超过了内存上限，加内存只是延缓了崩溃时间。最后用两阶段聚合才彻底解决。
坑二：skewindata和map.aggr同时开。这两个参数同时开启时，Hive的行为可能不符合预期。我建议只开其中一个，根据倾斜类型选择。
坑三：Map Join的小表阈值调得太大。小表阈值调到200MB后，每个Map任务都要加载200MB数据到内存，几百个Map任务同时跑，网络带宽直接被占满。后来把阈值降回50MB，问题解决。
坑四：忽略了数据预处理。很多倾斜问题其实可以在ETL阶段就避免。比如上游任务输出文件时用distribute by rand()打散，下游任务就不会遇到文件大小不均的问题。预防比治疗更重要。
坑五：没有监控和告警。倾斜问题往往是突发的——某天某个key突然爆量，任务就跑不动了。我后来在关键任务上加了监控，每天统计Top key的分布，一旦发现某个key的占比超过阈值就提前告警，在问题爆发前处理。
5.5 不同引擎下的倾斜处理差异
Hive底层可以跑MR、Tez、Spark三种引擎，不同引擎下倾斜的表现和处理方式略有差异。
MR引擎：最稳定但最慢。倾斜问题最明显，因为MR的shuffle机制决定了热点key必然集中到一个Reducer。上面的参数调优主要针对MR引擎。
Tez引擎：比MR快很多，支持DAG优化。Tez下hive.groupby.skewindata的效果更好，因为它可以在一个DAG里完成两阶段聚合，不需要启动两个独立的MR Job。
Spark引擎：最快但内存管理更复杂。Spark下可以用REPARTITION和DISTRIBUTE BY更灵活地控制数据分布。另外Spark的AQE（自适应查询执行）可以自动处理部分倾斜问题，但效果有限，复杂倾斜还是需要手动干预。
实操心得：如果集群资源允许，我建议优先用Tez或Spark引擎，它们的倾斜处理能力比MR强很多。但参数调优的逻辑是相通的——理解倾斜的本质，才能在不同引擎下灵活应对。
数据倾斜这个问题，说到底就是分布不均四个字。Map端看文件大小，Reduce端看key分布，Join看关联键的集中度。把这三个场景的根因搞清楚，再配合参数调优和数据预处理，90%以上的倾斜问题都能解决。剩下10%的极端情况，就用热点key拆分这种"笨办法"兜底。我在实际工作中最大的体会是——不要等到任务跑不动了才去处理倾斜，而是在数据建模和ETL设计阶段就把分布均匀作为一条硬性标准。这样后面能省下大量的调优时间。
