目录

一、调优思路1、SQL优化1.1 大小表join1.2 大大表join

2、insert into替换union all3、排序order by换位sort by4、并行执行5、数据倾斜优化6、小文件优化

二、实战2.1 场景2.2 限制所需的字段,间接mapjoin2.2 解决异常值倾斜,如NULL加随机数打散2.3 扩容解决数据倾斜2.3.1 客户表扩大N倍2.3.2 部分倾斜key扩容,大卖家扩容2.3.3 推荐:分而治之:倾斜和非倾斜再union all

在Hive中,优化器会根据统计信息决定是将大表放在前面(Join的左边)还是小表放在前面。通常,优化器会选择数据量较小的表作为驱动表(小表作为左边),因为这样可以减少内存消耗并提高效率。

但是,如果你有特定的需求,比如你知道大部分数据能快速过滤掉,希望减少任务的执行时间,那么你可以强制指定某个表作为小表。在Hive中,可以使用/*+ MAPJOIN(table_name) */ 注释来强制将一个大表作为小表处理。 例如,如果你想要将big_table作为小表:

SELECT /*+ MAPJOIN(big_table) */

a.column1, a.column2, b.column1, b.column2

FROM

small_table a

JOIN

big_table b

ON

a.common_column = b.common_column;

一、调优思路

1、SQL优化

1.1 大小表join

1、mapjoin,小表使用mapjoin,或者强制hint 2、将大表放后头,原因:Hive假定查询中最后的一个表是大表。它会将其它表缓存起来,然后扫描最后那个表。因此通常需要将小表放前面,或者标记哪张表是大表:/*streamtable(table_name) */ 3、过滤无效值:空值、不使用的字段等。 4、不能过滤的空值,将空值转化为随机数避免数据倾斜。

1.2 大大表join

1)创建第二张大表

create table bigtable2(

id bigint,

t bigint,

uid string,

keyword string,

url_rank int,

click_num int,

click_url string)

row format delimited fields terminated by '\t';

load data local inpath '/opt/module/data/bigtable' into table bigtable2;

2)测试大表直接JOIN

insert overwrite table jointable

select b.id, b.t, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url

from bigtable a

join bigtable2 b

on a.id = b.id;

测试结果:Time taken: 72.289 seconds

insert overwrite table jointable

select b.id, b.t, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url

from bigtable a

join bigtable2 b

on a.id = b.id;

3)创建分桶表1

create table bigtable_buck1(

id bigint,

t bigint,

uid string,

keyword string,

url_rank int,

click_num int,

click_url string)

clustered by(id)

sorted by(id)

into 6 buckets

row format delimited fields terminated by '\t';

load data local inpath '/opt/module/data/bigtable' into table bigtable_buck1;

4)创建分桶表2,分桶数和第一张表的分桶数为倍数关系

create table bigtable_buck2(

id bigint,

t bigint,

uid string,

keyword string,

url_rank int,

click_num int,

click_url string)

clustered by(id)

sorted by(id)

into 6 buckets

row format delimited fields terminated by '\t';

load data local inpath '/opt/module/data/bigtable' into table bigtable_buck2;

5)设置参数

set hive.optimize.bucketmapjoin = true;

set hive.optimize.bucketmapjoin.sortedmerge = true;

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

6)测试 Time taken: 34.685 seconds

insert overwrite table jointable

select b.id, b.t, b.uid, b.keyword, b.url_rank, b.click_num, b.click_url

from bigtable_buck1 s

join bigtable_buck2 b

on b.id = s.id;

1、使用相同的连接键

当对3个或者更多个表进行join连接时,如果每个on子句都使用相同的连接键的话,那么只会产生一个MapReduce job。

2、过滤无效、未使用的数据:减少每个阶段的数据量,对于分区表要加分区,同时只选择需要使用到的字段。

加随机数打散

1)空值0值 或 关联不上的,用随机数

from a join b

on if(a.key=’’, rand(id)%10, a.key)=b.key

–rand() 0-1之间的小数

(2)都是有用的key,则加随机数后缀

group by concat(key, cast(round(rand()*10) as int))

缺点是分成10份是提前写好的,数据变更大时,还是会跑得慢。

3、逻辑拆分,使用中间表计算

尽量原子化操作:多个表关联时,避免包含复杂逻辑大sql(因为无法控制中间job),最好分拆成小段,可以使用中间表来完成复杂的逻辑写入HDFS之后:多次INSERT OVERWRITE TABLE写法参考:spark调优-小文件问题

4、列裁剪,避免使用select * 如果查询的是分区表,一定要记得带上分区条件 5、where条件写在子查询中:先过滤再关联(最好使用这种笨办法,虽然hive3.0自带谓词下推) 6、关联条件写在on中,而不是where中

