Hive千億級資料傾斜解決方案

資料傾斜問題剖析

本文首發于公眾號【五分鐘學大資料】

資料傾斜是分布式系統不可避免的問題，任何分布式系統都有幾率發生資料傾斜，但有些小伙伴在平時作業中感知不是很明顯，這里要注意本篇文章的標題—“千億級資料”，為什么說千億級，因為如果一個任務的資料量只有幾百萬，它即使發生了資料傾斜，所有資料都跑到一臺機器去執行，對于幾百萬的資料量，一臺機器執行起來還是毫無壓力的，這時資料傾斜對我們感知不大，只有資料達到一個量級時，一臺機器應付不了這么多的資料，這時如果發生資料傾斜，那么最后就很難算出結果，

所以就需要我們對資料傾斜的問題進行優化，盡量避免或減輕資料傾斜帶來的影響，

在解決資料傾斜問題之前，還要再提一句：沒有瓶頸時談論優化，都是自尋煩惱，

大家想想，在map和reduce兩個階段中，最容易出現資料傾斜的就是reduce階段，因為map到reduce會經過shuffle階段，在shuffle中默認會按照key進行hash，如果相同的key過多，那么hash的結果就是大量相同的key進入到同一個reduce中，導致資料傾斜，

那么有沒有可能在map階段就發生資料傾斜呢，是有這種可能的，

一個任務中，資料檔案在進入map階段之前會進行切分，默認是128M一個資料塊，但是如果當對檔案使用GZIP壓縮等不支持檔案分割操作的壓縮方式時，MR任務讀取壓縮后的檔案時，是對它切分不了的，該壓縮檔案只會被一個任務所讀取，如果有一個超大的不可切分的壓縮檔案被一個map讀取時，就會發生map階段的資料傾斜，

所以，從本質上來說，發生資料傾斜的原因有兩種：一是任務中需要處理大量相同的key的資料，二是任務讀取不可分割的大檔案，

資料傾斜解決方案

MapReduce和Spark中的資料傾斜解決方案原理都是類似的，以下討論Hive使用MapReduce引擎引發的資料傾斜，Spark資料傾斜也可以此為參照，

1. 空值引發的資料傾斜

實際業務中有些大量的null值或者一些無意義的資料參與到計算作業中，表中有大量的null值，如果表之間進行join操作，就會有shuffle產生，這樣所有的null值都會被分配到一個reduce中，必然產生資料傾斜，

之前有小伙伴問，如果A、B兩表join操作，假如A表中需要join的欄位為null，但是B表中需要join的欄位不為null，這兩個欄位根本就join不上啊，為什么還會放到一個reduce中呢？

這里我們需要明確一個概念，資料放到同一個reduce中的原因不是因為欄位能不能join上，而是因為shuffle階段的hash操作，只要key的hash結果是一樣的，它們就會被拉到同一個reduce中，

解決方案：

第一種：可以直接不讓null值參與join操作，即不讓null值有shuffle階段

SELECT *
FROM log a
 JOIN 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;

第二種：因為null值參與shuffle時的hash結果是一樣的，那么我們可以給null值隨機賦值，這樣它們的hash結果就不一樣，就會進到不同的reduce中：

SELECT *
FROM log a
 LEFT JOIN users b ON CASE 
   WHEN a.user_id IS NULL THEN concat('hive_', rand())
   ELSE a.user_id
  END = b.user_id;

2. 不同資料型別引發的資料傾斜

對于兩個表join，表a中需要join的欄位key為int，表b中key欄位既有string型別也有int型別，當按照key進行兩個表的join操作時，默認的Hash操作會按int型的id來進行分配，這樣所有的string型別都被分配成同一個id，結果就是所有的string型別的欄位進入到一個reduce中，引發資料傾斜，

解決方案：

如果key欄位既有string型別也有int型別，默認的hash就都會按int型別來分配，那我們直接把int型別都轉為string就好了，這樣key欄位都為string，hash時就按照string型別分配了：

SELECT *
FROM users a
 LEFT JOIN logs b ON a.usr_id = CAST(b.user_id AS string);

3. 不可拆分大檔案引發的資料傾斜

當集群的資料量增長到一定規模，有些資料需要歸檔或者轉儲，這時候往往會對資料進行壓縮；當對檔案使用GZIP壓縮等不支持檔案分割操作的壓縮方式，在日后有作業涉及讀取壓縮后的檔案時，該壓縮檔案只會被一個任務所讀取，如果該壓縮檔案很大，則處理該檔案的Map需要花費的時間會遠多于讀取普通檔案的Map時間，該Map任務會成為作業運行的瓶頸，這種情況也就是Map讀取檔案的資料傾斜，

解決方案：

這種資料傾斜問題沒有什么好的解決方案，只能將使用GZIP壓縮等不支持檔案分割的檔案轉為bzip和zip等支持檔案分割的壓縮方式，

所以，我們在對檔案進行壓縮時，為避免因不可拆分大檔案而引發資料讀取的傾斜，在資料壓縮的時候可以采用bzip2和Zip等支持檔案分割的壓縮演算法，

4. 資料膨脹引發的資料傾斜

在多維聚合計算時，如果進行分組聚合的欄位過多，如下：

select a，b，c，count（1）from log group by a，b，c with rollup;

