标签：style   blog   http   color   os   使用   io   ar   for   

一个Hive查询生成多个map reduce job，一个map reduce job又有map，reduce，spill，shuffle，sort等多个阶段，所以针对hive查询的优化可以大致分为针对MR中单个步骤的优化（其中又会有细分），针对MR全局的优化，和针对整个查询（多MR job）的优化，下文会分别阐述。

在开始之前，先把MR的流程图帖出来（摘自Hadoop权威指南），方便后面对照。另外要说明的是，这个优化只是针对Hive 0.9版本，而不是后来Hortonwork发起Stinger项目之后的版本。相对应的Hadoop版本是1.x而非2.x。

1、Map阶段的优化(map phase)

Map阶段的优化，主要是确定合适的map数。那么首先要了解map数的计算公式：

num_map_tasks = max[${mapred.min.split.size},
                min(${dfs.block.size}, ${mapred.max.split.size})]

• mapred.min.split.size指的是数据的最小分割单元大小。
• mapred.max.split.size指的是数据的最大分割单元大小。
• dfs.block.size指的是HDFS设置的数据块大小。

hive> set dfs.block.size;
dfs.block.size is undefined

hive> set mapred.min.split.size;
mapred.min.split.size=1
hive> set mapred.max.split.size;
mapred.max.split.size=256000000

2、Reduce阶段的优化(reduce phase)

num_reduce_tasks = min[${hive.exec.reducers.max}, 
                      (${input.size} / ${ hive.exec.reducers.bytes.per.reducer})]

3、Map与Reduce之间的优化(spill, copy, sort phase)

3.1 Spill 与 Sort

3.2 Copy

4、文件格式的优化

create table rc_file_test( col int ) stored as rcfile;
set hive.exec.compress.output = true;
insert overwrite table rc_file_test
select * from source_table;

set hive.default.fileformat = SequenceFile;
set hive.exec.compress.output = true; 
/*对于sequencefile，有record和block两种压缩方式可选，block压缩比更高*/
set mapred.output.compression.type = BLOCK; 
create table seq_file_test
as select * from source_table;

5、Job整体优化

5.1 Job执行模式

5.2 JVM重用

set mapred.job.reuse.jvm.num.tasks = 5;

5.3 索引

/*在index_test_table表的id字段上创建索引*/
create index idx on table index_test_table(id)  
as ‘org.apache.hadoop.hive.ql.index.compact.CompactIndexHandler‘ with deferred rebuild;
alter index idx on index_test_table rebuild;
	
/*索引的剪裁。找到上面建的索引表，根据你最终要用的查询条件剪裁一下。*/
/*如果你想跟RDBMS一样建完索引就用，那是不行的，会直接报错，这也是其麻烦的地方*/
create table my_index
as select _bucketname, `_offsets`
from default__index_test_table_idx__ where id = 10;
	
/*现在可以用索引了，注意最终查询条件跟上面的剪裁条件一致*/
set hive.index.compact.file = /user/hive/warehouse/my_index; 
set hive.input.format = org.apache.hadoop.hive.ql.index.compact.HiveCompactIndexInputFormat;
select count(*) from index_test_table where id = 10;

5.4 Join算法

create table map_join_test(id int)
clustered by (id) sorted by (id) into 32 buckets
stored as textfile;

set hive.enforce.bucketing = true;
insert overwrite table map_join_test
select * from map_join_source_data;

select /*+mapjoin(a) */count(*)
from map_join_test a
join map_join_test b on a.id = b.id;

2013-08-31 09:08:43     Starting to launch local task to process map join;      maximum memory = 1004929024
2013-08-31 09:08:45     Processing rows:   200000  Hashtable size: 199999  Memory usage:   38823016        rate:   0.039
2013-08-31 09:08:46     Processing rows:   300000  Hashtable size: 299999  Memory usage:   56166968        rate:   0.056
……
2013-08-31 09:12:39     Processing rows:  4900000 Hashtable size: 4899999 Memory usage:   896968104       rate:   0.893
2013-08-31 09:12:47     Processing rows:  5000000 Hashtable size: 4999999 Memory usage:   922733048       rate:   0.918
Execution failed with exit status: 2
Obtaining error information

Task failed!
Task ID:
  Stage-4

set hive.optimize.bucketmapjoin = true;

2013-08-31 09:20:39     Starting to launch local task to process map join;      maximum memory = 1004929024
2013-08-31 09:20:41     Processing rows:   200000  Hashtable size: 199999  Memory usage:   38844832        rate:   0.039
2013-08-31 09:20:42     Processing rows:   275567  Hashtable size: 275567  Memory usage:   51873632        rate:   0.052
2013-08-31 09:20:42     Dump the hashtable into file: file:/tmp/hadoop/hive_2013-08-31_21-20-37_444_1135806892100127714/-local-10003/HashTable-Stage-1/MapJoin-a-10-000000_0.hashtable
2013-08-31 09:20:46     Upload 1 File to: file:/tmp/hadoop/hive_2013-08-31_21-20-37_444_1135806892100127714/-local-10003/HashTable-Stage-1/MapJoin-a-10-000000_0.hashtable File size: 11022975
2013-08-31 09:20:47     Processing rows:   300000  Hashtable size: 24432   Memory usage:   8470976 rate:   0.008
2013-08-31 09:20:47     Processing rows:   400000  Hashtable size: 124432  Memory usage:   25368080        rate:   0.025
2013-08-31 09:20:48     Processing rows:   500000  Hashtable size: 224432  Memory usage:   42968080        rate:   0.043
2013-08-31 09:20:49     Processing rows:   551527  Hashtable size: 275960  Memory usage:   52022488        rate:   0.052
2013-08-31 09:20:49     Dump the hashtable into file: file:/tmp/hadoop/hive_2013-08-31_21-20-37_444_1135806892100127714/-local-10003/HashTable-Stage-1/MapJoin-a-10-000001_0.hashtable
……

