數據傾斜指mapreduce計算架構或flink等流式計算平臺下，在進(jìn)行計算時(shí)，發(fā)生的數據操作卡在某一個(gè)子計算任務(wù)而導致整個(gè)任務(wù)被卡住的現象。例如寫(xiě)hive SQL計算指標時(shí)，發(fā)現數據從1%到99%很快，確一直卡在99%很長(cháng)一段時(shí)間而無(wú)法完成任務(wù)。

實(shí)際上，發(fā)生這樣的原因是，在計算一條SQL的時(shí)候，某個(gè)任務(wù)被分配的key太多了。從而導致，其他任務(wù)很快完成任務(wù)，但是某個(gè)任務(wù)確一直在計算，造成『一人累死，其他人閑死』的情況。

二、MapReduce原理

什么是map任務(wù)？什么是recude任務(wù)呢？

我們知道hive的底層是通過(guò)HDFS將文件數據存在磁盤(pán)上，數據以key、value的形式存儲，而map操作相當于將把key、value鍵值對讀取出來(lái)，重新組合。然后將整理好的數據，交給recude，reduce進(jìn)行聚合計算。

舉個(gè)例子，如下數據：（HDFS文件存儲數據時(shí)，存在行存儲和列存儲兩種，下面以列存儲為例）

HDFS中存儲文件的格式如下：

（0,"hive spark hive hbase" ) （21,"hadoop hive spark hive" ) （39,"sqoop flume scala scala" )

map操作將對HDFS文件進(jìn)行分隔，并將每一行分成一個(gè)的<key,value>值（初始的k值是根據偏移量來(lái)的），然后在將<key,value>值，轉成如下的格式，并存入緩存中：

(hive,1),(spark,1),(hive,1),(hbase,1);
(hadoop,1),(hive,1),(spark,1), (hive,1);
(sqoop,1),(flume,1),(scala,1), (scala,1);

map操作將不同分區中的key值進(jìn)行整合排序，存放到一個(gè)集合當中，格式如下：

(hive,[1,1,1,1]),(spark,[1,1]),(hbase,[1]),(hadoop,[1]),(sqoop,[1]),(flume,[1]),(scala,[1,1])

隨后，Reduce任務(wù)先處理多個(gè)Map任務(wù)的輸出結果，再根據分區將其分配到不同的Reduce節點(diǎn)上（這個(gè)過(guò)程就是shuffle）；Reduce任務(wù)對多個(gè)Map的輸出結果進(jìn)行合并、排序、計算，生成新的 (k,v)  值，具體如下：

(hive,4),(spark,2), (scala,2) ,(hbase,1),(hadoop,1),(sqoop,1),(flume,1)

Reduce任務(wù)會(huì )將上一步輸出的<k,v>寫(xiě)到HDFS中，生成文件。

三、數據傾斜解決方案

數據傾斜一般都發(fā)生在reduce階段。

reduce階段最容易發(fā)生傾斜的操作是join和count distinct

下面列舉幾個(gè)常見(jiàn)的hive數據傾斜場(chǎng)景，和其對應解決方案：

3.1 空值數據傾斜優(yōu)化

在上報的日志信息中，通常會(huì )出現信息丟失的情況，如果用上報缺失的字段去關(guān)聯(lián)相關(guān)字段時(shí)就會(huì )出現數據傾斜的問(wèn)題。

案例：日志中的user_id上報缺失，如果取其中的user_id和用戶(hù)表的user_id進(jìn)行關(guān)聯(lián)的時(shí)候就會(huì )出現數據傾斜。

解決辦法： 數據傾斜主要原因是join的key值發(fā)生傾斜，key值包含很多空值或是異常值，通常的做法是，對空值或者異常值賦一個(gè)隨機的值來(lái)分散key。

3.2 大表和小表的優(yōu)化

在hive SQL的join操作中，mr過(guò)程是按照join的key進(jìn)行分發(fā)，而在join左邊的表的數據會(huì )首先被讀進(jìn)內存，如果左邊表的key相對分散，讀入內存的數據會(huì )比較小，join任務(wù)就會(huì )執行的比較快；如果左邊的key比較集中并且數據量比較大時(shí)，數據傾斜就會(huì )比較嚴重，執行時(shí)間將會(huì )增加。

