Hive性能優化（全面）

作者：浪尖 原文連結 本文轉載自公衆号：Spark學習技巧

1.介紹

首先，我們來看看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)容易産生傾斜問題。
• 對小檔案進行合并，是行至有效的提高排程效率的方法，假如所有的作業設定合理的檔案數，對雲梯的整體排程效率也會産生積極的正向影響。
• 優化時把握整體，單個作業最優不如整體最優。

而接下來，我們心中應該會有一些疑問，影響性能的根源是什麼？

2.性能低下的根源

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 等各種方法，分解資料傾斜造成的負擔。

3.配置角度優化

我們知道了性能低下的根源，同樣，我們也可以從Hive的配置解讀去優化。Hive系統内部已針對不同的查詢預設定了優化方法，使用者可以通過調整配置進行控制， 以下舉例介紹部分優化的政策以及優化控制選項。

3.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（預設值為真）

3.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（預設值為真）

3.3JOIN操作

在編寫帶有 join 操作的代碼語句時，應該将條目少的表/子查詢放在 Join 操作符的左邊。因為在 Reduce 階段，位于 Join 操作符左邊的表的内容會被加載進記憶體，載入條目較少的表 可以有效減少 OOM（out of memory）即記憶體溢出。是以對于同一個 key 來說，對應的 value 值小的放前，大的放後，這便是“小表放前”原則。若一條語句中有多個 Join，依據 Join 的條件相同與否，有不同的處理方法。

3.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);
           

3.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

3.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 中），最後完成最終的聚合操作。

3.6合并小檔案

我們知道檔案數目小，容易在檔案存儲端造成瓶頸，給 HDFS 帶來壓力，影響處理效率。對此，可以通過合并Map和Reduce的結果檔案來消除這樣的影響。

用于設定合并屬性的參數有：

• 是否合并Map輸出檔案：hive.merge.mapfiles=true（預設值為真）
• 是否合并Reduce 端輸出檔案：hive.merge.mapredfiles=false（預設值為假）
• 合并檔案的大小：hive.merge.size.per.task=25610001000（預設值為 256000000）

4.程式角度優化

4.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 分布式計算方式中也可以達到效果。

4.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（比如-99、 ‘’，null 等）産生的傾斜問題。把空值的 key 變成一個字元串加上随機數，就能把傾斜的 資料分到不同的Reduce上，進而解決資料傾斜問題。因為空值不參與關聯，即使分到不同 的 Reduce 上，也不會影響最終的結果。附上 Hadoop 通用關聯的實作方法是：關聯通過二次排序實作的，關聯的列為 partion key，關聯的列和表的 tag 組成排序的 group key，根據 pariton key配置設定Reduce。同一Reduce内根據group key排序。

4.3不同資料類型關聯産生的傾斜問題

問題：不同資料類型 id 的關聯會産生資料傾斜問題。

一張表 s8 的日志，每個商品一條記錄，要和商品表關聯。但關聯卻碰到傾斜的問題。s8 的日志中有 32 為字元串商品 id，也有數值商品 id，日志中類型是 string 的，但商品中的 數值 id 是 bigint 的。猜想問題的原因是把 s8 的商品 id 轉成數值 id 做 hash 來配置設定 Reduce， 是以字元串 id 的 s8 日志，都到一個 Reduce 上了，解決的方法驗證了這個猜測。

解決方法：把資料類型轉換成字元串類型

SELECT * FROM s8_log a LEFT OUTERJOIN r_auction_auctions b ON a.auction_id=CASE(b.auction_id AS STRING)           

調優結果顯示：資料表處理由 1 小時 30 分鐘經代碼調整後可以在 20 分鐘内完成。

4.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，變成兩個對，<(b,數字 id),value>，<(b,字元串 id),value>。

是以商品表的 HDFS 讀取隻會是一次。

4.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分鐘。

4.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，這樣在資料量小的時候，資料不會存在傾斜問題。

5.優化總結

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

• 長期觀察hadoop處理資料的過程，有幾個顯著的特征:

1. 不怕資料多，就怕資料傾斜。
2. 對jobs數比較多的作業運作效率相對比較低，比如即使有幾百行的表，如果多次關聯多次彙總，産生十幾個jobs，沒半小時是跑不完的。map reduce作業初始化的時間是比較長的。
3. 對sum，count來說，不存在資料傾斜問題。
4. 對count(distinct ),效率較低，資料量一多，準出問題，如果是多count(distinct )效率更低。

• 優化可以從幾個方面着手：

5. 自己動手寫sql解決資料傾斜問題是個不錯的選擇。set hive.groupby.skewindata=true;這是通用的算法優化，但算法優化總是漠視業務，習慣性提供通用的解決方法。Etl開發人員更了解業務，更了解資料，是以通過業務邏輯解決傾斜的方法往往更精确，更有效。
6. 對count(distinct)采取漠視的方法，尤其資料大的時候很容易産生傾斜問題，不抱僥幸心理。自己動手，豐衣足食。
7. 對小檔案進行合并，是行至有效的提高排程效率的方法，假如我們的作業設定合理的檔案數，對雲梯的整體排程效率也會産生積極的影響。

6.優化的常用手段

主要由三個屬性來決定：

• 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。

6.1參數設定的影響

如果reduce太少：如果資料量很大，會導緻這個reduce異常的慢，進而導緻這個任務不能結束，也有可能會OOM 2、如果reduce太多： 産生的小檔案太多，合并起來代價太高，namenode的記憶體占用也會增大。如果我們不指定mapred.reduce.tasks， hive會自動計算需要多少個reducer。

阿裡巴巴開源大資料技術團隊成立Apache Spark中國技術社群，定期推送精彩案例，技術專家直播，問答區近萬人Spark技術同學線上提問答疑，隻為營造純粹的Spark氛圍，歡迎釘釘掃碼加入！

對開源大資料和感興趣的同學可以加小編微信（下圖二維碼，備注“進群”）進入技術交流微信群。