一例 Hive join 優化實戰

因爲 hive 與傳統關係型數據庫面對的業務場景及底層技術架構都有着很大差別，所以，傳統數據庫領域的一些技能放到 Hive 中可能已再也不適用。關於 hive 的優化與原理、應用的文章，前面也陸陸續續的介紹了一些，但大多都偏向理論層面，本文就介紹一個實例，從實例中一步步加深對 hive 調優的認識與意識。

一、需求

需求我作了簡化，很簡單，兩張表作個 join，求指定城市，天天的 pv，用傳統的 RDBMS SQL 寫出來就這樣的：

SELECT t.statdate,
       c.cname,
       count(t.cookieid)
FROM tmpdb.city c
JOIN ecdata.ext_trackflow t ON (t.area1= c.cname
                                OR t.area2 =c.cname
                                OR t.area3 = c.cname)
WHERE t.statdate>='20140818' and t.statdate<='20140824'
  AND platform='pc'
GROUP BY t.statdate,
         c.cname;

怎麼樣？根據 SQL 看懂需求沒問題吧？

二、非等值 join 問題

而後把這條 SQL 貼到 hive 中去執行，而後你會發現報錯了：

FAILED: SemanticException [Error 10019]: Line 5:32 OR not supported in JOIN currently 'cname'

這是由於 hive 受限於 MapReduce 算法模型，只支持 equi-joins（等值 join），要實現上述的非等值 join，你能夠採用笛卡兒積（ full Cartesian product ）來實現：

SELECT t.statdate,
       c.cname,
       count(t.cookieid)
FROM tmpdb.city c
JOIN ecdata.ext_trackflow t
WHERE t.statdate>='20140818'
  AND t.statdate<='20140824'
  AND platform='pc'
  AND (t.area1= c.cname
       OR t.area2 =c.cname
       OR t.area3 = c.cname)
GROUP BY t.statdate,
         c.cname;

而後再拿着這條語句執行下。

三、優化：reduce side join VS Cartesian product

若是你真的把這條語句放到 Hive 上執行，而後剛好你有張表還很是大，那麼恭喜你。。。集羣管理員估計會找你的麻煩了。。。

友情提示：笛卡兒積這種語句在 Hive 下慎用，大數據場景下的 m * n 映射結果你懂的。。。對此，Hive 特地提供了一個環境變量：hive.mapred.mode=strict; 防止笛卡兒積的執行：

FAILED: SemanticException [Error 10052]: In strict mode, cartesian product is not allowed. If you really want to perform the operation, set hive.mapred.mode=nonstrict

從 2 中的觀察得知咱們在 on 後面跟 join 條件，走的是 reduce side join，若是你在 where 後跟則是走 Cartesian product，可是這裏單條 sql 又無法實現 reduce side join，還有沒有其它辦法呢？

四、改寫非等值 join：union all

既然不容許非等值 join，那咱們換一下思路，多個子查詢 union all，而後彙總：

SELECT dt,
       name,
       count(cid)
