Hive|如何避免數(shù)據(jù)傾斜
1. hive中桶的概述
對于每一個表（table）或者分區(qū)， Hive可以進(jìn)一步組織成桶，也就是說桶是更為細(xì)粒度的數(shù)據(jù)范圍劃分。Hive也是 針對某一列進(jìn)行桶的組織。Hive采用對列值哈希，然后除以桶的個數(shù)求余的方式?jīng)Q定該條記錄存放在哪個桶當(dāng)中。
把表（或者分區(qū)）組織成桶（Bucket）有兩個理由：
（1）獲得更高的查詢處理效率。
桶為表加上了額外的結(jié)構(gòu)，Hive 在處理有些查詢時(shí)能利用這個結(jié)構(gòu)。具體而言，連接兩個在（包含連接列的）相同列上劃分了桶的表，可以使用 Map 端連接 （Map-side join）高效的實(shí)現(xiàn)。比如JOIN操作。對于JOIN操作兩個表有一個相同的列，如果對這兩個表都進(jìn)行了桶操作。那么將保存相同列值的桶進(jìn)行JOIN操作就可以，可以大大較少JOIN的數(shù)據(jù)量。
（2）使取樣（sampling）更高效。
在處理大規(guī)模數(shù)據(jù)集時(shí)，在開發(fā)和修改查詢的階段，如果能在數(shù)據(jù)集的一小部分?jǐn)?shù)據(jù)上試運(yùn)行查詢，會帶來很多方便。
創(chuàng)建帶桶的 table
create?table?bucketed_user(id?int,name?string)?clustered?by?(id)?sorted?by(name)?into?4?buckets?row?format?delimited?fields?terminated?by?'\t'?stored?as?textfile;?
首先，我們來看如何告訴Hive—個表應(yīng)該被劃分成桶。我們使用CLUSTERED BY 子句來指定劃分桶所用的列和要劃分的桶的個數(shù)：
CREATE?TABLE?bucketed_user?(id?INT)?name?STRING)?
CLUSTERED?BY?(id)?INTO?4?BUCKETS;?
在這里，我們使用用戶ID來確定如何劃分桶(Hive使用對值進(jìn)行哈希并將結(jié)果除 以桶的個數(shù)取余數(shù)。這樣，任何一桶里都會有一個隨機(jī)的用戶集合（PS：其實(shí)也能說是隨機(jī)，不是嗎？）。
對于map端連接的情況，兩個表以相同方式劃分桶。處理左邊表內(nèi)某個桶的 mapper知道右邊表內(nèi)相匹配的行在對應(yīng)的桶內(nèi)。因此，mapper只需要獲取那個桶 (這只是右邊表內(nèi)存儲數(shù)據(jù)的一小部分)即可進(jìn)行連接。這一優(yōu)化方法并不一定要求 兩個表必須桶的個數(shù)相同，兩個表的桶個數(shù)是倍數(shù)關(guān)系也可以。用HiveQL對兩個劃分了桶的表進(jìn)行連接，可參見“map連接”部分（P400）。
桶中的數(shù)據(jù)可以根據(jù)一個或多個列另外進(jìn)行排序。由于這樣對每個桶的連接變成了高效的歸并排序(merge-sort), 因此可以進(jìn)一步提升map端連接的效率。以下語法聲明一個表使其使用排序桶：
CREATE?TABLE?bucketed_users?(id?INT,?name?STRING)?
CLUSTERED?BY?(id)?SORTED?BY?(id?ASC)?INTO?4?BUCKETS;?
我們?nèi)绾伪ＷC表中的數(shù)據(jù)都劃分成桶了呢？把在Hive外生成的數(shù)據(jù)加載到劃分成 桶的表中，當(dāng)然是可以的。其實(shí)讓Hive來劃分桶更容易。這一操作通常針對已有的表。
Hive并不檢查數(shù)據(jù)文件中的桶是否和表定義中的桶一致(無論是對于桶 的數(shù)量或用于劃分桶的列）。如果兩者不匹配，在査詢時(shí)可能會碰到錯 誤或未定義的結(jié)果。因此，建議讓Hive來進(jìn)行劃分桶的操作。
2.hive中join優(yōu)化
mapside join
方法一：
select?/*+?MAPJOIN(time_dim)?*/?count(*)?from
store_sales?join?time_dim?on?(ss_sold_time_sk?=?t_time_sk)
方法二：這個可以由hive自動進(jìn)行map端join
set?hive.auto.convert.join=true;
select?count(*)?from
store_sales?join?time_dim?on?(ss_sold_time_sk?=?t_time_sk)
執(zhí)行下面這段代碼將會產(chǎn)生兩個map-only方法：
select?/*+?MAPJOIN(time_dim,?date_dim)?*/?count(*)?from
store_sales
join?time_dim?on?(ss_sold_time_sk?=?t_time_sk)?
join?date_dim?on?(ss_sold_date_sk?=?d_date_sk)
where?t_hour?=?8?and?d_year?=?2002
設(shè)置下面兩個屬性hive將會進(jìn)行自動執(zhí)行上述過程，第一個屬性默認(rèn)為true，第二個屬性是設(shè)置map端join適合讀取內(nèi)存文件的大小。
set?hive.auto.convert.join.noconditionaltask?=?true;
set?hive.auto.convert.join.noconditionaltask.size?=?10000000;
Sort-Merge-Bucket (SMB) joins可以被轉(zhuǎn)化為SMB map joins。
我們只需要設(shè)置一下幾個參數(shù)即可：
set?hive.auto.convert.sortmerge.join=true;
set?hive.optimize.bucketmapjoin?=?true;
set?hive.optimize.bucketmapjoin.sortedmerge?=?true;
已設(shè)置大表選擇的策略
使用下面屬性：
set?hive.auto.convert.sortmerge.join.bigtable.selection.policy=?org.apache.hadoop.hive.ql.optimizer.TableSizeBasedBigTableSelectorForAutoSMJ;
幾種策略設(shè)置
org.apache.hadoop.hive.ql.optimizer.AvgPartitionSizeBasedBigTableSelectorForAutoSMJ?(default)
org.apache.hadoop.hive.ql.optimizer.LeftmostBigTableSelectorForAutoSMJ
org.apache.hadoop.hive.ql.optimizer.TableSizeBasedBigTableSelectorForAutoSMJ
詳細(xì)請參考一下連接：hive中進(jìn)行連接方案詳解(https://cwiki.apache.org/confluence/display/Hive/LanguageManual+Joins)
3，數(shù)據(jù)傾斜
Hive的執(zhí)行是分階段的，map處理數(shù)據(jù)量的差異取決于上一個stage的reduce輸出，所以如何將數(shù)據(jù)均勻的分配到各個reduce中，就是解決數(shù)據(jù)傾斜的根本所在。
操作
原因
1，數(shù)據(jù)在節(jié)點(diǎn)上分布不均
2，key分布不均(key中存在個別值數(shù)據(jù)量比較大，比如NULL,那么join時(shí)就會容易發(fā)生數(shù)據(jù)傾斜)
3，count(disctinct key)，在數(shù)據(jù)兩比較大的時(shí)候容易發(fā)生數(shù)據(jù)傾斜，因?yàn)閏ount（distinct）是按照group by字段進(jìn)行分組的
4，group by的使用容易造成數(shù)據(jù)傾斜
5，業(yè)務(wù)數(shù)據(jù)本身的特性
6，建表時(shí)考慮不周
7，某些SQL語句本身就有數(shù)據(jù)傾斜
表現(xiàn)
任務(wù)進(jìn)度長時(shí)間維持在99%左右，查看任務(wù)監(jiān)控頁面發(fā)現(xiàn)只有少量reduce任務(wù)未完成。因?yàn)槠涮幚淼臄?shù)據(jù)量和其他reduce差異過大。單一reduce的記錄數(shù)與平均記錄數(shù)差異過大，通?？赡苓_(dá)到3倍甚至更多。最長時(shí)長遠(yuǎn)大于平均時(shí)長。
4，數(shù)據(jù)傾斜的解決方案
參數(shù)調(diào)節(jié)
set?hive.map.aggr=true;
Map?端部分聚合，相當(dāng)于Combiner
set?hive.groupby.skewindata=true;
有數(shù)據(jù)傾斜的時(shí)候進(jìn)行負(fù)載均衡，當(dāng)選項(xiàng)設(shè)定為 true，生成的查詢計(jì)劃會有兩個 MR Job。第一個 MR Job 中，Map 的輸出結(jié)果集合會隨機(jī)分布到 Reduce 中，每個 Reduce 做部分聚合操作，并輸出結(jié)果，這樣處理的結(jié)果是相同的 Group By Key 有可能被分發(fā)到不同的 Reduce 中，從而達(dá)到負(fù)載均衡的目的；第二個 MR Job 再根據(jù)預(yù)處理的數(shù)據(jù)結(jié)果按照 Group By Key 分布到 Reduce 中（這個過程可以保證相同的 Group By Key 被分布到同一個 Reduce 中），最后完成最終的聚合操作。
SQL語句調(diào)節(jié)：
如何Join 關(guān)于驅(qū)動表的選取，選用join key分布最均勻的表作為驅(qū)動表 做好列裁剪和filter操作，以達(dá)到兩表做join的時(shí)候，數(shù)據(jù)量相對變小的效果。 大小表Join 使用map join讓小的維度表（1000條以下的記錄條數(shù)） 先進(jìn)內(nèi)存。在map端完成reduce. 大表Join大表 把空值的key變成一個字符串加上隨機(jī)數(shù)，把傾斜的數(shù)據(jù)分到不同的reduce上，由于null值關(guān)聯(lián)不上，處理后并不影響最終結(jié)果。 count distinct大量相同特殊值 容易傾斜，當(dāng)xx字段存在大量的某個值時(shí)，NULL或者空的記錄 解決思路 將特定的值，進(jìn)行特定的處理。比如是null，過濾掉，where case 特定方式轉(zhuǎn)換特定的值，使得這些值不一樣，同時(shí)這些值不影響分析。 group by維度過小：Group by在Map端進(jìn)行部分?jǐn)?shù)據(jù)合并
set?hive.map.aggr?;?-->?是否在Map端進(jìn)行數(shù)據(jù)聚合，默認(rèn)設(shè)置為true；
set?hive.groupby.mapaggr.checkinterval ;?-->?在Map端進(jìn)行聚合操作的條目數(shù)。
進(jìn)行負(fù)載均衡負(fù)載均衡
set?hive.groupby.skewindata?;
默認(rèn)值是false，需要設(shè)置成true?;
當(dāng)設(shè)置為true時(shí)，會變成兩個MapReduce?;
第一個MR JOb中，map的輸出結(jié)果會隨機(jī)分布到Reduce中，每個Reduce做部分聚合操作，并輸出結(jié)果，這樣出來的結(jié)果相同的Group By Key有可能被分發(fā)到不同的Reduce中，從而達(dá)到輔助均衡目的。
第二個MR JOb，會根據(jù)預(yù)處理數(shù)據(jù)結(jié)果按照key分布到Reduce中，最終完成聚合操作。
5，典型的業(yè)務(wù)場景
空值產(chǎn)生的數(shù)據(jù)傾斜
場景：如日志中，常會有信息丟失的問題，比如日志中的 user_id，如果取其中的 user_id 和 用戶表中的user_id 關(guān)聯(lián)，會碰到數(shù)據(jù)傾斜的問題。
解決方法1：user_id為空的不參與關(guān)聯(lián)
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值
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 。這個優(yōu)化適合無效 id (比如 -99 , ’’, null 等) 產(chǎn)生的傾斜問題。把空值的 key 變成一個字符串加上隨機(jī)數(shù)，就能把傾斜的數(shù)據(jù)分到不同的reduce上 ,解決數(shù)據(jù)傾斜問題。
不同數(shù)據(jù)類型關(guān)聯(lián)產(chǎn)生數(shù)據(jù)傾斜
場景：用戶表中user_id字段為int，log表中user_id字段既有string類型也有int類型。當(dāng)按照user_id進(jìn)行兩個表的Join操作時(shí)，默認(rèn)的Hash操作會按int型的id來進(jìn)行分配，這樣會導(dǎo)致所有string類型id的記錄都分配到一個Reducer中。
解決方法：把數(shù)字類型轉(zhuǎn)換成字符串類型
select?*?from?users?a
??left?outer?join?logs?b
??on?a.usr_id?=?cast(b.user_id?as?string)
小表不小不大，怎么用 map join 解決傾斜問題
使用 map join解決小表(記錄數(shù)少)關(guān)聯(lián)大表的數(shù)據(jù)傾斜問題，這個方法使用的頻率非常高，但如果小表很大，大到map join會出現(xiàn)bug或異常，這時(shí)就需要特別的處理。以下例子:
select?*?from?log?a
??left?outer?join?users?b
??on?a.user_id?=?b.user_id;
users 表有 600w+ 的記錄，把 users 分發(fā)到所有的 map 上也是個不小的開銷，而且 map join 不支持這么大的小表。如果用普通的 join，又會碰到數(shù)據(jù)傾斜的問題。
解決方法：
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有上百萬個，這就又回到原來map join問題。所幸，每日的會員uv不會太多，有交易的會員不會太多，有點(diǎn)擊的會員不會太多，有傭金的會員不會太多等等。所以這個方法能解決很多場景下的數(shù)據(jù)傾斜問題。
6，數(shù)據(jù)傾斜總結(jié)
使map的輸出數(shù)據(jù)更均勻的分布到reduce中去，是我們的最終目標(biāo)。由于Hash算法的局限性，按key Hash會或多或少的造成數(shù)據(jù)傾斜。大量經(jīng)驗(yàn)表明數(shù)據(jù)傾斜的原因是人為的建表疏忽或業(yè)務(wù)邏輯可以規(guī)避的。在此給出較為通用的步驟：
1、采樣log表，哪些user_id比較傾斜，得到一個結(jié)果表tmp1。由于對計(jì)算框架來說，所有的數(shù)據(jù)過來，他都是不知道數(shù)據(jù)分布情況的，所以采樣是并不可少的。
2、數(shù)據(jù)的分布符合社會學(xué)統(tǒng)計(jì)規(guī)則，貧富不均。傾斜的key不會太多，就像一個社會的富人不多，奇特的人不多一樣。所以tmp1記錄數(shù)會很少。把tmp1和users做map join生成tmp2,把tmp2讀到distribute file cache。這是一個map過程。
3、map讀入users和log，假如記錄來自log,則檢查user_id是否在tmp2里，如果是，輸出到本地文件a,否則生成的key,value對，假如記錄來自member,生成的key,value對，進(jìn)入reduce階段。
4、最終把a(bǔ)文件，把Stage3 reduce階段輸出的文件合并起寫到hdfs。
如果確認(rèn)業(yè)務(wù)需要這樣傾斜的邏輯，考慮以下的優(yōu)化方案：
1、對于join，在判斷小表不大于1G的情況下，使用map join
2、對于group by或distinct，設(shè)定
hive.groupby.skewindata=true
4.中間件等
更多信息請關(guān)注公眾號：「軟件老王」，關(guān)注不迷路，軟件老王和他的IT朋友們，分享一些他們的技術(shù)見解和生活故事。