Hive 自0.14.0開始，加入了一項”Cost based Optimizer”來對HQL執行計劃進行優化，這個功能通過”hive.cbo.enable”來開啟。在Hive 1.1.0之後，這個feature是預設開啟的,它可以自動優化HQL中多個JOIN的順序，並選擇合適的JOIN演算法

Join reordering and join algorithm selection are few of the optimizations that can benefit from a cost based optimizer. Cost based optimizer would free up user from having to rearrange joins in the right order or from having to specify join algorithm by using query hints and configuration options. This can potentially free up users to model their reporting and ETL needs close to business process without having to worry about query optimizations.

不過在工作中，發現這個功能在某些情況下反而會導致Query的執行效率降低，這裡寫個blog記錄一下。以下僅針對Hive on tez , 至於 Hive on mapreduce 的情況待考察。

假如我們有三張表:
1. 事實表 table_fact

2. 維度表 table_dim1

3. 維度表 table_dim2

這是一個常見的星型模型的縮影，由一個事實表和兩個維度表組成。通常我們在查詢時需要從維度表中關聯出我們需要的資訊，比如:

select 
    d1.field_res_1,
    d2.field_res_2,
    sum(f.score) 
from 
    table_fact f 
left join table_dim_1 d1 on f.field_dim_1=d1.field_dim_1 
left join table_dim_2 d2 on f.field_dim_1=d2.field_dim_1 and f.field_dim_2=d2.field_dim_2 
where 
    f.field_dim_2="abc" 
group by 
    d1.field_res_1,
    d2.field_res_2;
 

這是一個比較常見的查詢語句。在cbo開關開啟的情況下，執行計劃是這個樣子的:

+----------------------------------------------------+
|                      Explain                       |
+----------------------------------------------------+
| Plan optimized by CBO.                             |
|                                                    |
| Vertex dependency in root stage                    |
| Reducer 2 <- Map 1 (SIMPLE_EDGE), Map 4 (SIMPLE_EDGE), Map 5 (SIMPLE_EDGE) |
| Reducer 3 <- Reducer 2 (SIMPLE_EDGE)               |
|                                                    |
| Stage-0                                            |
|   Fetch Operator                                   |
|     limit:-1                                       |
|     Stage-1                                        |
|       Reducer 3                                    |
|       File Output Operator [FS_17]                 |
|         Group By Operator [GBY_15] (rows=1 width=0) |
|           Output:["_col0","_col1","_col2"],aggregations:["sum(VALUE._col0)"],keys:KEY._col0, KEY._col1 |
|         <-Reducer 2 [SIMPLE_EDGE]                  |
|           SHUFFLE [RS_14]                          |
|             PartitionCols:_col0, _col1             |
|             Group By Operator [GBY_13] (rows=2 width=0) |
|               Output:["_col0","_col1","_col2"],aggregations:["sum(_col2)"],keys:_col4, _col7 |
|               Select Operator [SEL_12] (rows=2 width=0) |
|                 Output:["_col4","_col7","_col2"]   |
|                 Merge Join Operator [MERGEJOIN_23] (rows=2 width=0) |
|                   Conds:RS_8._col0=RS_9._col0(Left Outer),RS_8._col0=RS_10._col0(Left Outer),Output:["_col2","_col4","_col7"] |
|                 <-Map 1 [SIMPLE_EDGE]              |
|                   SHUFFLE [RS_8]                   |
|                     PartitionCols:_col0            |
|                     Select Operator [SEL_2] (rows=1 width=0) |
|                       Output:["_col0","_col2"]     |
|                       Filter Operator [FIL_20] (rows=1 width=0) |
|                         predicate:(field_dim_2 = 'abc') |
|                         TableScan [TS_0] (rows=1 width=0) |
|                           [email protected]_fact,f,Tbl:PARTIAL,Col:NONE,Output:["field_dim_1","field_dim_2","score"] |
|                 <-Map 4 [SIMPLE_EDGE]              |
|                   SHUFFLE [RS_9]                   |
|                     PartitionCols:_col0            |
|                     Select Operator [SEL_4] (rows=1 width=0) |
|                       Output:["_col0","_col1"]     |
|                       TableScan [TS_3] (rows=1 width=0) |
|                         [email protected]_dim_1,d1,Tbl:PARTIAL,Col:NONE,Output:["field_dim_1","field_res_1"] |
|                 <-Map 5 [SIMPLE_EDGE]              |
|                   SHUFFLE [RS_10]                  |
|                     PartitionCols:_col0            |
|                     Select Operator [SEL_7] (rows=1 width=0) |
|                       Output:["_col0","_col2"]     |
|                       Filter Operator [FIL_22] (rows=1 width=0) |
|                         predicate:('abc' = field_dim_2) |
|                         TableScan [TS_5] (rows=1 width=0) |
|                           [email protected]_dim_2,d2,Tbl:PARTIAL,Col:NONE,Output:["field_dim_1","field_dim_2","field_res_2"] |
|                                                    |
+----------------------------------------------------+

我們可以看到，line 4提示這個查詢經過了CBO的優化。hive對這個SQL做了一系列的優化(不僅僅是CBO的優化):