注：對于最后的with rollup關鍵字不知道大家用過沒，with rollup是用來在分組統計資料的基礎上再進行統計匯總，即用來得到group by的匯總資訊，

如果上面的log表的資料量很大，并且Map端的聚合不能很好地起到資料壓縮的情況下，會導致Map端產出的資料急速膨脹，這種情況容易導致作業記憶體溢位的例外，如果log表含有資料傾斜key，會加劇Shuffle程序的資料傾斜，

解決方案：

可以拆分上面的sql，將with rollup拆分成如下幾個sql：

SELECT a, b, c, COUNT(1)
FROM log
GROUP BY a, b, c;

SELECT a, b, NULL, COUNT(1)
FROM log
GROUP BY a, b;

SELECT a, NULL, NULL, COUNT(1)
FROM log
GROUP BY a;

SELECT NULL, NULL, NULL, COUNT(1)
FROM log;

但是，上面這種方式不太好，因為現在是對3個欄位進行分組聚合，那如果是5個或者10個欄位呢，那么需要拆解的SQL陳述句會更多，

在Hive中可以通過引數 hive.new.job.grouping.set.cardinality 配置的方式自動控制作業的拆解，該引數默認值是30，表示針對grouping sets/rollups/cubes這類多維聚合的操作，如果最后拆解的鍵組合大于該值，會啟用新的任務去處理大于該值之外的組合，如果在處理資料時，某個分組聚合的列有較大的傾斜，可以適當調小該值，

5. 表連接時引發的資料傾斜

兩表進行普通的repartition join時，如果表連接的鍵存在傾斜，那么在 Shuffle 階段必然會引起資料傾斜，

解決方案：

通常做法是將傾斜的資料存到分布式快取中，分發到各個 Map任務所在節點，在Map階段完成join操作，即MapJoin，這避免了 Shuffle，從而避免了資料傾斜，

MapJoin是Hive的一種優化操作，其適用于小表JOIN大表的場景，由于表的JOIN操作是在Map端且在記憶體進行的，所以其并不需要啟動Reduce任務也就不需要經過shuffle階段，從而能在一定程度上節省資源提高JOIN效率，

在Hive 0.11版本之前，如果想在Map階段完成join操作，必須使用MAPJOIN來標記顯示地啟動該優化操作，由于其需要將小表加載進記憶體所以要注意小表的大小，

如將a表放到Map端記憶體中執行，在Hive 0.11版本之前需要這樣寫：

select /* +mapjoin(a) */ a.id , a.name, b.age 
from a join b 
on a.id = b.id;

如果想將多個表放到Map端記憶體中，只需在mapjoin()中寫多個表名稱即可，用逗號分隔，如將a表和c表放到Map端記憶體中，則 /* +mapjoin(a,c) */ ，

在Hive 0.11版本及之后，Hive默認啟動該優化，也就是不在需要顯示的使用MAPJOIN標記，其會在必要的時候觸發該優化操作將普通JOIN轉換成MapJoin，可以通過以下兩個屬性來設定該優化的觸發時機：

hive.auto.convert.join=true 默認值為true，自動開啟MAPJOIN優化，

hive.mapjoin.smalltable.filesize=2500000 默認值為2500000(25M)，通過配置該屬性來確定使用該優化的表的大小，如果表的大小小于此值就會被加載進記憶體中，

注意：使用默認啟動該優化的方式如果出現莫名其妙的BUG(比如MAPJOIN并不起作用)，就將以下兩個屬性置為fase手動使用MAPJOIN標記來啟動該優化:

hive.auto.convert.join=false (關閉自動MAPJOIN轉換操作)

hive.ignore.mapjoin.hint=false (不忽略MAPJOIN標記)

再提一句：將表放到Map端記憶體時，如果節點的記憶體很大，但還是出現記憶體溢位的情況，我們可以通過這個引數 mapreduce.map.memory.mb 調節Map端記憶體的大小，

6. 確實無法減少資料量引發的資料傾斜

在一些操作中，我們沒有辦法減少資料量，如在使用 collect_list 函式時：

select s_age,collect_list(s_score) list_score
from student
group by s_age

collect_list：將分組中的某列轉為一個陣列回傳，

在上述sql中，s_age有資料傾斜，但如果資料量大到一定的數量，會導致處理傾斜的Reduce任務產生記憶體溢位的例外，

collect_list輸出一個陣列，中間結果會放到記憶體中，所以如果collect_list聚合太多資料，會導致記憶體溢位，

有小伙伴說這是 group by 分組引起的資料傾斜，可以開啟hive.groupby.skewindata引數來優化，我們接下來分析下：

開啟該配置會將作業拆解成兩個作業，第一個作業會盡可能將Map的資料平均分配到Reduce階段，并在這個階段實作資料的預聚合，以減少第二個作業處理的資料量；第二個作業在第一個作業處理的資料基礎上進行結果的聚合，

hive.groupby.skewindata的核心作用在于生成的第一個作業能夠有效減少數量，但是對于collect_list這類要求全量操作所有資料的中間結果的函式來說，明顯起不到作用，反而因為引入新的作業增加了磁盤和網路I/O的負擔，而導致性能變得更為低下，

解決方案：

這類問題最直接的方式就是調整reduce所執行的記憶體大小，

調整reduce的記憶體大小使用mapreduce.reduce.memory.mb這個配置，