HIVE優化

1.概述
　　繼續《那些年使用Hive踩過的坑》一文中的剩餘部分，本篇博客贅述了在工做中總結Hive的經常使用優化手段和在工做中使用Hive出現的問題。下面開始本篇文章的優化介紹。

2.介紹
　　首先，咱們來看看Hadoop的計算框架特性，在此特性下會衍生哪些問題？

數據量大不是問題，數據傾斜是個問題。
jobs數比較多的做業運行效率相對比較低，好比即便有幾百行的表，若是屢次關聯屢次彙總，產生十幾個jobs，耗時很長。緣由是map reduce做業初始化的時間是比較長的。
sum,count,max,min等UDAF，不怕數據傾斜問題,hadoop在map端的彙總合併優化，使數據傾斜不成問題。
count(distinct ),在數據量大的狀況下，效率較低，若是是多count(distinct )效率更低，由於count(distinct)是按group by 字段分組，按distinct字段排序，通常這種分佈方式是很傾斜的。舉個例子：好比男uv,女uv，像淘寶一天30億的pv，若是按性別分組，分配2個reduce,每一個reduce處理15億數據。
　　面對這些問題，咱們能有哪些有效的優化手段呢？下面列出一些在工做有效可行的優化手段：

好的模型設計事半功倍。
解決數據傾斜問題。
減小job數。
設置合理的map reduce的task數，能有效提高性能。(好比，10w+級別的計算，用160個reduce，那是至關的浪費，1個足夠)。
瞭解數據分佈，本身動手解決數據傾斜問題是個不錯的選擇。set hive.groupby.skewindata=true;這是通用的算法優化，但算法優化有時不能適應特定業務背景，開發人員瞭解業務，瞭解數據，能夠經過業務邏輯精確有效的解決數據傾斜問題。
數據量較大的狀況下，慎用count(distinct)，count(distinct)容易產生傾斜問題。
對小文件進行合併，是行至有效的提升調度效率的方法，假如全部的做業設置合理的文件數，對雲梯的總體調度效率也會產生積極的正向影響。
優化時把握總體，單個做業最優不如總體最優。
　　而接下來，咱們心中應該會有一些疑問，影響性能的根源是什麼？

3.性能低下的根源
　　hive性能優化時，把HiveQL當作M/R程序來讀，即從M/R的運行角度來考慮優化性能，從更底層思考如何優化運算性能，而不只僅侷限於邏輯代碼的替換層面。

　　RAC（Real Application Cluster）真正應用集羣就像一輛機動靈活的小貨車，響應快；Hadoop就像吞吐量巨大的輪船，啓動開銷大，若是每次只作小數量的輸入輸出，利用率將會很低。因此用好Hadoop的首要任務是增大每次任務所搭載的數據量。

　　Hadoop的核心能力是parition和sort，於是這也是優化的根本。

　　觀察Hadoop處理數據的過程，有幾個顯著的特徵：

數據的大規模並非負載重點，形成運行壓力過大是由於運行數據的傾斜。
jobs數比較多的做業運行效率相對比較低，好比即便有幾百行的表，若是屢次關聯對此彙總，產生幾十個jobs，將會須要30分鐘以上的時間且大部分時間被用於做業分配，初始化和數據輸出。M/R做業初始化的時間是比較耗時間資源的一個部分。
在使用SUM，COUNT，MAX，MIN等UDAF函數時，不怕數據傾斜問題，Hadoop在Map端的彙總合併優化過，使數據傾斜不成問題。
COUNT(DISTINCT)在數據量大的狀況下，效率較低，若是多COUNT(DISTINCT)效率更低，由於COUNT(DISTINCT)是按GROUP BY字段分組，按DISTINCT字段排序，通常這種分佈式方式是很傾斜的；好比：男UV，女UV，淘寶一天30億的PV，若是按性別分組，分配2個reduce,每一個reduce處理15億數據。
數據傾斜是致使效率大幅下降的主要緣由，能夠採用多一次 Map/Reduce 的方法， 避免傾斜。
　　最後得出的結論是：避實就虛，用 job 數的增長，輸入量的增長，佔用更多存儲空間，充分利用空閒 CPU 等各類方法，分解數據傾斜形成的負擔。

4.配置角度優化
　　咱們知道了性能低下的根源，一樣，咱們也能夠從Hive的配置解讀去優化。Hive系統內部已針對不一樣的查詢預設定了優化方法，用戶能夠經過調整配置進行控制， 如下舉例介紹部分優化的策略以及優化控制選項。

4.1列裁剪
　　Hive 在讀數據的時候，能夠只讀取查詢中所須要用到的列，而忽略其它列。 例如，如有如下查詢：

