Hive 数据倾斜&优化

大数据 评论

数据倾斜的原因

注意的SQL操作

关键词情形后果
Join其中一个表较小，
但是key集中		分发到某一个或几个Reduce上的数据远高于平均值
大表与大表，但是分桶的判断字段0值或空值过多这些空值都由一个reduce处理，灰常慢
group bygroup by 维度过小，
某值的数量过多		处理某值的reduce灰常耗时
Count Distinct某特殊值过多处理此特殊值的reduce耗时

原因

1)、key分布不均匀
2)、业务数据本身的特性
3)、建表时考虑不周
4)、某些SQL语句本身就有数据倾斜

表现

任务进度长时间维持在99%（或100%），查看任务监控页面，发现只有少量（1个或几个）reduce子任务未完成。因为其处理的数据量和其他reduce差异过大。


单一reduce的记录数与平均记录数差异过大，通常可能达到3倍甚至更多。 最长时长远大于平均时长。

数据倾斜的解决方案

参数调节

1
hive.map.aggr=true


Map 端部分聚合，相当于Combiner

1
hive.groupby.skewindata=true


有数据倾斜的时候进行负载均衡，当选项设定为 true，生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；


第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。

SQL语句调节

如何Join
关于驱动表的选取，选用join key分布最均匀的表作为驱动表
做好列裁剪和filter操作，以达到两表做join的时候，数据量相对变小的效果。

大小表Join
使用map join让小的维度表（1000条以下的记录条数） 先进内存。在map端完成reduce.

大表Join大表
把空值的key变成一个字符串加上随机数，把倾斜的数据分到不同的reduce上，由于null值关联不上，处理后并不影响最终结果。

count distinct大量相同特殊值
count distinct时，将值为空的情况单独处理，如果是计算count distinct，可以不用处理，直接过滤，在最后结果中加1。如果还有其他计算，需要进行group by，可以先将值为空的记录单独处理，再和其他计算结果进行union。

group by维度过小
采用sum() group by的方式来替换count(distinct)完成计算。

特殊情况特殊处理**
在业务逻辑优化效果的不大情况下，有些时候是可以将倾斜的数据单独拿出来处理。最后union回去。

典型的业务场景

空值产生的数据倾斜

场景：如日志中，常会有信息丢失的问题，比如日志中的 user_id，如果取其中的 user_id 和 用户表中的user_id 关联，会碰到数据倾斜的问题。

解决方法1：** user_id为空的不参与关联（红色字体为修改后）

1
2
3
4
5
6
7
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;

解决方法2 ：赋与空值分新的key值

1
2
3
4
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;

结论：方法2比方法1效率更好，不但io少了，而且作业数也少了。解决方法1中 log读取两次，jobs是2。解决方法2 job数是1 。这个优化适合无效 id (比如 -99 , ’’, null 等) 产生的倾斜问题。把空值的 key 变成一个字符串加上随机数，就能把倾斜的数据分到不同的reduce上 ,解决数据倾斜问题。

**

不同数据类型关联产生数据倾斜

场景：用户表中user_id字段为int，log表中user_id字段既有string类型也有int类型。当按照user_id进行两个表的Join操作时，默认的Hash操作会按int型的id来进行分配，这样会导致所有string类型id的记录都分配到一个Reducer中。
解决方法：把数字类型转换成字符串类型

1
2
3
select * from users a
  left outer join logs b
  on a.usr_id = cast(b.user_id as string)

**

小表不小不大，怎么用 map join 解决倾斜问题

使用 map join 解决小表(记录数少)关联大表的数据倾斜问题，这个方法使用的频率非常高，但如果小表很大，大到map join会出现bug或异常，这时就需要特别的处理。 以下例子:

1
2
3
select * from log a
  left outer join users b
  on a.user_id = b.user_id;

users 表有 600w+ 的记录，把 users 分发到所有的 map 上也是个不小的开销，而且 map join 不支持这么大的小表。如果用普通的 join，又会碰到数据倾斜的问题。
解决方法：

1
2
3
4
5
6
7
8
select /*+mapjoin(x)*/* from log a
  left outer join (
    select  /*+mapjoin(c)*/d.*
      from ( select distinct user_id from log ) c
      join users d
      on c.user_id = d.user_id
    ) x
  on a.user_id = b.user_id;

假如，log里user_id有上百万个，这就又回到原来map join问题。所幸，每日的会员uv不会太多，有交易的会员不会太多，有点击的会员不会太多，有佣金的会员不会太多等等。所以这个方法能解决很多场景下的数据倾斜问题。