set hive.optimize.bucketmapjoin.sortedmerge = true;
set hive.input.format = org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;

5.5 数据倾斜

所谓数据倾斜，说的是由于数据分布不均匀，个别值集中占据大部分数据量，加上hadoop的计算模式，导致计算资源不均匀引起性能下降。下图就是一个例子：

还是拿网站的访问日志说事吧。假设网站访问日志中会记录用户的user_id，并且对于注册用户使用其用户表的user_id，对于非注册用户使用一个user_id=0代表。那么鉴于大多数用户是非注册用户（只看不写），所以user_id=0占据了绝大多数。而如果进行计算的时候如果以user_id作为group by的维度或者是join key，那么个别reduce会收到比其他reduce多得多的数据——因为它要接收所有user_id=0的记录进行处理，使得其处理效果会非常差，其他reduce都跑完很久了它还在运行。
倾斜分成group by造成的倾斜和join造成的倾斜，需要分开看。
group by造成的倾斜有两个参数可以解决，一个是hive.map.aggr，默认值已经为true，意思是会做map端的combiner。所以如果你的group by查询只是做count(*)的话，其实是看不出倾斜效果的，但是如果你做的是count(distinct)，那么还是会看出一点倾斜效果。另一个参数是hive.groupby.skewindata。这个参数的意思是做reduce操作的时候，拿到的key并不是所有相同值给同一个reduce，而是随机分发，然后reduce做聚合，做完之后再做一轮MR，拿前面聚合过的数据再算结果。所以这个参数其实跟hive.map.aggr做的是类似的事情，只是拿到reduce端来做，而且要额外启动一轮job，所以其实不怎么推荐用，效果不明显。
如果说要改写SQL来优化的话，可以按照下面这么做：

/*改写前*/
select a, count(distinct b) as c from tbl group by a;
/*改写后*/
select a, count(*) as c
from (select distinct a, b from tbl) group by a;

select a.* from logs a join users b on a.user_id = b.user_id;

set hive.optimize.skewjoin = true;

另外对于特殊值的处理往往跟业务有关系，所以也可以从业务角度重写sql解决。比如前面这种倾斜join，可以把特殊值隔离开来（从业务角度说，users表应该不存在user_id = 0的情况，但是这里还是假设有这个值，使得这个写法更加具有通用性）：

SELECT a.*
FROM
  (SELECT a.*
   FROM
     (SELECT *
      FROM logs
      WHERE user_id = 0) a
   JOIN
     (SELECT *
      FROM users
      WHERE user_id = 0) b ON a.user_id = b.user_id
   UNION ALL SELECT a.*
   FROM logs a
   JOIN users b ON a.user_id <> 0
   AND a.user_id = b.user_id)t;

6、SQL整体优化

6.1 Job间并行

select * from 
(
   select count(*) from logs 
   where log_date = 20130801 and item_id = 1
   union all 
   select count(*) from logs 
   where log_date = 20130802 and item_id = 2
   union all 
   select count(*) from logs 
   where log_date = 20130803 and item_id = 3
)t

6.2 减少Job数

SELECT count(*)
FROM
  (SELECT DISTINCT user_id
   FROM logs
   WHERE page_name = ‘a‘) a
JOIN
  (SELECT DISTINCT user_id
   FROM logs
   WHERE blog_owner = ‘b‘) b ON a.user_id = b.user_id;

select count(*) 
from logs group by user_id
having (count(case when page_name = ‘a‘ then 1 end) > 0
    and count(case when page_name = ‘b‘ then 1 end) > 0)

第一种查询方法符合思考问题的直觉，是工程师和分析师在实际查数据中最先想到的写法，但是如果在目前hive的query planner不是那么智能的情况下，想要更加快速的跑出结果，懂一点工具的内部机理也是必须的。

当然了，也有同学有其它的思路，只是没有上面那么高效：

select count(*) from
(
    select user_id,
    count(case when blog_owner = ‘a‘ then 1 end) as visit_z,
    count(case when blog_owner = ‘b‘ then 1 end) as visit_l
    from cnblogs_visit_20130801 group by user_id
) t
where visit_z > 0 and visit_l > 0;

7、Refer：

[1] 数据仓库中的SQL性能优化（Hive篇）

http://sunyi514.github.io/2013/09/01/%E6%95%B0%E6%8D%AE%E4%BB%93%E5%BA%93%E4%B8%AD%E7%9A%84sql%E6%80%A7%E8%83%BD%E4%BC%98%E5%8C%96%EF%BC%88hive%E7%AF%87%EF%BC%89/

数据仓库中的 SQL 性能优化（Hive篇）

标签：style   blog   http   color   os   使用   io   ar   for   

原文地址：http://my.oschina.net/leejun2005/blog/308427