天天看點

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資料傾斜也可以此為參照。

實際業務中有些大量的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階段

第二種:因為null值參與shuffle時的hash結果是一樣的,那麼我們可以給null值随機指派,這樣它們的hash結果就不一樣,就會進到不同的reduce中:

對于兩個表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類型配置設定了:

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

這種資料傾斜問題沒有什麼好的解決方案,隻能将使用GZIP壓縮等不支援檔案分割的檔案轉為bzip和zip等支援檔案分割的壓縮方式。

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

在多元聚合計算時,如果進行分組聚合的字段過多,如下:

<code>select a,b,c,count(1)from log group by a,b,c with rollup;</code>

注:對于最後的<code>with rollup</code>關鍵字不知道大家用過沒,with rollup是用來在分組統計資料的基礎上再進行統計彙總,即用來得到group by的彙總資訊。

如果上面的log表的資料量很大,并且Map端的聚合不能很好地起到資料壓縮的情況下,會導緻Map端産出的資料急速膨脹,這種情況容易導緻作業記憶體溢出的異常。如果log表含有資料傾斜key,會加劇Shuffle過程的資料傾斜。

可以拆分上面的sql,将<code>with rollup</code>拆分成如下幾個sql:

但是,上面這種方式不太好,因為現在是對3個字段進行分組聚合,那如果是5個或者10個字段呢,那麼需要拆解的SQL語句會更多。

在Hive中可以通過參數 <code>hive.new.job.grouping.set.cardinality</code>

配置的方式自動控制作業的拆解,該參數預設值是30。表示針對grouping

sets/rollups/cubes這類多元聚合的操作,如果最後拆解的鍵組合大于該值,會啟用新的任務去處理大于該值之外的組合。如果在處理資料時,某個分組聚合的列有較大的傾斜,可以适當調小該值。

兩表進行普通的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版本之前需要這樣寫:

如果想将多個表放到Map端記憶體中,隻需在mapjoin()中寫多個表名稱即可,用逗号分隔,如将a表和c表放到Map端記憶體中,則 <code>/* +mapjoin(a,c) */</code> 。

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

<code>hive.auto.convert.join=true</code> 預設值為true,自動開啟MAPJOIN優化。

<code>hive.mapjoin.smalltable.filesize=2500000</code> 預設值為2500000(25M),通過配置該屬性來确定使用該優化的表的大小,如果表的大小小于此值就會被加載進記憶體中。

注意:使用預設啟動該優化的方式如果出現莫名其妙的BUG(比如MAPJOIN并不起作用),就将以下兩個屬性置為fase手動使用MAPJOIN标記來啟動該優化:

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

<code>hive.ignore.mapjoin.hint=false</code> (不忽略MAPJOIN标記)

再提一句:将表放到Map端記憶體時,如果節點的記憶體很大,但還是出現記憶體溢出的情況,我們可以通過這個參數 <code>mapreduce.map.memory.mb</code> 調節Map端記憶體的大小。

在一些操作中,我們沒有辦法減少資料量,如在使用 collect_list 函數時:

collect_list:将分組中的某列轉為一個數組傳回。

在上述sql中,s_age如果存在資料傾斜,當資料量大到一定的數量,會導緻處理傾斜的reduce任務産生記憶體溢出的異常。

注:collect_list輸出一個數組,中間結果會放到記憶體中,是以如果collect_list聚合太多資料,會導緻記憶體溢出。

有小夥伴說這是 group by 分組引起的資料傾斜,可以開啟<code>hive.groupby.skewindata</code>參數來優化。我們接下來分析下:

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

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

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

調整reduce的記憶體大小使用<code>mapreduce.reduce.memory.mb</code>這個配置。

--END--