一個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/

來自：http://www.cnblogs.com/sunyi514/p/3279957.html