GROUP BY替代COUNT(DISTINCT)达到优化效果

计算 uv 的时候，经常会用到 COUNT(DISTINCT)，但在数据比较倾斜的时候 COUNT(DISTINCT) 会比较慢。这时可以尝试用 GROUP BY 改写代码计算 uv。

1
2
INSERT OVERWRITE TABLE s_dw_tanx_adzone_uv PARTITION (ds=20120329) 
SELECT 20120329 AS thedate,adzoneid,COUNT(DISTINCT acookie) AS uv FROM s_ods_log_tanx_pv t WHERE t.ds=20120329 GROUP BY adzoneid

关于COUNT(DISTINCT)的数据倾斜问题不能一概而论，要依情况而定，下面是我测试的一组数据：
测试数据：169857条


*统计每日IP *

1
2
CREATE TABLE ip_2014_12_29 AS SELECT COUNT(DISTINCT ip) AS IP FROM logdfs WHERE logdate='2014_12_29'; 
耗时：24.805 seconds


*统计每日IP（改造） *

1
2
CREATE TABLE ip_2014_12_29 AS SELECT COUNT(1) AS IP FROM (SELECT DISTINCT ip from logdfs WHERE logdate='2014_12_29') tmp; 
耗时：46.833 seconds

测试结果表名：明显改造后的语句比之前耗时，这是因为改造后的语句有2个SELECT，多了一个job，这样在数据量小的时候，数据不会存在倾斜问题。

解决Hive对UNION ALL优化的短板

Hive 对 union all 的优化的特性：对 union all 优化只局限于非嵌套查询。

  • 消灭子查询内的 group by

    示例 1：子查询内有 group by

    1
    2
    3
    SELECT * FROM 
    (SELECT * FROM t1 GROUP BY c1,c2,c3 UNION ALL SELECT * FROM t2 GROUP BY c1,c2,c3)t3 
    GROUP BY c1,c2,c3

    从业务逻辑上说，子查询内的 GROUP BY 怎么都看显得多余（功能上的多余，除非有 COUNT(DISTINCT)），如果不是因为 Hive Bug 或者性能上的考量（曾经出现如果不执行子查询 GROUP BY，数据得不到正确的结果的 Hive Bug）。所以这个 Hive 按经验转换成如下所示：

    1
    SELECT * FROM (SELECT * FROM t1 UNION ALL SELECT * FROM t2)t3 GROUP BY c1,c2,c3

    调优结果：经过测试，并未出现 union all 的 Hive Bug，数据是一致的。MapReduce 的 作业数由 3 减少到 1。
    t1 相当于一个目录，t2 相当于一个目录，对 Map/Reduce 程序来说，t1，t2 可以作为 Map/Reduce 作业的 mutli inputs。这可以通过一个 Map/Reduce 来解决这个问题。Hadoop 的 计算框架，不怕数据多，就怕作业数多。
    但如果换成是其他计算平台如 Oracle，那就不一定了，因为把大的输入拆成两个输入， 分别排序汇总后 merge（假如两个子排序是并行的话），是有可能性能更优的（比如希尔排 序比冒泡排序的性能更优）。

  • 消灭子查询内的 COUNT(DISTINCT)，MAX，MIN

    1
    2
    3
    4
    SELECT * FROM 
    (SELECT * FROM t1 
    UNION ALL SELECT c1,c2,c3 COUNT(DISTINCT c4) FROM t2 GROUP BY c1,c2,c3) t3 
    GROUP BY c1,c2,c3;

    由于子查询里头有 COUNT(DISTINCT)操作，直接去 GROUP BY 将达不到业务目标。这时采用 临时表消灭 COUNT(DISTINCT)作业不但能解决倾斜问题，还能有效减少 jobs。

    1
    2
    3
    4
    5
    6
    INSERT t4 SELECT c1,c2,c3,c4 FROM t2 GROUP BY c1,c2,c3; 
    SELECT c1,c2,c3,SUM(income),SUM(uv) FROM 
    (SELECT c1,c2,c3,income,0 AS uv FROM t1 
    UNION ALL 
    SELECT c1,c2,c3,0 AS income,1 AS uv FROM t2) t3 
    GROUP BY c1,c2,c3;

    job 数是 2，减少一半，而且两次 Map/Reduce 比 COUNT(DISTINCT)效率更高。
    调优结果：千万级别的类目表，member 表，与 10 亿级得商品表关联。原先 1963s 的任务经过调整，1152s 即完成。

  • 消灭子查询内的 JOIN

    1
    2
    3
    SELECT * FROM 
    (SELECT * FROM t1 UNION ALL SELECT * FROM t4 UNION ALL SELECT * FROM t2 JOIN t3 ON t2.id=t3.id) x 
    GROUP BY c1,c2;

    上面代码运行会有 5 个 jobs。加入先 JOIN 生存临时表的话 t5，然后 UNION ALL，会变成 2 个 jobs。

    1
    2
    3
    INSERT OVERWRITE TABLE t5 
    SELECT * FROM t2 JOIN t3 ON t2.id=t3.id; 
    SELECT * FROM (t1 UNION ALL t4 UNION ALL t5);

    调优结果显示：针对千万级别的广告位表，由原先 5 个 Job 共 15 分钟，分解为 2 个 job 一个 8-10 分钟，一个3分钟。