• filter的下推(line 33,49)
• projection的下推(line 31,40,47)
• 對join key的優化 (line 26等)
• 儘可能的合併了MR任務(line 7、8)

其他的優化不一一列舉了。這個執行計劃中將2個join用一次map-reduce就解決了，最後一個Reduce-3是為了做group by的。

但是如果事實表 table_fact 中的欄位 field_dim_1 存在資料傾斜的話會怎麼樣呢?我們可以看到，在這個執行計劃中，shuffle的key都是針對 field_dim_1的，如果這個欄位存在大量的資料傾斜的話，那麼這個查詢會執行的非常緩慢。

針對資料傾斜，我們通常的解決方法無外乎就是 mapjoin、 salting ,或者做任務的拆分等幾種方法。這裡的查詢是一張較大的事實表和兩張維度表進行的關聯。如果維度表(或者經過filter && projection)之後的維度表足夠小的話，那麼這裡實際上hive是可以自動將這個查詢轉化為 map side join 的。

由於我為了說明這個情景，以上的三張表建立的都是空表，所以理論上hive應該會把這個查詢轉化為 map side join 。但是為什麼沒有呢? 這就是cbo的關係了。

+----------------------------------------------------+
|                      Explain                       |
+----------------------------------------------------+
| Vertex dependency in root stage                    |
| Map 1 <- Map 3 (BROADCAST_EDGE), Map 4 (BROADCAST_EDGE) |
| Reducer 2 <- Map 1 (SIMPLE_EDGE)                   |
|                                                    |
| Stage-0                                            |
|   Fetch Operator                                   |
|     limit:-1                                       |
|     Stage-1                                        |
|       Reducer 2                                    |
|       File Output Operator [FS_15]                 |
|         Group By Operator [GBY_13] (rows=1 width=0) |
|           Output:["_col0","_col1","_col2"],aggregations:["sum(VALUE._col0)"],keys:KEY._col0, KEY._col1 |
|         <-Map 1 [SIMPLE_EDGE]                      |
|           SHUFFLE [RS_12]                          |
|             PartitionCols:_col0, _col1             |
|             Group By Operator [GBY_11] (rows=1 width=0) |
|               Output:["_col0","_col1","_col2"],aggregations:["sum(_col2)"],keys:_col7, _col13 |
|               Select Operator [SEL_10] (rows=1 width=0) |
|                 Output:["_col7","_col13","_col2"]  |
|                 Map Join Operator [MAPJOIN_22] (rows=1 width=0) |
|                   Conds:MAPJOIN_21._col0, _col1=RS_7.field_dim_1, field_dim_2(Left Outer),HybridGraceHashJoin:true,Output:["_col2","_col7","_col13"] |
|                 <-Map 4 [BROADCAST_EDGE]           |
|                   BROADCAST [RS_7]                 |
|                     PartitionCols:field_dim_1, field_dim_2 |
|                     Filter Operator [FIL_20] (rows=1 width=0) |
|                       predicate:(field_dim_2 = 'abc') |
|                       TableScan [TS_2] (rows=1 width=0) |
|                         [email protected]_dim_2,d2,Tbl:PARTIAL,Col:NONE,Output:["field_dim_1","field_dim_2","field_res_2"] |
|                 <-Map Join Operator [MAPJOIN_21] (rows=1 width=0) |
|                     Conds:FIL_18.field_dim_1=RS_4.field_dim_1(Left Outer),HybridGraceHashJoin:true,Output:["_col0","_col1","_col2","_col7"] |
|                   <-Map 3 [BROADCAST_EDGE]         |
|                     BROADCAST [RS_4]               |
|                       PartitionCols:field_dim_1    |
|                       TableScan [TS_1] (rows=1 width=0) |
|                         [email protected]_dim_1,d1,Tbl:PARTIAL,Col:NONE,Output:["field_dim_1","field_res_1"] |
|                   <-Filter Operator [FIL_18] (rows=1 width=0) |
|                       predicate:(field_dim_2 = 'abc') |
|                       TableScan [TS_0] (rows=1 width=0) |
|                         [email protected]_fact,f,Tbl:PARTIAL,Col:NONE,Output:["field_dim_1","field_dim_2","score"] |
|                                                    |
+----------------------------------------------------+

首先CBO優化的標誌不見了(line 4),然後 Stage1 針對兩個維表的 Map 操作變成了 BROADCAST (line 5) ，並且輸出結果傳給了針對事實表的 Map .

由此可以看到關閉CBO之後， Hive會使用map join來執行這個查詢,另外整個查詢的shuffle次數還從2次減少到了1次，即使在沒有資料傾斜的情況下，查詢效率也很可能會快很多。

工作中實際查詢的SQL很複雜，就不貼了。不過查詢瓶頸最後抽象出來基本就是上面的情形，而且關聯欄位的某個值傾斜非常嚴重，整個查詢持續了數個小時也沒有會結束的跡象。當我們把CBO關閉以後，Hive將這個查詢優化為了 Map Side Join,很快就返回了結果。

由此可見，一些優化的開關也未必是開啟就好，小心過度優化