非主表谓词下推情况下,可以理解为where是全部执行完在reduce中进行过滤,on是在关联过程中filter

7、数据量小时,用in代替join 8、使用semi join替代in/exists inner join和left semi join的联系和区别

2、insert into替换union all

如果union all的部分个数大于2,或者每个union部分数据量大,应该拆成多个insert into 语句,效率有提升。 ?insert into到不同分区?

3、排序order by换位sort by

order by:对输入做全局排序,因此只有一个reducer(多个reducer无法保证全局有序),只有一个reducer,会导致当输入规模较大时,需要较长的计算时间。 sort by:局部排序,保证每个reducer的输出文件是有序的。 hive order by,sort by, distribute by, cluster by作用以及用法

4、并行执行

Hive会将一个查询转化成一个或者多个阶段。这样的阶段可以是MapReduce阶段、抽样阶段、合并阶段、limit阶段。或者Hive执行过程中可能需要的其他阶段。 默认情况下,Hive一次只会执行一个阶段。不过,某个特定的job可能包含众多的阶段,而这些阶段可能并非完全互相依赖的,也就是说有些阶段是可以并行执行的,这样可能使得整个job的执行时间缩短。如果有更多的阶段可以并行执行,那么job可能就越快完成。 通过设置参数hive.exec.parallel值为true,就可以开启并发执行。在共享集群中,需要注意下,如果job中并行阶段增多,那么集群利用率就会增加。

set hive.exec.parallel=true; //打开任务并行执行

set hive.exec.parallel.thread.number=16; //同一个sql允许最大并行度,默认为8。

set hive.exec.parallel=true

5、数据倾斜优化

spark-数据倾斜、 hadoop-hive-数据倾斜问题

6、小文件优化

spark调优-小文件问题

参考链接 HiveSQL大表join大表数据倾斜

二、实战

添加链接描述

2.1 场景

【背景】 A表为一个汇总表,汇总的是卖家买家最近N天交易汇总信息,即对于每个卖家最近N天,其每个买家共成交了多少单,总金额是多少,假设N取90天,汇总值仅取成交单数。 A表的字段有:buyer_id、seller_id、pay_cnt_90day。 B表为卖家基本信息表,其字段有seller_id、sale_level,其中sale_levels是卖家的一个分层评级信息,比如吧卖家分为6个级别:S0、S1、S2、S3、S4和S5。 要获得的结果是每个买家在各个级别的卖家的成交比例信息,比如:某买家:S0:10%;S1:20%;S2:20%;S3:10%;S4:20%;S5:10%。 【初始思路】 第一反应是直接join两表并统计:

select

         m.buyer_id,

        sum(pay_cnt_90day) as pay_cnt_90day,

        sum(case when m.sale_level = 0 then pay_cnt_90day end) as pay_cnt_90day_s0,

        sum(case when m.sale_level = 1 then pay_cnt_90day end) as pay_cnt_90day_s1,

        sum(case when m.sale_level = 2 then pay_cnt_90day end) as pay_cnt_90day_s2,

        sum(case when m.sale_level = 3 then pay_cnt_90day end) as pay_cnt_90day_s3,

        sum(case when m.sale_level = 4 then pay_cnt_90day end) as pay_cnt_90day_s4,

        sum(case when m.sale_level = 5 then pay_cnt_90day end) as pay_cnt_90day_s5

      from (

        select a.buer_id, a.seller_id, b.sale_level, a.pay_cnt_90day

        from ( select buyer_id, seller_id, pay_cnt_90day from table_A) a

        join

         (select seller_id, sale_level from table_B) b

        on a.seller_id = b.seller_id

        ) m

      group by m.buyer_id

但是此SQL会引起数据倾斜,原因在于卖家的二八准则,某些卖家90天内会有几百万甚至上千万的买家,但是大部分的卖家90天内买家的数目并不多,join table_A和table_B的时候,ODPS会按照seller_id进行分发,table_A的大卖家引起了数据倾斜。但是数据本身无法用mapjoin table_B解决,因为卖家超过千万条,文件大小有几个GB,超过了1GB的限制。

2.2 限制所需的字段,间接mapjoin

