数据倾斜问题

数据倾斜是 Hive / MapReduce 场景下最常见的性能问题之一。

一、数据倾斜的表现

• reduce 阶段卡在 99.99% 一直不动；
• 各种 container 报错 OOM（内存溢出）；
• 某些 task 读写数据量很大，远超其他正常的 reduce。

二、数据倾斜的原理

在进行 shuffle 时，必须将各个节点上相同的 key 拉取到某个节点上的一个 task 来处理（比如按 key 聚合或 join）。如果某个 key 对应的数据量特别大，就会发生数据倾斜。

数据倾斜的根本原因在于 Task 的数据分配不均衡，分两种情况：

1. 数据本身倾斜：数据中某种数据出现的次数过多；
2. 分区规则导致：相同的数据被分配给了同一个 Task，导致该 Task 拿到大量数据，而其他 Task 数据量少，运行较慢。

三、数据倾斜的处理

1. group by 产生数据倾斜

解决方法：

• 开启 Map 端聚合参数设置；
• 或根据业务，合理调整分组维度。

2. count(distinct) 产生数据倾斜

解决方法：

• 使用 sum ... group by 代替：

  select a, sum(1)
  from (select a, b from t group by a, b)
  group by a;

• 在业务逻辑优化效果不大的情况下，可将倾斜的数据单独拿出来处理，最后 union 回去。

3. 大表和小表 join 产生数据倾斜

解决方法：使用 mapjoin，将小表加载到内存中。

4. 空值产生数据倾斜

解决方法一：id 为空的不参与关联。

select * from log a join users b
on a.user_id is not null and a.user_id = b.user_id
union all
select * from log a where a.user_id is null;

解决方法二：给空值分配随机的 key 值（一般更推荐）。

select * from log a left outer join users b
on case when a.user_id is null then concat('hive', rand()) else a.user_id end = b.user_id;

5. 合理设置 map 和 reduce 数避免数据倾斜

调整 Map 数：

• 小文件合并，减少 map 数（默认开启）。在 map 执行前合并小文件：

  set hive.input.format = org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

  CombineHiveInputFormat 具有小文件合并功能（系统默认格式），而 HiveInputFormat 没有。

• 复杂文件增加 Map 数：当 input 文件很大、任务逻辑复杂、map 执行很慢时，可增加 map 数，使每个 map 处理的数据量减少，提高执行效率。

  MapReduce 无法直接控制 map 数量，只能通过设置每个 map 处理的数据量来间接控制：

  -- HDFS 默认数据块 128M、最小分片 1、最大分片 256M
  set dfs.blocksize = 134217728;
  set mapreduce.input.fileinputformat.split.minsize = 1;
  set mapreduce.input.fileinputformat.split.maxsize = 256000000;

  分片大小计算：computeSplitSize(Math.max(minSize, Math.min(maxSize, blocksize))) = blocksize = 128M。让 maxsize 低于 blocksize 即可增加 map 个数。

调整 Reduce 数：

方法一（参数方式）：

-- 每个 Reduce 处理的数据量默认 256MB
hive.exec.reducers.bytes.per.reducer = 256000000
-- 每个任务最大的 reduce 数，默认 1009
hive.exec.reducers.max = 1009

reducer 数计算公式：

N = min(hive.exec.reducers.max, 总输入数据量 / hive.exec.reducers.bytes.per.reducer)

方法二（配置文件方式）：在 mapred-default.xml 中修改每个 job 的 Reduce 个数。

set mapreduce.job.reduces = 15;

附：原笔记 PDF

下载原始 PDF