背景
在刚使用hive的过程中,碰到过很多问题,任务经常需要运行7,8个小时甚至更久,在此记录一下这个过程中,我的一些收获
join长尾
背景
SQL在Join执行阶段会将Join Key相同的数据分发到同一个执行Instance上处理。如果某个Key上的数据量比较多,会导致该Instance执行时间比其他Instance执行时间长。其表现为:执行日志中该Join Task的大部分Instance都已执行完成,但少数几个Instance一直处于执行中,这种现象称之为长尾
长尾类别&优化方法
小表长尾
Join倾斜时,如果某路输入比较小,可以采用Mapjoin避免倾斜。Mapjoin的原理是将Join操作提前到Map端执行,这样可以避免因为分发Key不均匀导致数据倾斜。但是Mapjoin的使用有限制,必须是Join中的从表比较小才可用。所谓从表,即Left Outer Join中的右表,或者Right Outer Join中的左表。
热点值长尾
如果是因为热点值导致长尾,并且Join的输入比较大无法用Mapjoin,可以先将热点Key取出,对于主表数据用热点Key切分成热点数据和非热点数据两部分分别处理,最后合并。我们举一个电商的例子,假设我们需要计算所有商品的pv。我们有如下两张表
日志表 log 用户点击的日志, 包含商品的id p_id
商品表 product 包含商品名称 p_name, 商品id p_id- 取热点值, 取商品pv大于10000的商品到临时表
INSERT TABLE topk_product SELECT distinct p_idFROM ( SELECT p_id, count(1) AS pv FROM log GROUP BY p_id ) aWHERE pv >= 10000
- 取出非热点值和商品(product) join 得到非热点商品的pv
SELECT p.p_id , p.p_name , l.pvFROM ( SELECT p_id , p_name FROM product) pJOIN ( SELECT /*+mapjoin(b)*/ a.* FROM ( SELECT p_id , COUNT(1) AS pv FROM log ) a LEFT OUTER JOIN ( SELECT p_id FROM topk_product ) b ON a.p_id = b.p_id AND b.p_id IS NULL) lON p.p_id = l.p_id
- 取出热点值和商品(product) join 得到热点商品的pv
SELECT p.p_id , p.p_name , l.pvFROM ( SELECT /*+mapjoin(b)*/ a.* FROM ( SELECT p_id , p_name FROM product ) a JOIN ( SELECT p_id FROM topk_product ) b ON a.p_id = b.p_id) pJOIN ( SELECT /*+mapjoin(d)*/ c.* FROM ( SELECT p_id , COUNT(1) AS pv FROM log ) c JOIN ( SELECT p_id FROM topk_product ) d ON c.p_id = d.p_id) lON p.p_id = l.p_id
- union all 热点和非热点的数据
空值长尾
join时,假设左表(left_table)存在大量的空值,空值聚集到一个reduce上。由于left_table 存在大量的记录,无法使用mapjoin 。此时可以使用 coalesce(left_table.key, rand()*9999)将key为空的情况下赋予随机值,来避免空值集中造成长尾。
map长尾
Map端读取数据时,由于文件大小分布不均匀,一些map任务读取并处理的数据特别多,一些map任务处理的数据特别少,造成map端长尾。这种情形没有特别好的方法,只能调节splitsize来增加mapper数量,让数据分片更小,以期望获得更均匀的分配。
reduce长尾
由于Distinct操作的存在,数据无法在Map端的Shuffle阶段根据Group By先做一次聚合操作,减少传输的数据量,而是将所有的数据都传输到Reduce端,当Key的数据分发不均匀时,就会导致Reduce端长尾,特别当多个Distinct同时出现在一段SQL代码中时,数据会被分发多次,不仅会造成数据膨胀N倍,也会把长尾现象放大N倍。
我们用代码举个例子:
只有一个distinct 的情况
- 原sql
SELECT D1 , D2 , COUNT(DISTINCT CASE WHEN A IS NOT NULL THEN B END) AS B_distinct_cntFROM xxxGROUP BY D1, D2
- 改后的sql
create table tmp1asselect D1,D2,B,count( case when A is not null then B end ) as B_cntfrom xxxgroup by D1, D1, Bselect D1,D2,sum(case when B_cnt > 0 then 1 else 0 end) as B_distinct_cntfrom tmp1group by D1,D2
多个distinct的情况
- 原始sql
select D1,D2,count(distinct case when A is not null then B end) as B_distinct_cnt ,count(distinct case when E is not null then C end) as C_distinct_cnt from xxx group by D1,D2
- 修改后的sql
create table tmp1asselect D1,D2,B,count( case when A is not null then B end ) as B_cntfrom xxxgroup by D1, D1, Bcreate table tmp1_1asselect D1,D2,sum(case when B_cnt > 0 then 1 else 0 end) as B_distinct_cntfrom tmp1group by D1,D2create table tmp2asselect D1,D2,C,count( case when E is not null then C end ) as C_cntfrom xxxgroup by D1, D1, Ccreate table tmp2_1asselect D1,D2,sum(case when C_cnt > 0 then 1 else 0 end) as C_distinct_cntfrom tmp1group by D1,D2select t1.D1,t1.D2,t1.B_distinct_cnt,t2.C_distinct_cntfrom tmp1_1 t1left outer join tmp2_1 t2on t1.D1=t2.D1 and t1.D2=t2.D2