数据分析工具篇-HQL原理及优化

 HQL是数据分析过程中的必备技能，随着数据量增加，这一技能越来越重要，熟练应用的同时会带来效率的问题，动辄十几亿的数据量如果处理不完善的话有可能导致一个作业运行几个小时，更严重的还有可能因占用过多资源而引发生产问题，所以HQL优化就变得非常重要，本文我们就深入HQL的原理中，探索HQL优化的方法和逻辑。

group by的计算原理

代码为：

SELECT uid, SUM(COUNT) FROM logs GROUP BY uid; 

 

可以看到，group by本身不是全局变量，任务会被分到各个map中进行分组，然后再在reduce中聚合。

默认设置了hive.map.aggr=true，所以会在mapper端先group by一次，最后再把结果merge起来，为了减少reducer处理的数据量。注意看explain的mode是不一样的。mapper是hash，reducer是mergepartial。如果把hive.map.aggr=false，那将groupby放到reducer才做，他的mode是complete。

优化点：

Group by主要是面对数据倾斜的问题。

很多聚合操作可以现在map端进行，最后在Reduce端完成结果输出：

Set hive.map.aggr = true；  # 是否在Map端进行聚合，默认为true； 
Set hive.groupby.mapaggr.checkinterval = 1000000；  # 在Map端进行聚合操作的条目数目； 

当使用Group by有数据倾斜的时候进行负载均衡：

Set hive.groupby.skewindata = true； # hive自动进行负载均衡； 

策略就是把MR任务拆分成两个MR Job：第一个先做预汇总，第二个再做最终汇总;

第一个Job：

Map输出结果集中缓存到maptask中，每个Reduce做部分聚合操作，并输出结果，这样处理的结果是相同Group by Key有可能被分到不同的reduce中，从而达到负载均衡的目的;

第二个Job：

根据第一阶段处理的数据结果按照group by key分布到reduce中，保证相同的group by key分布到同一个reduce中，最后完成最终的聚合操作。

join的优化原理

代码为:

SELECT a.id,a.dept,b.age FROM a join b ON (a.id = b.id); 

 

1)Map阶段：

读取源表的数据，Map输出时候以Join on条件中的列为key，如果Join有多个关联键，则以这些关联键的组合作为key;

Map输出的value为join之后所关心的(select或者where中需要用到的)列;同时在value中还会包含表的Tag信息，用于标明此value对应哪个表;

按照key进行排序;

2)Shuffle阶段：

根据key的值进行hash,并将key/value按照hash值推送至不同的reduce中，这样确保两个表中相同的key位于同一个reduce中。

3)Reduce阶段：

根据key的值完成join操作，期间通过Tag来识别不同表中的数据。

在多表join关联时：

如果 Join 的 key 相同，不管有多少个表，都会合并为一个 Map-Reduce，例如：

SELECT pv.pageid, u.age  
FROM page_view p  
JOIN user u  
ON (pv.userid = u.userid)  
JOIN newuser x  
ON (u.userid = x.userid); 

如果 Join 的 key不同，Map-Reduce 的任务数目和 Join 操作的数目是对应的，例如：

SELECT pv.pageid, u.age  
FROM page_view p  
JOIN user u  
ON (pv.userid = u.userid)  
JOIN newuser x  
on (u.age = x.age); 

优化点：

1)应该将条目少的表/子查询放在 Join 操作符的左边。

2)我们知道文件数目小，容易在文件存储端造成瓶颈，给 HDFS 带来压力，影响处理效率。对此，可以通过合并Map和Reduce的结果文件来消除这样的影响。用于设置合并属性的参数有：

合并Map输出文件：hive.merge.mapfiles=true（默认值为真） 
合并Reduce端输出文件：hive.merge.mapredfiles=false（默认值为假） 
合并文件的大小：hive.merge.size.per.task=256*1000*1000（默认值为 256000000） 

3) Common join即普通的join，性能较差，因为涉及到了shuffle的过程(Hadoop/spark开发的过程中，有一个原则：能避免不使用shuffle就不使用shuffle)，可以转化成map join。

hive.auto.convert.join=true；# 表示将运算转化成map join方式 

使用的前提条件是需要的数据在 Map 的过程中可以访问到。

 

1)启动Task A：Task A去启动一个MapReduce的local task;通过该local task把small table data的数据读取进来;之后会生成一个HashTable Files;之后将该文件加载到分布式缓存(Distributed Cache)中来;

2)启动MapJoin Task：去读大表的数据，每读一个就会去和Distributed Cache中的数据去关联一次，关联上后进行输出。

整个阶段，没有reduce 和 shuffle，问题在于如果小表过大，可能会出现OOM。

Union与union all优化原理

union将多个结果集合并为一个结果集，结果集去重。代码为：

select id,name  
from t1  
union  
select id,name  
from t2  
union  
select id,name  
from t3 

对应的运行逻辑为：

union all将多个结果集合并为一个结果集，结果集不去重。使用时多与group by结合使用，代码为：

select all.id, all.name  
from( 
   select id,name  
   from t1  
   union all  
   select id,name  
   from t2  
   union all  
   select id,name  
   from t3 
)all  
group by all.id ,all.name 

对应的运行逻辑为：