SELECT a,b FROM q WHERE e<10;
　　在實施此項查詢中，Q 表有 5 列（a，b，c，d，e），Hive 只讀取查詢邏輯中真實須要 的 3 列 a、b、e，而忽略列 c，d；這樣作節省了讀取開銷，中間表存儲開銷和數據整合開銷。

　　裁剪所對應的參數項爲：hive.optimize.cp=true（默認值爲真）

4.2分區裁剪
　　能夠在查詢的過程當中減小沒必要要的分區。 例如，如有如下查詢：

SELECT * FROM (SELECTT a1,COUNT(1) FROM T GROUP BY a1) subq WHERE subq.prtn=100; #（多餘分區）
SELECT * FROM T1 JOIN (SELECT * FROM T2) subq ON (T1.a1=subq.a2) WHERE subq.prtn=100;
　　查詢語句若將「subq.prtn=100」條件放入子查詢中更爲高效，能夠減小讀入的分區 數目。 Hive 自動執行這種裁剪優化。

　　分區參數爲：hive.optimize.pruner=true（默認值爲真）

4.3JOIN操做
　　在編寫帶有 join 操做的代碼語句時，應該將條目少的表/子查詢放在 Join 操做符的左邊。 由於在 Reduce 階段，位於 Join 操做符左邊的表的內容會被加載進內存，載入條目較少的表 能夠有效減小 OOM（out of memory）即內存溢出。因此對於同一個 key 來講，對應的 value 值小的放前，大的放後，這即是「小表放前」原則。 若一條語句中有多個 Join，依據 Join 的條件相同與否，有不一樣的處理方法。

4.3.1JOIN原則
　　在使用寫有 Join 操做的查詢語句時有一條原則：應該將條目少的表/子查詢放在 Join 操做符的左邊。緣由是在 Join 操做的 Reduce 階段，位於 Join 操做符左邊的表的內容會被加載進內存，將條目少的表放在左邊，能夠有效減小發生 OOM 錯誤的概率。對於一條語句中有多個 Join 的狀況，若是 Join 的條件相同，好比查詢：

INSERT OVERWRITE TABLE pv_users
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
一個 Map-Reduce 任務，而不是 ‘n’ 個
在作 OUTER JOIN 的時候也是同樣
　　若是 Join 的條件不相同，好比：

INSERT OVERWRITE TABLE pv_users
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);
　　Map-Reduce 的任務數目和 Join 操做的數目是對應的，上述查詢和如下查詢是等價的：

INSERT OVERWRITE TABLE tmptable
SELECT * FROM page_view p JOIN user u
ON (pv.userid = u.userid);
INSERT OVERWRITE TABLE pv_users
SELECT x.pageid, x.age FROM tmptable x
JOIN newuser y ON (x.age = y.age);
4.4MAP JOIN操做
　　Join 操做在 Map 階段完成，再也不須要Reduce，前提條件是須要的數據在 Map 的過程當中能夠訪問到。好比查詢：

INSERT OVERWRITE TABLE pv_users
SELECT /+ MAPJOIN(pv) / pv.pageid, u.age
FROM page_view pv
JOIN user u ON (pv.userid = u.userid);
　　能夠在 Map 階段完成 Join，如圖所示：

　　相關的參數爲：

hive.join.emit.interval = 1000
hive.mapjoin.size.key = 10000
hive.mapjoin.cache.numrows = 10000
4.5GROUP BY操做
　　進行GROUP BY操做時須要注意一下幾點：

Map端部分聚合
　　事實上並非全部的聚合操做都須要在reduce部分進行，不少聚合操做均可以先在Map端進行部分聚合，而後reduce端得出最終結果。

　　這裏須要修改的參數爲：

　　hive.map.aggr=true（用於設定是否在 map 端進行聚合，默認值爲真） hive.groupby.mapaggr.checkinterval=100000（用於設定 map 端進行聚合操做的條目數）

有數據傾斜時進行負載均衡
　　此處須要設定 hive.groupby.skewindata，當選項設定爲 true 是，生成的查詢計劃有兩 個 MapReduce 任務。在第一個 MapReduce 中，map 的輸出結果集合會隨機分佈到 reduce 中， 每一個 reduce 作部分聚合操做，並輸出結果。這樣處理的結果是，相同的 Group By Key 有可 能分發到不一樣的 reduce 中，從而達到負載均衡的目的；第二個 MapReduce 任務再根據預處 理的數據結果按照 Group By Key 分佈到 reduce 中（這個過程能夠保證相同的 Group By Key 分佈到同一個 reduce 中），最後完成最終的聚合操做。

