福哥答案2020-08-17：

數(shù)據(jù)傾斜是大數(shù)據(jù)領(lǐng)域繞不開的攔路虎，當(dāng)你所需處理的數(shù)據(jù)量到達(dá)了上億甚至是千億條的時(shí)候，數(shù)據(jù)傾斜將是橫在你面前一道巨大的坎。很可能有幾周甚至幾月都要頭疼于數(shù)據(jù)傾斜導(dǎo)致的各類詭異的問題。

數(shù)據(jù)傾斜是指：mapreduce程序執(zhí)行時(shí)，reduce節(jié)點(diǎn)大部分執(zhí)行完畢，但是有一個(gè)或者幾個(gè)reduce節(jié)點(diǎn)運(yùn)行很慢，導(dǎo)致整個(gè)程序的處理時(shí)間很長(zhǎng)，這是因?yàn)槟骋粋€(gè)key的條數(shù)比其他key多很多(有時(shí)是百倍或者千倍之多)，這條key所在的reduce節(jié)點(diǎn)所處理的數(shù)據(jù)量比其他節(jié)點(diǎn)就大很多，從而導(dǎo)致某幾個(gè)節(jié)點(diǎn)遲遲運(yùn)行不完。Hive的執(zhí)行是分階段的，map處理數(shù)據(jù)量的差異取決于上一個(gè)stage的reduce輸出，所以如何將數(shù)據(jù)均勻的分配到各個(gè)reduce中，就是解決數(shù)據(jù)傾斜的根本所在。

以下是一些常見的數(shù)據(jù)傾斜情形：

一、Group by 傾斜

group by造成的傾斜相對(duì)來說比較容易解決。hive提供兩個(gè)參數(shù)可以解決：

1.1 hive.map.aggr

一個(gè)是hive.map.aggr，默認(rèn)值已經(jīng)為true，他的意思是做map aggregation，也就是在mapper里面做聚合。這個(gè)方法不同于直接寫mapreduce的時(shí)候可以實(shí)現(xiàn)的combiner，但是卻實(shí)現(xiàn)了類似combiner的效果。事實(shí)上各種基于mr的框架如pig，cascading等等用的都是map aggregation(或者叫partial aggregation)而非combiner的策略，也就是在mapper里面直接做聚合操作而不是輸出到buffer給combiner做聚合。對(duì)于map aggregation，hive還會(huì)做檢查，如果aggregation的效果不好，那么hive會(huì)自動(dòng)放棄map aggregation。判斷效果的依據(jù)就是經(jīng)過一小批數(shù)據(jù)的處理之后，檢查聚合后的數(shù)據(jù)量是否減小到一定的比例，默認(rèn)是0.5，由hive.map.aggr.hash.min.reduction這個(gè)參數(shù)控制。所以如果確認(rèn)數(shù)據(jù)里面確實(shí)有個(gè)別取值傾斜，但是大部分值是比較稀疏的，這個(gè)時(shí)候可以把比例強(qiáng)制設(shè)為1，避免極端情況下map aggr失效。hive.map.aggr還有一些相關(guān)參數(shù)，比如map aggr的內(nèi)存占用等，具體可以參考這篇文章。

1.2 hive.groupby.skewindata

另一個(gè)參數(shù)是hive.groupby.skewindata。這個(gè)參數(shù)的意思是做reduce操作的時(shí)候，拿到的key并不是所有相同值給同一個(gè)reduce，而是隨機(jī)分發(fā)，然后reduce做聚合，做完之后再做一輪MR，拿前面聚合過的數(shù)據(jù)再算結(jié)果。所以這個(gè)參數(shù)其實(shí)跟hive.map.aggr做的是類似的事情，只是拿到reduce端來做，而且要額外啟動(dòng)一輪job，所以其實(shí)不怎么推薦用，效果不明顯。

1.3 count distinct 改寫

另外需要注意的是count distinct操作往往需要改寫SQL，可以按照下面這么做：

```bash

/*改寫前*/

select a, count(distinct b) as c from tbl group by a;

/*改寫后*/

select a, count(*) as c from (select a, b from tbl group by a, b) group by a;

```

二、Join傾斜

2.1 skew join

join造成的傾斜，常見情況是不能做map join的兩個(gè)表(能做map join的話基本上可以避免傾斜)，其中一個(gè)是行為表，另一個(gè)應(yīng)該是屬性表。比如我們有三個(gè)表，一個(gè)用戶屬性表users，一個(gè)商品屬性表items，還有一個(gè)用戶對(duì)商品的操作行為表日志表logs。假設(shè)現(xiàn)在需要將行為表關(guān)聯(lián)用戶表：

```bash

select * from logs a join users b on a.user_id = b.user_id;

```

其中l(wèi)ogs表里面會(huì)有一個(gè)特殊用戶user_id = 0，代表未登錄用戶，假如這種用戶占了相當(dāng)?shù)谋壤?#xff0c;那么個(gè)別reduce會(huì)收到比其他reduce多得多的數(shù)據(jù)，因?yàn)樗邮账衭ser_id = 0的記錄進(jìn)行處理，使得其處理效果會(huì)非常差，其他reduce都跑完很久了它還在運(yùn)行。