經(jīng)驗總結：為了能夠避免數據傾斜現象，通常將數據量小的表放在join的左邊，此外，還需要使用map join讓小維度的表先進(jìn)內存，在map完成reduce操作。

SQL操作：/*+MAPJOIN(smallTable)*/

上面操作主要是將小表全部讀入內存中，在map階段大表的每個(gè)map和小表進(jìn)行匹配，節省了reduce階段的時(shí)間，提高了數據執行效率。

3.3 大表和大表的join優(yōu)化

• 情況一：Map階段輸出的key數量上，導致reduce階段的reduce數量為1

  案例：日志表中的user_id和用戶(hù)表關(guān)聯(lián)，user表上有500W+條記錄，把user分發(fā)到所有的map開(kāi)銷(xiāo)很大，并且map join不支持大表操作。用普通的join操作，數據傾斜會(huì )產(chǎn)生。

  解決辦法：這類(lèi)問(wèn)題產(chǎn)生的根本原因就是數據業(yè)務(wù)特性強，兩個(gè)表都是大表。因此我們可以針對性的削減業(yè)務(wù)過(guò)程，比如log表中user_id有上百萬(wàn)個(gè)，但是每天會(huì )員的uv不會(huì )太多，有交易的會(huì )員不會(huì )很多，有點(diǎn)擊的會(huì )員數不會(huì )很多等，在業(yè)務(wù)過(guò)程削減冗余的數據量，避免數據傾斜的發(fā)生。

  -- 思路：先過(guò)濾出需要被使用的唯一的user_id，再計算相關(guān)指標 select /*MAPJOIN(b)*/ from log a left join ( select /*MAPJOIN(b)*/ b1.* from (select distinct user_id from log) b join user b1 on b.user_id = b1.user_id ) a1 on a.user_id = a1.user_id;

• 情況二：Map輸出key分布不均勻，商品信息表的key對應大量的value，導致數據傾斜。

  案例：商品信息表a中的信息填充到商品瀏覽日志表b中，使用商品id進(jìn)行關(guān)聯(lián)。但是某些人買(mǎi)商品瀏覽量較大，造成數據偏移。

  解決辦法：熱點(diǎn)數據和非熱點(diǎn)數據拆分處理

  select * from ( select /*MAPJOIN(i)*/ a.id, a.time, a.amount, b.name, b.loc, b.cat from a left join (select * from a where uid in ('1001','1002')) as b -- 熱點(diǎn)子表 on a.uid = b.uid where a.uid in('1001','1002') -- 熱點(diǎn)數據 ) union all select a.id, a.time, a.amount, b.name, b.loc, b.cat from a left join b on a.uid = b.uid where a.uid is not in('1001','1002') -- 非熱點(diǎn)數據

3.4 count(distinct)數據傾斜優(yōu)化

場(chǎng)景：在多個(gè)維度的同一個(gè)度量的count distinct

案例：根據月份和性別，統計買(mǎi)家的1月份男顧客數，女顧客數

原始方案：

改造方案：

把DISTINCT用到的buyer，也加到group by統計上，然后再進(jìn)行業(yè)務(wù)計算

3.5 經(jīng)驗總結

我們平時(shí)編寫(xiě)時(shí)需要記住以下幾點(diǎn)：

• 使用分區剪裁、列剪裁，分區一定要加

• 少用 COUNT DISTINCT，group by 代替 distinct

• 是否存在多對多的關(guān)聯(lián)

• 連接表時(shí)使用相同的關(guān)鍵詞，這樣只會(huì )產(chǎn)生一個(gè) job

• 減少每個(gè)階段的數據量，只選出需要的，在 join 表前就進(jìn)行過(guò)濾

• 大表放后面

• 謂詞下推：where 謂詞邏輯都盡可能提前執行，減少下游處理的數據量

• 上一篇：【低代碼搭建】物資管理信息化解決方案
• 下一篇：【低代碼搭建】實(shí)驗室管理系統建設指南