4.6合併小文件
　　咱們知道文件數目小，容易在文件存儲端形成瓶頸，給 HDFS 帶來壓力，影響處理效率。對此，能夠經過合併Map和Reduce的結果文件來消除這樣的影響。

　　用於設置合併屬性的參數有：

是否合併Map輸出文件：hive.merge.mapfiles=true（默認值爲真）
是否合併Reduce 端輸出文件：hive.merge.mapredfiles=false（默認值爲假）
合併文件的大小：hive.merge.size.per.task=25610001000（默認值爲 256000000）
5.程序角度優化
5.1熟練使用SQL提升查詢
　　熟練地使用 SQL，能寫出高效率的查詢語句。

　　場景：有一張 user 表，爲賣家天天收到表，user_id，ds（日期）爲 key，屬性有主營類目，指標有交易金額，交易筆數。天天要取前10天的總收入，總筆數，和最近一天的主營類目。 　　解決方法 1

　　以下所示：經常使用方法

複製代碼
INSERT OVERWRITE TABLE t1
SELECT user_id,substr(MAX(CONCAT(ds,cat),9) AS main_cat) FROM users
WHERE ds=20120329 // 20120329 爲日期列的值，實際代碼中能夠用函數表示出當天日期 GROUP BY user_id;

INSERT OVERWRITE TABLE t2
SELECT user_id,sum(qty) AS qty,SUM(amt) AS amt FROM users
WHERE ds BETWEEN 20120301 AND 20120329
GROUP BY user_id

SELECT t1.user_id,t1.main_cat,t2.qty,t2.amt FROM t1
JOIN t2 ON t1.user_id=t2.user_id
複製代碼
　　下面給出方法1的思路，實現步驟以下：

　　第一步：利用分析函數，取每一個 user_id 最近一天的主營類目，存入臨時表 t1。

　　第二步：彙總 10 天的總交易金額，交易筆數，存入臨時表 t2。

　　第三步：關聯 t1，t2，獲得最終的結果。

　　解決方法 2

　　以下所示：優化方法　

SELECT user_id,substr(MAX(CONCAT(ds,cat)),9) AS main_cat,SUM(qty),SUM(amt) FROM users
WHERE ds BETWEEN 20120301 AND 20120329
GROUP BY user_id
　　在工做中咱們總結出：方案 2 的開銷等於方案 1 的第二步的開銷，性能提高，由原有的 25 分鐘完成，縮短爲 10 分鐘之內完成。節省了兩個臨時表的讀寫是一個關鍵緣由，這種方式也適用於 Oracle 中的數據查找工做。

SQL 具備普適性，不少 SQL 通用的優化方案在 Hadoop 分佈式計算方式中也能夠達到效果。

5.2無效ID在關聯時的數據傾斜問題
　　問題：日誌中常會出現信息丟失，好比每日約爲 20 億的全網日誌，其中的 user_id 爲主 鍵，在日誌收集過程當中會丟失，出現主鍵爲 null 的狀況，若是取其中的 user_id 和 bmw_users 關聯，就會碰到數據傾斜的問題。緣由是 Hive 中，主鍵爲 null 值的項會被當作相同的 Key 而分配進同一個計算 Map。

解決方法 1：user_id 爲空的不參與關聯，子查詢過濾 null

SELECT * FROM log a
JOIN bmw_users b ON a.user_id IS NOT NULL AND a.user_id=b.user_id
UNION All SELECT * FROM log a WHERE a.user_id IS NULL
　　解決方法 2 以下所示：函數過濾 null

SELECT * FROM log a LEFT OUTER
JOIN bmw_users b ON
CASE WHEN a.user_id IS NULL THEN CONCAT(‘dp_hive’,RAND()) ELSE a.user_id END =b.user_id;
　　調優結果：原先因爲數據傾斜致使運行時長超過 1 小時，解決方法 1 運行每日平均時長 25 分鐘，解決方法 2 運行的每日平均時長在 20 分鐘左右。優化效果很明顯。

　　咱們在工做中總結出：解決方法2比解決方法1效果更好，不但IO少了，並且做業數也少了。解決方法1中log讀取兩次，job 數爲2。解決方法2中 job 數是1。這個優化適合無效 id（好比-9九、 ‘’，null 等）產生的傾斜問題。把空值的 key 變成一個字符串加上隨機數，就能把傾斜的 數據分到不一樣的Reduce上，從而解決數據傾斜問題。由於空值不參與關聯，即便分到不一樣 的 Reduce 上，也不會影響最終的結果。附上 Hadoop 通用關聯的實現方法是：關聯經過二次排序實現的，關聯的列爲 partion key，關聯的列和表的 tag 組成排序的 group key，根據 pariton key分配Reduce。同一Reduce內根據group key排序。