hive給出的解決方案叫skew join，其原理把這種user_id = 0的特殊值先不在reduce端計(jì)算掉，而是先寫入hdfs，然后啟動(dòng)一輪map join專門做這個(gè)特殊值的計(jì)算，期望能提高計(jì)算這部分值的處理速度。當(dāng)然你要告訴hive這個(gè)join是個(gè)skew join，即：set

> hive.optimize.skewjoin = true;

還有要告訴hive如何判斷特殊值，根據(jù)hive.skewjoin.key設(shè)置的數(shù)量hive可以知道，比如默認(rèn)值是100000，那么超過100000條記錄的值就是特殊值。總結(jié)起來，skew join的流程可以用下圖描述：

2.2 特殊值分開處理法

不過，上述方法還要去考慮閾值之類的情況，其實(shí)也不夠通用。所以針對(duì)join傾斜的問題，一般都是通過改寫sql解決。對(duì)于上面這個(gè)問題，我們已經(jīng)知道user_id = 0是一個(gè)特殊key，那么可以把特殊值隔離開來單獨(dú)做join，這樣特殊值肯定會(huì)轉(zhuǎn)化成map join，非特殊值就是沒有傾斜的普通join了：

```bash

select

*

from

(

select * from logs where user_id = 0

)

a

join

(

select * from users where user_id = 0

)

b

on

a.user_id = b.user_id

union all

select * from logs a join users b on a.user_id <> 0 and a.user_id = b.user_id;

```

2.3 隨機(jī)數(shù)分配法

上面這種個(gè)別key傾斜的情況只是一種傾斜情況。最常見的傾斜是因?yàn)閿?shù)據(jù)分布本身就具有長(zhǎng)尾性質(zhì)，比如我們將日志表和商品表關(guān)聯(lián)：

```bash

select * from logs a join items b on a.item_id = b.item_id;

```

這個(gè)時(shí)候，分配到熱門商品的reducer就會(huì)很慢，因?yàn)闊衢T商品的行為日志肯定是最多的，而且我們也很難像上面處理特殊user那樣去處理item。這個(gè)時(shí)候就會(huì)用到加隨機(jī)數(shù)的方法，也就是在join的時(shí)候增加一個(gè)隨機(jī)數(shù)，隨機(jī)數(shù)的取值范圍n相當(dāng)于將item給分散到n個(gè)reducer：

```bash

select

a.*,

b.*

from

(

select *, cast(rand() * 10 as int) as r_id from logs

)

a

join

(

select *, r_id from items lateral view explode(range_list(1, 10)) rl as r_id

)

b

on

a.item_id = b.item_id

and a.r_id = b.r_id

```

上面的寫法里，對(duì)行為表的每條記錄生成一個(gè)1-10的隨機(jī)整數(shù)，對(duì)于item屬性表，每個(gè)item生成10條記錄，隨機(jī)key分別也是1-10，這樣就能保證行為表關(guān)聯(lián)上屬性表。其中range_list(1,10)代表用udf實(shí)現(xiàn)的一個(gè)返回1-10整數(shù)序列的方法。這個(gè)做法是一個(gè)解決join傾斜比較根本性的通用思路，就是如何用隨機(jī)數(shù)將key進(jìn)行分散。當(dāng)然，可以根據(jù)具體的業(yè)務(wù)場(chǎng)景做實(shí)現(xiàn)上的簡(jiǎn)化或變化。

2.4 業(yè)務(wù)設(shè)計(jì)

除了上面兩類情況，還有一類情況是因?yàn)闃I(yè)務(wù)設(shè)計(jì)導(dǎo)致的問題，也就是說即使行為日志里面join key的數(shù)據(jù)分布本身并不明顯傾斜，但是業(yè)務(wù)設(shè)計(jì)導(dǎo)致其傾斜。比如對(duì)于商品item_id的編碼，除了本身的id序列，還人為的把item的類型也作為編碼放在最后兩位，這樣如果類型1(電子產(chǎn)品)的編碼是00，類型2(家居產(chǎn)品)的編碼是01，并且類型1是主要商品類，將會(huì)造成以00為結(jié)尾的商品整體傾斜。這時(shí)，如果reduce的數(shù)量恰好是100的整數(shù)倍，會(huì)造成partitioner把00結(jié)尾的item_id都hash到同一個(gè)reducer，引爆問題。這種特殊情況可以簡(jiǎn)單的設(shè)置合適的reduce值來解決，但是這種坑對(duì)于不了解業(yè)務(wù)的情況下就會(huì)比較隱蔽。

三、典型的業(yè)務(wù)場(chǎng)景

3.1 空值產(chǎn)生的數(shù)據(jù)傾斜

場(chǎng)景：如日志中，常會(huì)有信息丟失的問題，比如日志中的 user_id，如果取其中的 user_id 和 用戶表中的user_id 關(guān)聯(lián)，會(huì)碰到數(shù)據(jù)傾斜的問題。

解決方法1：user_id為空的不參與關(guān)聯(lián)

```bash

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;

```

解決方法2 ：賦與空值分新的key值

