浅谈Hive SQL的优化

目前团队的数据处理都在Hadoop集群上，

一是因为需要处理的数据量都是亿级的，这种规模的数据适合用Hadoop集群并行处理；

二是免除了分库分表给查询处理上带来的麻烦。Hive是基于Hadoop的一个数据仓库工具，它将存储在HDFS上的结构化的文件映射成一张关系型数据库表，提供简单的SQL查询功能。本文结合Hive SQL的运行原理谈一谈Hive SQL的优化问题。

1. 数据过滤应尽早做，同时只选择所需要的列

这个原则在传统的RDMS数据中应该也适用，因为数据提前过滤了之后进行join或者其他操作的数据量就会变少，这时往往会带来效率上很大的提升。例如：

select 
    b.f_pass_id, 
    a.F_buyer_user_id, 
    a.F_trans_id, 
    a.F_state, 
    a.F_create_time, 
    a.F_pay_time, 
    a.F_seller_user_id, 
    a.F_seller_true_name, 
    a.F_total_amount, 
    a.F_mkt_solution_amount
from jxyy.ods_t_trans_df a 
left outer join jxyy.ods_t_user_reflect_df b
on a.F_buyer_user_id=b.f_user_id
where a.ds='20220119' 
and a.F_buyer_user_id>=10000000 
and a.F_buyer_user_id < 100000000000 
and a.F_create_time >='2014-01-01' 
and a.F_trans_type in (1,2) 
and a.F_trans_mode=1 
and b.ds='20220119' 
and b.f_enabled=1

应该修改成：

select 
    b.f_pass_id, 
    a.F_buyer_user_id, 
    a.F_trans_id, 
    a.F_state, 
    a.F_create_time, 
    a.F_pay_time, 
    a.F_seller_user_id, 
    a.F_seller_true_name, 
    a.F_total_amount, 
    a.F_mkt_solution_amount
from 
(
    select 
        F_buyer_user_id, 
        F_trans_id, F_state, 
        F_create_time, 
        F_pay_time, 
        F_seller_user_id, 
        F_seller_true_name, 
        F_total_amount, 
        F_mkt_solution_amount 
    from jxyy.ods_t_trans_df 
    where ds='20220119' 
    and F_buyer_user_id >= 10000000 
    and F_buyer_user_id<100000000000 
    and F_create_time >='2014-01-01' 
    and F_trans_type in (1,2) 
    and F_trans_mode=1
) a
left outer join 
(
    select 
        f_pass_id 
    from jxyy.ods_t_user_reflect_df 
    where ds='20220119' 
    and f_enabled=1
) b
on a.F_buyer_user_id=b.f_user_id

2. 多个job间可以并行

一个Hive查询是可以生成多个job的，而且有些job之间是可以并行的，典型的比如多个子查询之间是可以并行做的。

同样是上面的查询优化的例子，生成a表的子查询和生成b表的子查询是可以同时进行的。我们可以通过设置hive.exec.parallel=true来设置一个Hive产生的多个有偏序关系的job间是可以并行的执行的，并且可以通过hive.exec.parallel.thread.number这个参数来设着并行度，默认的并行度是8。

但是并不是生成的job数越多越好，我们实际开发中应该避免生成多余的job而影响效率。

例如有个查询需求：我们要统计pay_transaction表中使用过支付宝支付和钱包支付的用户有多少。一种思路是：先统计出使用过支付宝支付的用户id，然后同样的方法统计用钱包支付的用户id，最后这两个用户集做交集（也就是两个子查询的结果做join）：

select 
    a.F_pass_uid 
from (
    select 
        distinct F_pass_uid 
    from ods.t_pay_transaction_df 
    where ds = '20220119' 
    and F_agent_name like '%AliPay%'
) a
join
(
    select 
        distinct F_pass_uid 
    from ods.t_pay_transaction_df 
    where ds = '20220119' 
    and F_agent_name like '%Baifubao%'
) b
on a.F_pass_uid=b.F_pass_uid