思路:只看90天内有交易的卖家,不join全部的卖家表 局限:此方案在一些情况可以起作用,但是很多时候还是无法解决上述问题,因为大部分卖家尽管90天内买家不多,但还是有一些的,过滤后的B表仍然很多。

 select

         m.buyer_id,

        sum(pay_cnt_90day) as pay_cnt_90day,

        sum(case when m.sale_level = 0 then pay_cnt_90day end) as pay_cnt_90day_s0,

        sum(case when m.sale_level = 1 then pay_cnt_90day end) as pay_cnt_90day_s1,

        sum(case when m.sale_level = 2 then pay_cnt_90day end) as pay_cnt_90day_s2,

        sum(case when m.sale_level = 3 then pay_cnt_90day end) as pay_cnt_90day_s3,

        sum(case when m.sale_level = 4 then pay_cnt_90day end) as pay_cnt_90day_s4,

        sum(case when m.sale_level = 5 then pay_cnt_90day end) as pay_cnt_90day_s5

      from (

        select /*+mapjoin(b)*/

          a.buer_id, a.seller_id, b.sale_level, a.pay_cnt_90day

        from ( select buyer_id, seller_id, pay_cnt_90day from table_A) a

        join (

           select seller_id, sale_level from table_B b0

           join

           (select seller_id from table_A group by seller_id) a0

           on b0.seller_id = a0.selller_id

          ) b

        on a.seller_id = b.seller_id

        ) m

      group by m.buyer_id   

2.2 解决异常值倾斜,如NULL加随机数打散

**思路:**核心是将这些引起倾斜的值随机分发到Reduce,join时对这些特殊值concat随机数,从而达到随机分发的目的。 **适用于:**倾斜的值是明确的而且数量很少,比如null值引起的倾斜。 **局限:**无法解决本问题场景的倾斜问题,因为倾斜的卖家大量存在而且动态变化。 此方案的核心逻辑如下:

select a.user_id, a.order_id, b.user_id

      from table_a a

      join table_b b

      on (case when a.user_is is null then concat('hive', rand(id)) else a.user_id end) = b.user_id

Hive 已对此进行了优化,只需要设置参数skewinfo和skewjoin参数,不修改SQL代码,例如,由于table_B的值“0” 和“1”引起了倾斜,值需要做如下设置:

set hive.optimize.skewinfo=table_B:(selleer_id) [ ( "0") ("1") ) ]

set hive.optimize.skewjoin = true;

2.3 扩容解决数据倾斜

推荐”2.3.3 推荐:分而治之:倾斜和非倾斜再union all“,可直接看。若不行,推荐方案2.3.2倾斜key扩容。

2.3.1 客户表扩大N倍

思路:按照seller_id分发会倾斜,那么再人工增加一列进行分发,这样之前倾斜的值的倾斜程度会减少到原来的1/10,可以通过配置numbers表改放大倍数来降低倾斜程度。 代码实现:建立一个numbers表,其值只有一列int 行,比如从1到10(具体值可根据倾斜程度确定),然后放大B表10倍,再取模join。 局限性:数据量翻倍,B表也会膨胀N倍。

SELECT m.buyer_id,

sum(pay_cnt_90day) AS pay_cnt_90day,

sum(case WHEN m.sale_level = 0 THEN pay_cnt_90day end) AS pay_cnt_90day_s0,

sum(case WHEN m.sale_level = 1 THEN pay_cnt_90day end) AS pay_cnt_90day_s1,

sum(case WHEN m.sale_level = 2 THEN pay_cnt_90day end) AS pay_cnt_90day_s2,

     sum(case when m.sale_level = 3 then pay_cnt_90day end) as pay_cnt_90day_s3,

     sum(case when m.sale_level = 4 then pay_cnt_90day end) as pay_cnt_90day_s4,

     sum(case when m.sale_level = 5 then pay_cnt_90day end) as pay_cnt_90day_s5

FROM

(SELECT a.buer_id,

a.seller_id,

b.sale_level,

a.pay_cnt_90day

FROM

(SELECT buyer_id,

seller_id,

pay_cnt_90day

FROM table_A) a

JOIN -- 将B表扩容N倍

(SELECT /*+mapjoin(members)*/ seller_id,

sale_level ,

member

FROM table_B

JOIN members -- 扩容N倍的表

) b

ON a.seller_id = b.seller_id

AND mod(a.pay_cnt_90day,10)+1 = b.number ) m

GROUP BY m.buyer_id

2.3.2 部分倾斜key扩容,大卖家扩容

思路:把大卖家放大倍数即可:需要首先知道大卖家的名单,即先建立一个临时表动态存放每天最新的大卖家(比如dim_big_seller),同时此表的大卖家要膨胀预先设定的倍数(1000倍)。 代码实现:在A表和B表分别新建一个join列,其逻辑为:如果是大卖家,那么concat一个随机分配正整数(0到预定义的倍数之间,本例为0~1000);如果不是,保持不变。 局限性: 相比全部数据扩容,仅倾斜指标扩容的运行效率有提升,但代码复杂性高,必须首先建立大数据表。

SELECT m.buyer_id,

sum(pay_cnt_90day) AS pay_cnt_90day,

sum(case WHEN m.sale_level = 0 THEN pay_cnt_90day end) AS pay_cnt_90day_s0,

sum(case WHEN m.sale_level = 1 THEN pay_cnt_90day end) AS pay_cnt_90day_s1,