FROM
  (SELECT t.statdate dt,
          c.cname name,
          t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area1 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT t.statdate dt,
                    c.cname name,
                    t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area2 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT t.statdate dt,
                    c.cname name,
                    t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area3 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc') tmp_trackflow
GROUP BY dt,
         name;

五、優化：map side join

上述語句走的是 reduce side join，從咱們的需求及業務得知，tmpdb.city 是一張字典表，數據量很小，所以咱們能夠試試把上述的語句改寫成 mapjoin：

SELECT dt,
       name,
       count(cid)
FROM
  (SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                            c.cname name,
                            t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area1 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                                      c.cname name,
                                      t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area2 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                                      c.cname name,
                                      t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area3 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc') tmp_trackflow
GROUP BY dt,
         name;

六、優化無極限：開啓 parallel 和 控制 reduce 個數

上述語句執行時，你能夠看到執行計劃和狀態信息，以及結合你的 union all 語句可知，三個 union 語句之間沒有依賴關係，實際上是能夠並行執行的：

explain SQL...
...
STAGE DEPENDENCIES:
  Stage-11 is a root stage
  Stage-1 depends on stages: Stage-11
  Stage-2 depends on stages: Stage-1
  Stage-3 depends on stages: Stage-2, Stage-6, Stage-9
  Stage-12 is a root stage
  Stage-5 depends on stages: Stage-12
  Stage-6 depends on stages: Stage-5
  Stage-13 is a root stage
  Stage-8 depends on stages: Stage-13
  Stage-9 depends on stages: Stage-8
  Stage-0 is a root stage
...

咱們在 SQL 前加上以下環境變量選項：

set mapred.reduce.tasks=60;
set hive.exec.parallel=true;

讓執行計劃中的 Stage-十一、Stage-十二、Stage-13 並行執行，並控制好 reduce task 個數。

完整的語句以下：

hive -e "
SET mapred.reduce.tasks=60;


SET hive.exec.parallel=TRUE;


SELECT dt,
       name,
       count(cid)
FROM
  (SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                            c.cname name,
                            t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area1 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                                      c.cname name,
                                      t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area2 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc'
   UNION ALL SELECT /*+ MAPJOIN(c) */ t.statdate dt,
                                      c.cname name,
                                      t.cookieid cid
   FROM tmpdb.city c
   JOIN ecdata.ext_trackflow t ON t.area3 =c.cname
   WHERE t.statdate>='20140818'
     AND t.statdate<='20140824'
     AND platform='pc') tmp_trackflow
GROUP BY dt,
         name;

" > a1.txt

最後的優化效果是：2 中的語句三個小時沒出結果。。。5 比 4 快 8 倍左右，6 比 5 快 2 倍左右，最終 10min 出結果。

七、最後的問題：

在 6 的語句執行的時候你會發現，其掃描了 三遍 源文件。而 hive 自己是對 union all 的 join 作了優化的，當多個 union all 子查詢同一張表時，只掃描一次源文件，但這裏爲何會三個子查詢各掃描一次呢？

多是這裏的 union all 子查詢使用了 join 的緣故，致使 hive 的 union all 執行計劃優化失效了。

關於這塊怎麼能優化成只掃描一次源文件，或者你有更好的優化方案，歡迎留言交流。

八、關於 hive 中的 笛卡爾集（ full Cartesian product ）

在JION接連查詢中沒有ON鏈接key，而經過WHERE條件語句會產生笛卡爾集。
Hive自己是不支持笛卡爾集的，不能用select T1.*, T2.* from table1, table2這種語法。但有時候確實須要用到笛卡爾集的時候，能夠用下面的語法來實現一樣的效果：
select T1.*, T2.* from table1 T1 join table2 T2 where 1=1;
注意在Hive的Strict模式下不能用這種語法，由於這樣會產生笛卡爾集，而這種模式禁止產生笛卡爾集。須要先用set hive.mapred.mode=nonstrict;設爲非strict模式就能夠用了，或者將where改成on鏈接。
select T1.*, T2.* from table1 T1 join table2 T2 on  T1.id=T2.id;

九、關於Strict Mode

Hive中的嚴格模式能夠防止用戶發出（能夠有問題）的查詢無心中形成不良的影響。 將hive.mapred.mode設置成strict能夠禁止三種類型的查詢：
1）、在一個分區表上，若是沒有在WHERE條件中指明具體的分區，那麼這是不容許的，換句話說，不容許在分區表上全表掃描。這種限制的緣由是分區表一般會持很是大的數據集而且可能數據增加迅速，對這樣的一個大表作全表掃描會消耗大量資源，必需要再WHERE過濾條件中具體指明分區才能夠執行成功的查詢。
2）、第二種是禁止執行有ORDER BY的排序要求但沒有LIMIT語句的HiveQL查詢。由於ORDER BY全局查詢會致使有一個單一的reducer對全部的查詢結果排序，若是對大數據集作排序，這將致使不可預期的執行時間，必需要加上limit條件才能夠執行成功的查詢。
3）、第三種是禁止產生笛卡爾集。在JION接連查詢中沒有ON鏈接key而經過WHERE條件語句會產生笛卡爾集，須要改成JOIN...ON語句。

十、Refer：

[1] Hive Query- Joining two tables on three joining conditions with OR operator

http://stackoverflow.com/questions/16272804/hive-query-joining-two-tables-on-three-joining-conditions-with-or-operator

[2] LanguageManual JoinOptimization

https://cwiki.apache.org/confluence/display/Hive/LanguageManual+JoinOptimization

[3] hive 執行計劃

http://yychao.iteye.com/blog/1749562

[4] Hive SQL解析/執行計劃生成流程分析

http://yanbohappy.sinaapp.com/?p=265

[5] 數據倉庫中的SQL性能優化（Hive篇）

http://www.zihou.me/html/2014/02/12/9207.html

[6] Hive優化以及執行原理

http://www.smartcitychina.cn/upload/2014-01/14012015376829.pdf

[7] Hive做業優化總結

http://my.oschina.net/yangzhiyuan/blog/262910

[8] Hive鏈接產生笛卡爾集

http://blog.javachen.com/2013/10/17/cartesian-product-in-hive-inner-join/#