这样产生了一共四个job，两个子查询，这两个job可以并行，一个连接的job，一个计数的job。

但是我们改成以下的查询方式只生成一个job，而且更加符合M/R方式：

select 
    a.F_pass_uid
from ods.t_pay_transaction_df
where ds = '20160519'
group by F_pass_uid
having count(
    case when F_agent_name like '%AliPay%' then 1 else 0 end) > 0 
    and count(case when F_agent_name like '%Baifubao%' then 1 else 0 end) > 0

所以并不是job数越多越好，我们在写查询语句时应尽量避免生成冗余的job，能用一个job就能完成的就不要用两个，因为启动一个M/R job的系统开销还是挺大的。

3. 数据倾斜问题

数据倾斜问题是Hive优化的重点问题。在执行Hive查询时，如果任务的进度长时间的停留在百分之九十多，多半就是数据倾斜了。数据倾斜就是某一个或某几个reduce作业处理的数据量远高于其他的。数据倾斜的原因有多方面的，比如key的分布不均匀、业务数据本身就有倾斜、有些SQL语句本省就会产生数据倾斜。

常见的产生数据倾斜的场景就是使用join和group by时。比如group by时某个取值数量就是很多，这样不可避免的就有数据倾斜。

1) 调解参数

    设置hive.map.aggr = true，在 Map 端进行聚合，也就是combiner造作。这样就会使得在reduce端的数据量有效的减少，可以一定程度上缓解数据倾斜的程度。

    设置hive.groupby.skewindata=true，这样当有数据倾斜时就会进行负载均衡。如在group by时出现数据倾斜了，就可以把延时很长的作业分配一部分给其他已经完成的reduce做，最后再聚合结果。

2)处理key分布不均匀

    比如group by key或者join on key时，对于key中有空值或者数据量明显过多的key可以在原来的值得基础上加一个随机数，这样就可以把倾斜的数据分不到不同的reduce上，只是最后要把结果还原。如：

select 
    case when F_agent_name is null 
    then concat('null',rand()%20) 
    else F_agent_name end as agent_name, 
    count(1) as pv
from ods.t_pay_transaction_df 
where ds = '20220119'
group by case when F_agent_name is null then concat('null',rand()%20) else F_agent_name end

    这个例子中，因为发现F_agent_name大量的是空值，造成数据倾斜，所以把为空的key随机的分布到5个reduce作业中。

3）Join优化

    选用join key分布最均匀的表作为驱动表，并且大表放在右边，小表放在左边。也可以采用mapjoin。

4）排序优化

    order by 排序，最后只存在一个reduce，效率比较低。可以用sort by操作，然后结合distribute by作为reduce分区键。

5）特殊处理

    如果以上优化效果不是很大的情况下，可以将倾斜的数据单独处理。最后再将结果union回去。

4. 选用合适的Job执行模式

Hadoop的M/R作业有三种执行模式：

本地模式、伪分布模式和分布模式。

    顾名思义，本地模式就是单机模式，也就是map个reduce的作业数都是1。因为我们在Hadoop集群上处理数据时，有些时候数据量非常小的job启用分布式job会有大量的初始化、作业协调、进程控制等系统开销，而真正执行数据处理的时间却非常短，这样得不偿失，所以应该使用本地模式。

    通过设置hive.exec.mode.local.auto=true就能够自动开启local mr模式开启本地模式。除此之外，还需要控制输入文件数量和数据总量大小，通过两个参数分别设置，hive.exec.mode.local.auto.tasks.max和hive.exec.mode.local.auto.inputbytes.max。这两个参数默认值分别为4和128。也就是在hive.exec.mode.local.auto=true时，默认情况下文件数不超过4个并且文件总大小小于128MB就启用本地模式。

5. 总结

    本文结合Hive SQL的运行原理简单说明了Hive SQL的优化问题，其实Hive语句的优化做好了，对速度提升和资源的节约帮助非常大，希望能对各位以后再写Hive SQL处理大数据时有所启发和帮助。