sum(case WHEN m.sale_level = 2 THEN pay_cnt_90day end) AS pay_cnt_90day_s2,

     sum(case when m.sale_level = 3 then pay_cnt_90day end) as pay_cnt_90day_s3,

     sum(case when m.sale_level = 4 then pay_cnt_90day end) as pay_cnt_90day_s4,

     sum(case when m.sale_level = 5 then pay_cnt_90day end) as pay_cnt_90day_s5

pay_cnt_90day end) AS pay_cnt_90day_s5

FROM

(SELECT a.buer_id,

a.seller_id,

b.sale_level,

a.pay_cnt_90day

FROM

(SELECT /*+mapjoin(big)*/ buyer_id,

seller_id,

pay_cnt_90day,

if(big.seller_id is NOT null,

concat( table_A.seller_id,

'rnd', cast( rand() * 1000 AS bigint ), table_A.seller_id) AS seller_id_joinkey

FROM table_A left outer

JOIN --big表seller_id有重复,请注意一定要group by 后再join,保证table_A的行数保持不变

( SELECT seller_id

FROM dim_big_seller

GROUP BY seller_id

)big

ON table_A.seller_id = big.seller_id ) a

JOIN

(SELECT /*+mapjoin(big)*/ seller_id,

sale_level ,

--big表的seller_id_joinkey生成逻辑和上面的生成逻辑一样 coalesce(seller_id_joinkey,

table_B.seller_id) AS seller_id_joinkey

FROM table_B

left out JOIN --table_B表join大卖家表后大卖家行数扩大1000倍,其它卖家行数保持不变

(SELECT seller_id,

seller_id_joinkey

FROM dim_big_seller

) big

ON table_B.seller_id= big.seller_id ) b

ON a.seller_id_joinkey= b.seller_id_joinkey

AND mod(a.pay_cnt_90day,10)+1 = b.number ) m

GROUP BY m.buyer_id

2.3.3 推荐:分而治之:倾斜和非倾斜再union all

思路:对倾斜的键值和不倾斜的键值分开处理,不倾斜的正常join即可,倾斜的把他们找出来做mapjoin,最后union all其结果即可。 代码实现: 局限性:较麻烦,代码复杂而且需要一个临时表存放倾斜的键值。

--1、构建临时表,由于数据倾斜,先找出90天买家超过10000的卖家

insert overwrite table temp_table_B

SELECT m.seller_id,

n.sale_level

FROM

(SELECT seller_id

FROM

(SELECT seller_id,

count(buyer_id) AS byr_cnt

FROM table_A

GROUP BY seller_id ) a

WHERE a.byr_cnt >10000 ) m

LEFT JOIN

(SELECT seller_id,

sale_level

FROM table_B

) n

ON m.seller_id = n.seller_id;

--2、分而治之,不倾斜union all 倾斜。

--对于90天买家超过10000的卖家直接mapjoin,对其它卖家直接正常join即可

。SELECT m.buyer_id,

sum(pay_cnt_90day) AS pay_cnt_90day,

sum(case WHEN m.sale_level = 0 THEN pay_cnt_90day end) AS pay_cnt_90day_s0,

sum(case WHEN m.sale_level = 1 THEN pay_cnt_90day end) AS pay_cnt_90day_s1,

sum(case WHEN m.sale_level = 2 THEN pay_cnt_90day end) AS pay_cnt_90day_s2,

     sum(case when m.sale_level = 3 then pay_cnt_90day end) as pay_cnt_90day_s3,

     sum(case when m.sale_level = 4 then pay_cnt_90day end) as pay_cnt_90day_s4,

     sum(case when m.sale_level = 5 then pay_cnt_90day end) as pay_cnt_90day_s5

FROM

(SELECT a.buer_id,

a.seller_id,

b.sale_level,

a.pay_cnt_90day

FROM

(SELECT buyer_id,

seller_id,

pay_cnt_90day

FROM table_A

) a

JOIN

(SELECT seller_id,

a.sale_level

FROM table_A a

LEFT JOIN temp_table_B b

ON a.seller_id = b.seller_id

WHERE b.seller_id is NULL -- 限制为不倾斜的卖家

) b

ON a.seller_id = b.seller_id

UNION all

SELECT /*+mapjoin(b)*/ a.buer_id,

a.seller_id,

b.sale_level,

a.pay_cnt_90day

FROM

(SELECT buyer_id,

seller_id,

pay_cnt_90day

FROM table_A

) a

JOIN

(SELECT seller_id,

sale_level

FROM table_B -- 只看倾斜卖家

) b

ON a.seller_id = b.seller_id

) m

GROUP BY m.buyer_id ) m

GROUP BY m.buyer_id

推荐阅读

评论可见,请评论后查看内容,谢谢!!!评论后请刷新页面。