总结

使map的输出数据更均匀的分布到reduce中去，是我们的最终目标。由于Hash算法的局限性，按key Hash会或多或少的造成数据倾斜。大量经验表明数据倾斜的原因是人为的建表疏忽或业务逻辑可以规避的。在此给出较为通用的步骤：


1、采样log表，哪些user_id比较倾斜，得到一个结果表tmp1。由于对计算框架来说，所有的数据过来，他都是不知道数据分布情况的，所以采样是并不可少的。


2、数据的分布符合社会学统计规则，贫富不均。倾斜的key不会太多，就像一个社会的富人不多，奇特的人不多一样。所以tmp1记录数会很少。把tmp1和users做map join生成tmp2,把tmp2读到distribute file cache。这是一个map过程。


3、map读入users和log，假如记录来自log,则检查user_id是否在tmp2里，如果是，输出到本地文件a,否则生成<user_id,value>的key,value对，假如记录来自member,生成<user_id,value>的key,value对，进入reduce阶段。


4、最终把a文件，把Stage3 reduce阶段输出的文件合并起写到hdfs。


如果确认业务需要这样倾斜的逻辑，考虑以下的优化方案：
1、对于join，在判断小表不大于1G的情况下，使用map join
2、对于group by或distinct，设定 hive.groupby.skewindata=true
3、尽量使用上述的SQL语句调节进行优化


hadoop处理数据的过程，有几个显著的特征:

  1. 不怕数据多，就怕数据倾斜。
  2. 对jobs数比较多的作业运行效率相对比较低，比如即使有几百行的表，如果多次关联多次汇总，产生十几个jobs，没半小时是跑不完的。map reduce作业初始化的时间是比较长的。
  3. 对sum，count来说，不存在数据倾斜问题。
  4. 对count(distinct ),效率较低，数据量一多，准出问题，如果是多count(distinct )效率更低。


优化可以从几个方面着手
好的模型设计事半功倍。解决数据倾斜问题。减少job数。设置合理的map reduce的task数，能有效提升性能。(比如，10w+级别的计算，用160个reduce，那是相当的浪费，1个足够)。


自己动手写sql解决数据倾斜问题是个不错的选择。set hive.groupby.skewindata=true;这是通用的算法优化，但算法优化总是漠视业务，习惯性提供通用的解决方法。 Etl开发人员更了解业务，更了解数据，所以通过业务逻辑解决倾斜的方法往往更精确，更有效。


对count(distinct)采取漠视的方法，尤其数据大的时候很容易产生倾斜问题，不抱侥幸心理。自己动手，丰衣足食。


对小文件进行合并，是行至有效的提高调度效率的方法，假如我们的作业设置合理的文件数，对云梯的整体调度效率也会产生积极的影响。


优化时把握整体，单个作业最优不如整体最优。


细节上就是**

  1. 去除查询中不需要的column
  2. Where条件判断等在TableScan阶段就进行过滤
  3. 利用Partition信息，只读取符合条件的Partition
  4. Map端join，以大表作驱动，小表载入所有mapper内存中
  5. 调整Join顺序，确保以大表作为驱动表
  6. 对于数据分布不均衡的表Group by时，为避免数据集中到少数的reducer上，分成两个map-reduce阶段。第一个阶段先用Distinct列进行shuffle，然后在reduce端部分聚合，减小数据规模，第二个map-reduce阶段再按group-by列聚合。
  7. 在map端用hash进行部分聚合，减小reduce端数据处理规模。

Brent大数据工程师 & 机器学习

一心九用的工程师