```bash

select

*

from

log a

left outer join users b

on

case

when a.user_id is null

then concat(‘hive’, rand())

else a.user_id

end = b.user_id;

```

結(jié)論：方法2比方法1效率更好，不但io少了，而且作業(yè)數(shù)也少了。解決方法1中 log讀取兩次，jobs是2。解決方法2 job數(shù)是1 。這個(gè)優(yōu)化適合無效 id (比如 -99 , ’’, null 等) 產(chǎn)生的傾斜問題。把空值的 key 變成一個(gè)字符串加上隨機(jī)數(shù)，就能把傾斜的數(shù)據(jù)分到不同的reduce上 ,解決數(shù)據(jù)傾斜問題。

3.2 不同數(shù)據(jù)類型關(guān)聯(lián)產(chǎn)生數(shù)據(jù)傾斜

場(chǎng)景：用戶表中user_id字段為int，log表中user_id字段既有string類型也有int類型。當(dāng)按照user_id進(jìn)行兩個(gè)表的Join操作時(shí)，默認(rèn)的Hash操作會(huì)按int型的id來進(jìn)行分配，這樣會(huì)導(dǎo)致所有string類型id的記錄都分配到一個(gè)Reducer中。

解決方法：把數(shù)字類型轉(zhuǎn)換成字符串類型

```bash

select

*

from

users a

left outer join logs b

on

a.usr_id = cast(b.user_id as string)

```

3.3 小表不小不大，怎么用 map join 解決傾斜問題

使用 map join 解決小表(記錄數(shù)少)關(guān)聯(lián)大表的數(shù)據(jù)傾斜問題，這個(gè)方法使用的頻率非常高，但如果小表很大，大到map join會(huì)出現(xiàn)bug或異常，這時(shí)就需要特別的處理。以下例子:

```bash

select * from log a left outer join users b on a.user_id = b.user_id;

```

users 表有 600w+ 的記錄，把 users 分發(fā)到所有的 map 上也是個(gè)不小的開銷，而且 map join 不支持這么大的小表。如果用普通的 join，又會(huì)碰到數(shù)據(jù)傾斜的問題。

```bash

select

/*+mapjoin(x)*/

*

from

log a

left outer join

(

select

/*+mapjoin(c)*/

d.*

from

(

select distinct user_id from log

)

c

join users d

on

c.user_id = d.user_id

)

x on a.user_id = b.user_id;

```

假如，log里user_id有上百萬個(gè)，這就又回到原來map join問題。所幸，每日的會(huì)員uv不會(huì)太多，有交易的會(huì)員不會(huì)太多，有點(diǎn)擊的會(huì)員不會(huì)太多，有傭金的會(huì)員不會(huì)太多等等。所以這個(gè)方法能解決很多場(chǎng)景下的數(shù)據(jù)傾斜問題。

四、總結(jié)

使map的輸出數(shù)據(jù)更均勻的分布到reduce中去，是我們的最終目標(biāo)。由于Hash算法的局限性，按key Hash會(huì)或多或少的造成數(shù)據(jù)傾斜。大量經(jīng)驗(yàn)表明數(shù)據(jù)傾斜的原因是人為的建表疏忽或業(yè)務(wù)邏輯可以規(guī)避的。在此給出較為通用的步驟：

1)采樣log表，哪些user_id比較傾斜，得到一個(gè)結(jié)果表tmp1。由于對(duì)計(jì)算框架來說，所有的數(shù)據(jù)過來，他都是不知道數(shù)據(jù)分布情況的，所以采樣是并不可少的。

2)數(shù)據(jù)的分布符合社會(huì)學(xué)統(tǒng)計(jì)規(guī)則，貧富不均。傾斜的key不會(huì)太多，就像一個(gè)社會(huì)的富人不多，奇特的人不多一樣。所以tmp1記錄數(shù)會(huì)很少。把tmp1和users做map join生成tmp2,把tmp2讀到distribute file cache。這是一個(gè)map過程。

3)map讀入users和log，假如記錄來自log,則檢查user_id是否在tmp2里，如果是，輸出到本地文件a,否則生成的key,value對(duì)，假如記錄來自member,生成的key,value對(duì)，進(jìn)入reduce階段。

4)最終把a(bǔ)文件，把Stage3 reduce階段輸出的文件合并起寫到hdfs。

如果確認(rèn)業(yè)務(wù)需要這樣傾斜的邏輯，考慮以下的優(yōu)化方案：

1)對(duì)于join，在判斷小表不大于1G的情況下，使用map join

2)對(duì)于group by或distinct，設(shè)定 hive.groupby.skewindata=true

3)盡量使用上述的SQL語句調(diào)節(jié)進(jìn)行優(yōu)化

五、參考文獻(xiàn)

[數(shù)據(jù)分析系列(3)：數(shù)據(jù)傾斜](https://blog.csdn.net/anshuai_aw1/article/details/84033160)

***

[評(píng)論](https://user.qzone.qq.com/3182319461/blog/1597618770)

總結(jié)

以上是生活随笔為你收集整理的select怎么设置默认值_20200817：详细说下数据倾斜怎么解决？的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。