从上面的两个逻辑图可以看到，第二种写法性能要好。union写法每两份数据都要先合并去重一次，再和另一份数据合并去重，会产生较多次的reduce。第二种写法直接将所有数据合并再一次性去重。

对union all的操作除了与group by结合使用还有一些细节需要注意：

1)对 union all 优化只局限于非嵌套查询。

原代码：job有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 

这样的结构是不对的，应该修改为：job有1个：

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

这样的修改可以减少job数量，进而提高效率。

2)语句中出现count(distinct …)结构时：

原代码为：

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)作业不但能解决倾斜问题，还能有效减少jobs)。

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; 

Order by的优化原理

如果指定了hive.mapred.mode=strict(默认值是nonstrict),这时就必须指定limit来限制输出条数，原因是：所有的数据都会在同一个reducer端进行，数据量大的情况下可能不能出结果，那么在这样的严格模式下，必须指定输出的条数。

所以数据量大的时候能不用order by就不用，可以使用sort by结合distribute by来进行实现。

sort by是局部排序;

distribute by是控制map怎么划分reducer。

cluster by=distribute by + sort by 

被distribute by设定的字段为KEY，数据会被HASH分发到不同的reducer机器上，然后sort by会对同一个reducer机器上的每组数据进行局部排序。

例如：

select mid, money, name  
from store  
cluster by mid 
 
select mid, money, name  
from store  
distribute by mid  
sort by mid 

如果需要获得与上面的中语句一样的效果：

select mid, money, name  
from store  
cluster by mid  
sort by money 

注意被cluster by指定的列只能是降序，不能指定asc和desc。

不过即使是先distribute by然后sort by这样的操作，如果某个分组数据太大也会超出reduce节点的存储限制，常常会出现137内存溢出的错误，对大数据量的排序都是应该避免的。

Count(distinct …)优化

如下的sql会存在性能问题：

SELECT COUNT( DISTINCT id ) FROM TABLE_NAME WHERE ...; 

主要原因是COUNT这种“全聚合(full aggregates)”计算时，它会忽略用户指定的Reduce Task数，而强制使用1，这会导致最终Map的全部输出由单个的ReduceTask处理。这唯一的Reduce Task需要Shuffle大量的数据，并且进行排序聚合等处理，这使得它成为整个作业的IO和运算瓶颈。

图形如下：

为了避免这一结构，我们采用嵌套的方式优化sql：

SELECT COUNT(*)  
FROM ( 
  SELECT DISTINCT id FROM TABLE_NAME WHERE …  
) t; 

这一结构会将任务切分成两个，第一个任务借用多个reduce实现distinct去重并进行初步count计算，然后再将计算结果输出到第二个任务中进行计数。

 

另外，再有的方法就是用group by()嵌套代替count(distinct a)。

如果能用group by的就尽量使用group by，因为group by性能比distinct更好。

HiveSQL细节优化

1) 设置合理的mapreduce的task数，能有效提升性能。

set mapred.reduce.tasks=n 

2) 在sql中or的用法需要加括号，否则可能引起无分区限制：

Select x 
from t  
where ds=d1  
and (province=’gd’ or province=’gx’) 

3) 对运算结果进行压缩：

set hive.exec.compress.output=true; 

4) 减少生成的mapreduce步骤:

1)使用CASE…WHEN…代替子查询;

2)尽量尽早地过滤数据，减少每个阶段的数据量,对于分区表要加分区，同时只选择需要使用到的字段;

5) 在map阶段读取数据前，FileInputFormat会将输入文件分割成split。split的个数决定了map的个数。

mapreduce.input.fileinputformat.split.minsize 默认值 0 
mapreduce.input.fileinputformat.split.maxsize 默认值 Integer.MAX_VALUE 
dfs.blockSize 默认值 128M，所以在默认情况下 map的数量=block数 

6) 常用的参数：

hive.exec.reducers.bytes.per.reducer=1000000； 

设置每个reduce处理的数据量，reduce个数=map端输出数据总量/参数;

set hive.mapred.mode=nonstrict; 
set hive.exec.dynamic.partition=true; 
set hive.exec.dynamic.partition.mode=nonstrict; 
set mapred.job.name=p_${v_date}; 
set mapred.job.priority=HIGH; 
set hive.groupby.skewindata=true; 
set hive.merge.mapredfiles=true; 
set hive.exec.compress.output=true; 
set mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec; 
set mapred.output.compression.type=BLOCK; 
set mapreduce.map.memory.mb=4096; 
set mapreduce.reduce.memory.mb=4096; 
set hive.hadoop.supports.splittable.combineinputformat=true; 
set mapred.max.split.size=16000000; 
set mapred.min.split.size.per.node=16000000; 
set mapred.min.split.size.per.rack=16000000; 
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; 
set hive.exec.reducers.bytes.per.reducer=128000000; 

7)设置map个数：

map个数和来源表文件压缩格式有关，.gz格式的压缩文件无法切分，每个文件会生成一个map;

set hive.hadoop.supports.splittable.combineinputformat=true; 只有这个参数打开，下面的3个参数才能生效 
set mapred.max.split.size=16000000; 每个map负载； 
set mapred.min.split.size.per.node=100000000; 每个节点map的最小负载，这个值必须小于set mapred.max.split.size的值； 
set mapred.min.split.size.per.rack=100000000; 每个机架map的最小负载； 
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;