5.3不一樣數據類型關聯產生的傾斜問題
　　問題：不一樣數據類型 id 的關聯會產生數據傾斜問題。

　　一張表 s8 的日誌，每一個商品一條記錄，要和商品表關聯。但關聯卻碰到傾斜的問題。 s8 的日誌中有 32 爲字符串商品 id，也有數值商品 id，日誌中類型是 string 的，但商品中的 數值 id 是 bigint 的。猜測問題的緣由是把 s8 的商品 id 轉成數值 id 作 hash 來分配 Reduce， 因此字符串 id 的 s8 日誌，都到一個 Reduce 上了，解決的方法驗證了這個猜想。

　　解決方法：把數據類型轉換成字符串類型

　

SELECT * FROM s8_log a LEFT OUTER
JOIN r_auction_auctions b ON a.auction_id=CASE(b.auction_id AS STRING)
　　調優結果顯示：數據表處理由 1 小時 30 分鐘經代碼調整後能夠在 20 分鐘內完成。

5.4利用Hive對UNION ALL優化的特性
　　多表 union all 會優化成一個 job。

　　問題：好比推廣效果表要和商品表關聯，效果表中的 auction_id 列既有 32 爲字符串商 品 id，也有數字 id，和商品表關聯獲得商品的信息。

　　解決方法：Hive SQL 性能會比較好

SELECT * FROM effect a
JOIN
(SELECT auction_id AS auction_id FROM auctions
UNION All
SELECT auction_string_id AS auction_id FROM auctions) b
ON a.auction_id=b.auction_id
　　比分別過濾數字 id，字符串 id 而後分別和商品表關聯性能要好。

　　這樣寫的好處：1 個 MapReduce 做業，商品表只讀一次，推廣效果表只讀取一次。把 這個 SQL 換成 Map/Reduce 代碼的話，Map 的時候，把 a 表的記錄打上標籤 a，商品表記錄 每讀取一條，打上標籤 b，變成兩個<key,value>對，<(b,數字 id),value>，<(b,字符串 id),value>。

　　因此商品表的 HDFS 讀取只會是一次。

5.5解決Hive對UNION ALL優化的短板
　　Hive 對 union all 的優化的特性：對 union all 優化只侷限於非嵌套查詢。

消滅子查詢內的 group by
示例 1：子查詢內有 group by

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
　　從業務邏輯上說，子查詢內的 GROUP BY 怎麼都看顯得多餘（功能上的多餘，除非有 COUNT(DISTINCT)），若是不是由於 Hive Bug 或者性能上的考量（曾經出現若是不執行子查詢 GROUP BY，數據得不到正確的結果的 Hive Bug）。因此這個 Hive 按經驗轉換成以下所示：

SELECT * FROM (SELECT * FROM t1 UNION ALL SELECT * FROM t2)t3 GROUP BY c1,c2,c3
　　調優結果：通過測試，並未出現 union all 的 Hive Bug，數據是一致的。MapReduce 的 做業數由 3 減小到 1。

t1 至關於一個目錄，t2 至關於一個目錄，對 Map/Reduce 程序來講，t1，t2 能夠做爲 Map/Reduce 做業的 mutli inputs。這能夠經過一個 Map/Reduce 來解決這個問題。Hadoop 的 計算框架，不怕數據多，就怕做業數多。

　　但若是換成是其餘計算平臺如 Oracle，那就不必定了，由於把大的輸入拆成兩個輸入， 分別排序彙總後 merge（假如兩個子排序是並行的話），是有可能性能更優的（好比希爾排 序比冒泡排序的性能更優）。

消滅子查詢內的 COUNT(DISTINCT)，MAX，MIN。
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)操做，直接去 GROUP BY 將達不到業務目標。這時採用 臨時表消滅 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;
　　job 數是 2，減小一半，並且兩次 Map/Reduce 比 COUNT(DISTINCT)效率更高。

調優結果：千萬級別的類目表，member 表，與 10 億級得商品表關聯。原先 1963s 的任務通過調整，1152s 即完成。

消滅子查詢內的 JOIN
SELECT * FROM
(SELECT * FROM t1 UNION ALL SELECT * FROM t4 UNION ALL SELECT * FROM t2 JOIN t3 ON t2.id=t3.id) x
GROUP BY c1,c2;
　　上面代碼運行會有 5 個 jobs。加入先 JOIN 生存臨時表的話 t5，而後 UNION ALL，會變成 2 個 jobs。

INSERT OVERWRITE TABLE t5
SELECT * FROM t2 JOIN t3 ON t2.id=t3.id;
SELECT * FROM (t1 UNION ALL t4 UNION ALL t5);
　　調優結果顯示：針對千萬級別的廣告位表，由原先 5 個 Job 共 15 分鐘，分解爲 2 個 job 一個 8-10 分鐘，一個3分鐘。

5.6GROUP BY替代COUNT(DISTINCT)達到優化效果
　　計算 uv 的時候，常常會用到 COUNT(DISTINCT)，但在數據比較傾斜的時候 COUNT(DISTINCT) 會比較慢。這時能夠嘗試用 GROUP BY 改寫代碼計算 uv。

原有代碼
INSERT OVERWRITE TABLE s_dw_tanx_adzone_uv PARTITION (ds=20120329)
SELECT 20120329 AS thedate,adzoneid,COUNT(DISTINCT acookie) AS uv FROM s_ods_log_tanx_pv t WHERE t.ds=20120329 GROUP BY adzoneid
　　關於COUNT(DISTINCT)的數據傾斜問題不能一律而論，要依狀況而定，下面是我測試的一組數據：

　　測試數據：169857條

統計每日IP

CREATE TABLE ip_2014_12_29 AS SELECT COUNT(DISTINCT ip) AS IP FROM logdfs WHERE logdate='2014_12_29';
耗時：24.805 seconds

統計每日IP（改造）

CREATE TABLE ip_2014_12_29 AS SELECT COUNT(1) AS IP FROM (SELECT DISTINCT ip from logdfs WHERE logdate='2014_12_29') tmp;
耗時：46.833 seconds
　　測試結果表名：明顯改造後的語句比以前耗時，這是由於改造後的語句有2個SELECT，多了一個job，這樣在數據量小的時候，數據不會存在傾斜問題。

6.優化總結
　　優化時，把hive sql當作mapreduce程序來讀，會有意想不到的驚喜。理解hadoop的核心能力，是hive優化的根本。這是這一年來，項目組全部成員寶貴的經驗總結。

長期觀察hadoop處理數據的過程，有幾個顯著的特徵:
不怕數據多，就怕數據傾斜。
對jobs數比較多的做業運行效率相對比較低，好比即便有幾百行的表，若是屢次關聯屢次彙總，產生十幾個jobs，沒半小時是跑不完的。map reduce做業初始化的時間是比較長的。
對sum，count來講，不存在數據傾斜問題。
對count(distinct ),效率較低，數據量一多，準出問題，若是是多count(distinct )效率更低。
優化能夠從幾個方面着手：
好的模型設計事半功倍。
解決數據傾斜問題。
減小job數。
設置合理的map reduce的task數，能有效提高性能。(好比，10w+級別的計算，用160個reduce，那是至關的浪費，1個足夠)。
本身動手寫sql解決數據傾斜問題是個不錯的選擇。set hive.groupby.skewindata=true;這是通用的算法優化，但算法優化老是漠視業務，習慣性提供通用的解決方法。 Etl開發人員更瞭解業務，更瞭解數據，因此經過業務邏輯解決傾斜的方法每每更精確，更有效。
對count(distinct)採起漠視的方法，尤爲數據大的時候很容易產生傾斜問題，不抱僥倖心理。本身動手，豐衣足食。
對小文件進行合併，是行至有效的提升調度效率的方法，假如咱們的做業設置合理的文件數，對雲梯的總體調度效率也會產生積極的影響。
　　優化時把握總體，單個做業最優不如總體最優。

7.優化的經常使用手段
　　主要由三個屬性來決定：

hive.exec.reducers.bytes.per.reducer ＃這個參數控制一個job會有多少個reducer來處理，依據的是輸入文件的總大小。默認1GB。 hive.exec.reducers.max ＃這個參數控制最大的reducer的數量， 若是 input / bytes per reduce > max 則會啓動這個參數所指定的reduce個數。 這個並不會影響mapre.reduce.tasks參數的設置。默認的max是999。 mapred.reduce.tasks ＃這個參數若是指定了，hive就不會用它的estimation函數來自動計算reduce的個數，而是用這個參數來啓動reducer。默認是-1。 7.1參數設置的影響 　　若是reduce太少：若是數據量很大，會致使這個reduce異常的慢，從而致使這個任務不能結束，也有可能會OOM 二、若是reduce太多： 產生的小文件太多，合併起來代價過高，namenode的內存佔用也會增大。若是咱們不指定mapred.reduce.tasks， hive會自動計算須要多少個reducer。