云計(jì)算大數(shù)據(jù)教程中Spark數(shù)據(jù)傾斜解決方案

?黑馬程序員傳智播客旗下高端IT教育品牌改變中國IT教育，我們正在行動(dòng)【黑馬程序員】云計(jì)算大數(shù)據(jù)教程中Spark數(shù)

據(jù)傾斜解決方案大數(shù)據(jù)hadoop教程入門視頻網(wǎng)盤：/course/301.html?1912sxkqq資料網(wǎng)盤：/s/1Pyu8elrgkPrbPVDmdi-jBg提取碼：mc4g1、原理以及現(xiàn)象分析1.1、數(shù)據(jù)傾斜怎么出現(xiàn)的在執(zhí)行shuffle操作的時(shí)候，是按照key,來進(jìn)行values的數(shù)據(jù)的輸出、拉取和聚合的。同一個(gè)key的values，一定是分配到一個(gè)reducetask進(jìn)行處理的。多個(gè)key對應(yīng)的values，比如一共是90萬。可能某個(gè)key對應(yīng)了88萬數(shù)據(jù)，被分配到一個(gè)task上去面去執(zhí)行。另外兩個(gè)task，可能各分配到了1萬數(shù)據(jù)，可能是數(shù)百個(gè)key，對應(yīng)的1萬條數(shù)據(jù)。這樣就會出現(xiàn)數(shù)據(jù)傾斜問題。想象一下，出現(xiàn)數(shù)據(jù)傾斜以后的運(yùn)行的情況。很糟糕！其中兩個(gè)task，各分配到了1萬數(shù)據(jù)，可能同時(shí)在10分鐘內(nèi)都運(yùn)行完了。另外一個(gè)task有88萬條，88*10=880分鐘=14.5個(gè)小時(shí)。大家看，本來另外兩個(gè)task很快就運(yùn)行完畢了（10分鐘），但是由于一個(gè)拖后腿的家伙，第三個(gè)task，要14.5個(gè)小時(shí)才能運(yùn)行完，就導(dǎo)致整個(gè)spark作業(yè)，也得14.5個(gè)小時(shí)才能運(yùn)行完。數(shù)據(jù)傾斜，一旦出現(xiàn)，是不是性能殺手？！1.2、發(fā)生數(shù)據(jù)傾斜以后的現(xiàn)象資料領(lǐng)取：+Q1679806262資料領(lǐng)?。?Q1679806262傳智播客旗下高端IT教育品牌改變中國IT教育，我們正在行動(dòng)Spark數(shù)據(jù)傾斜，有兩種表現(xiàn)：1、 你的大部分的task,都執(zhí)行的特別特別快，（你要用client模式，standaloneclient，yarnclient，本地機(jī)器一執(zhí)行spark-submit腳本，就會開始打印log），task175finished，剩下幾個(gè)task，執(zhí)行的特別特別慢，前面的task，一般1s可以執(zhí)行完5個(gè)，最后發(fā)現(xiàn)1000個(gè)task，998,999task，要執(zhí)行1個(gè)小時(shí)，2個(gè)小時(shí)才能執(zhí)行完一個(gè)task。出現(xiàn)以上loginfo，就表明出現(xiàn)數(shù)據(jù)傾斜了。這樣還算好的，因?yàn)殡m然老牛拉破車一樣非常慢，但是至少還能跑。2、 另一種情況是，運(yùn)行的時(shí)候，其他task都執(zhí)行完了，也沒什么特別的問題，但是有的task，就是會突然間報(bào)了一個(gè)OOM，JVMOutOfMemory,內(nèi)存溢出了，taskfailed，tasklost，resubmittingtask。反復(fù)執(zhí)行幾次都到了某個(gè)task就是跑不通，最后就掛掉。某個(gè)task就直接OOM,那么基本上也是因?yàn)閿?shù)據(jù)傾斜了，task分配的數(shù)量實(shí)在是太大了！所以內(nèi)存放不下，然后你的task每處理一條數(shù)據(jù)，還要?jiǎng)?chuàng)建大量的對象，內(nèi)存爆掉了。這樣也表明出現(xiàn)數(shù)據(jù)傾斜了。這種就不太好了，因?yàn)槟愕某绦蛉绻蝗ソ鉀Q數(shù)據(jù)傾斜的問題，壓根兒就跑不出來。作業(yè)都跑不完，還談什么性能調(diào)優(yōu)這些東西？！根據(jù)log去定位出現(xiàn)數(shù)據(jù)傾斜的原因，基本只可能是因?yàn)榘l(fā)生了shuffle操作，在shuffle的過程中，出現(xiàn)了數(shù)據(jù)傾斜的問題。因?yàn)槟硞€(gè)或者某些key對應(yīng)的數(shù)據(jù)，遠(yuǎn)遠(yuǎn)的高于其他的key。1、 你在自己的程序里面找找，哪些地方用了會產(chǎn)生shuffle的算子,groupByKey、countByKey、reduceByKey、join2、 看loglog一般會報(bào)是在你的哪一行代碼，導(dǎo)致了OOM異常?；蛘呖磍og,看看是執(zhí)行到了第幾個(gè)stageospark代碼，是怎么劃分成一個(gè)一個(gè)的stage的。哪一個(gè)stage生成的task特別慢，就能夠自己用肉眼去對你的spark代碼進(jìn)行stage的劃分，就能夠通過stage定位到你的代碼，到底哪里發(fā)生了數(shù)據(jù)傾斜。2、聚合源數(shù)據(jù)以及過濾導(dǎo)致傾斜的key視頻庫網(wǎng)址：視頻庫網(wǎng)址：資料領(lǐng)?。?Q1679806262傳智播客旗下高端IT教育品牌改變中國IT教育，我們正在行動(dòng)數(shù)據(jù)傾斜解決方案，第一個(gè)方案和第二個(gè)方案，一起來講。這兩個(gè)方案是最直接、最有效、最簡單的解決數(shù)據(jù)傾斜問題的方案。第一個(gè)方案：聚合源數(shù)據(jù)。第二個(gè)方案：過濾導(dǎo)致傾斜的key。后面的五個(gè)方案，尤其是最后4個(gè)方案，都是那種特別狂拽炫酷吊炸天的方案。但沒有第一二個(gè)方案簡單直接。如果碰到了數(shù)據(jù)傾斜的問題。上來就先考慮第一個(gè)和第二個(gè)方案看能不能做，如果能做的話，后面的5個(gè)方案，都不用去搞了。有效、簡單、直接才是最好的，徹底根除了數(shù)據(jù)傾斜的問題。2.1、方案一：聚合源數(shù)據(jù)一些聚合的操作，比如groupByKey、reduceByKey，groupByKey說白了就是拿到每個(gè)key對應(yīng)的values。reduceByKey說白了就是對每個(gè)key對應(yīng)的values執(zhí)行一定的計(jì)算。這些操作，比如groupByKey和reduceByKey，包括之前說的join。都是在spark作業(yè)中執(zhí)行的。spark作業(yè)的數(shù)據(jù)來源，通常是哪里呢？90%的情況下，數(shù)據(jù)來源都是hive表（hdfs，大數(shù)據(jù)分布式存儲系統(tǒng)）°hdfs上存儲的大數(shù)據(jù)。hive表中的數(shù)據(jù)通常是怎么出來的呢？有了spark以后,hive比較適合做什么事情？hive就是適合做離線的，晚上凌晨跑的，ETL（extracttransformload，數(shù)據(jù)的采集、清洗、導(dǎo)入），hivesql，去做這些事情，從而去形成一個(gè)完整的hive中的數(shù)據(jù)倉庫。說白了，數(shù)據(jù)倉庫，就是一堆表。spark作業(yè)的源表，hive表，通常情況下來說，也是通過某些hiveetl生成的。hiveetl可能是晚上凌晨在那兒跑。今天跑昨天的數(shù)據(jù)。數(shù)據(jù)傾斜，某個(gè)key對應(yīng)的80萬數(shù)據(jù)，某些key對應(yīng)幾百條，某些key對應(yīng)幾十條?，F(xiàn)在咱們直接在生成hive表的hiveetl中對數(shù)據(jù)進(jìn)行聚合。比如按key來分組，將key對應(yīng)的所有的values全部用一種特殊的格式拼接到一個(gè)字符串里面去，比如"key=sessionid,value:action_seq=1|user_id=1|search_keyword=火鍋|category_id=001;action_seq=2|user_id=1|search_keyword=涮肉|category_id=001”。對key進(jìn)行g(shù)roup，在spark中，拿至Ukey=sessionid，values。hiveetl中，直接對key進(jìn)行了聚合。那么也就意味著，每個(gè)key就只對應(yīng)一條數(shù)據(jù)。在spark中，就不需要再去執(zhí)行g(shù)roupByKey+map這種操作了。直接對每個(gè)key對應(yīng)的values字符串進(jìn)行map操作，進(jìn)行你需要的操作即可。spark中，可能對這個(gè)操作，就不需要執(zhí)行shffule操作了，也就根本不可能導(dǎo)致視頻庫網(wǎng)址：視頻庫網(wǎng)址：資料領(lǐng)?。?Q1679806262傳智播客旗下高端IT教育品牌改變中國IT教育，我們正在行動(dòng)數(shù)據(jù)傾斜?；蛘呤菍γ總€(gè)key在hiveetl中進(jìn)行聚合，對所有values聚合一下，不一定是拼接起來，可能是直接進(jìn)行計(jì)算。reduceByKey計(jì)算函數(shù)應(yīng)用在hiveetl中，從而得到每個(gè)key的values。聚合源數(shù)據(jù)方案第二種做法是，你可能沒有辦法對每個(gè)key聚合出來一條數(shù)據(jù)。那么也可以做一個(gè)妥協(xié)，對每個(gè)key對應(yīng)的數(shù)據(jù)，10萬條。有好幾個(gè)粒度，比如10萬條里面包含了幾個(gè)城市、幾天、幾個(gè)地區(qū)的數(shù)據(jù)，現(xiàn)在放粗粒度。直接就按照城市粒度，做一下聚合，幾個(gè)城市，幾天、幾個(gè)地區(qū)粒度的數(shù)據(jù)，都給聚合起來。比如說city_iddatearea_idselect…from…groupbycity_id盡量去聚合，減少每個(gè)key對應(yīng)的數(shù)量，也許聚合到比較粗的粒度之后，原先有10萬數(shù)據(jù)量的key，現(xiàn)在只有1萬數(shù)據(jù)量。減輕數(shù)據(jù)傾斜的現(xiàn)象和問題。2?2、方案二：過濾導(dǎo)致傾斜的key如果你能夠接受某些數(shù)據(jù)在spark作業(yè)中直接就摒棄掉不使用。比如說，總共有100萬個(gè)key。只有2個(gè)key是數(shù)據(jù)量達(dá)到10萬的。其他所有的key，對應(yīng)的數(shù)量都是幾十萬。這個(gè)時(shí)候，你自己可以去取舍，如果業(yè)務(wù)和需求可以理解和接受的話，在你從hive表查詢源數(shù)據(jù)的時(shí)候，直接在sql中用where條件，過濾掉某幾個(gè)key。那么這幾個(gè)原先有大量數(shù)據(jù)，會導(dǎo)致數(shù)據(jù)傾斜的key，被過濾掉之后，那么在你的spark作業(yè)中，自然就不會發(fā)生數(shù)據(jù)傾斜了。3、提高shuffle操作reduce并行度3.1、問題描述第一個(gè)和第二個(gè)方案，都不適合做，然后再考慮這個(gè)方案。將reducetask的數(shù)量變多，就可以讓每個(gè)reducetask分配到更少的數(shù)據(jù)量。這樣的話也許就可以緩解甚至是基本解決掉數(shù)據(jù)傾斜的問題。3.2、提升shufflereduce端并行度的操作方法很簡單，主要給我們所有的shuffle算子，比如groupByKey、countByKey、reduceByKey。在調(diào)用的時(shí)候，傳入進(jìn)去一個(gè)參數(shù)。那個(gè)數(shù)字，就代表了那個(gè)視頻庫網(wǎng)址：視頻庫網(wǎng)址：資料領(lǐng)?。?Q1679806262傳智播客旗下高端IT教育品牌改變中國IT教育，我們正在行動(dòng)shuffle操作的reduce端的并行度。那么在進(jìn)行shuffle操作的時(shí)候，就會對應(yīng)著創(chuàng)建指定數(shù)量的reducetask。這樣的話，就可以讓每個(gè)reducetask分配到更少的數(shù)據(jù)?；究梢跃徑鈹?shù)據(jù)傾斜的問題。比如說，原本某個(gè)task分配數(shù)據(jù)特別多，直接00M,內(nèi)存溢出了，程序沒法運(yùn)行，直接掛掉。按照log，找到發(fā)生數(shù)據(jù)傾斜的shuffle操作，給它傳入一個(gè)并行度數(shù)字，這樣的話，原先那個(gè)task分配到的數(shù)據(jù)，肯定會變少。就至少可以避免00M的情況，程序至少是可以跑的。3.3、提升shufflereduce并行度的缺陷治標(biāo)不治本的意思，因?yàn)樗鼪]有從根本上改變數(shù)據(jù)傾斜的本質(zhì)和問題。不像第一個(gè)和第二個(gè)方案（直接避免了數(shù)據(jù)傾斜的發(fā)生）。原理沒有改變，只是說，盡可能地去緩解和減輕shufflereducetask的數(shù)據(jù)壓力，以及數(shù)據(jù)傾斜的問題。實(shí)際生產(chǎn)環(huán)境中的經(jīng)驗(yàn)：1、 如果最理想的情況下，提升并行度以后，減輕了數(shù)據(jù)傾斜的問題，或者甚至可以讓數(shù)據(jù)傾斜的現(xiàn)象忽略不計(jì)，那么就最好。就不用做其他的數(shù)據(jù)傾斜解決方案了。2、 不太理想的情況下，比如之前某個(gè)task運(yùn)行特別慢，要5個(gè)小時(shí)，現(xiàn)在稍微快了一點(diǎn)，變成了4個(gè)小時(shí)?；蛘呤窃冗\(yùn)行到某個(gè)task，直接OOM,現(xiàn)在至少不會OOM了，但是那個(gè)task運(yùn)行特別慢，要5個(gè)小時(shí)才能跑完。那么，如果出現(xiàn)第二種情況的話，各位，就立即放棄第三種方案，開始去嘗試和選擇后面的四種方案。4、使用隨機(jī)key實(shí)現(xiàn)雙重聚合4.1、使用場景groupByKey、reduceByKey比較適合使用這種方式。join咱們通常不會這樣來做，后面會講三種針對不同的join造成的數(shù)據(jù)傾斜的問題的解決方案。4.2、解決方案第一輪聚合的時(shí)候，對key進(jìn)行打散，將原先一樣的key，變成不一樣的key，相當(dāng)于是將每個(gè)key分為多組。先針對多個(gè)組，進(jìn)行key的局部聚合。接著，再去除掉每個(gè)key的前綴，然后對視頻庫網(wǎng)址：視頻庫網(wǎng)址：資料領(lǐng)?。?Q1679806262傳智播客旗下高端IT教育品牌改變中國IT教育，我們正在行動(dòng)所有的key進(jìn)行全局的聚合。對groupByKey、reduceByKey造成的數(shù)據(jù)傾斜，有比較好的效果。如果說，之前的第一、第二、第三種方案，都沒法解決數(shù)據(jù)傾斜的問題，那么就只能依靠這一種方式了。5、將reducejoin轉(zhuǎn)換為mapjoin5.1、使用方式普通的join，那么肯定是要走shuffle。既然是走shuffle，那么普通的join就肯定是走的是reducejoin。那怎么將reducejoin轉(zhuǎn)換為mapjoin呢？先將所有相同的key，對應(yīng)的value匯聚到一個(gè)task中，然后再進(jìn)行join。5.2、使用場景這種方式適合在什么樣的情況下來使用？如果兩個(gè)RDD要進(jìn)行join,其中一個(gè)RDD是比較小的。比如一個(gè)RDD是100萬數(shù)據(jù)，一個(gè)RDD是1萬數(shù)據(jù)。（一個(gè)RDD是1億數(shù)據(jù)，一個(gè)RDD是100萬數(shù)據(jù)）。其中一個(gè)RDD必須是比較小的，broadcast出去那個(gè)小RDD的數(shù)據(jù)以后，就會在每個(gè)executor的blockmanager中都保存一份。要確保你的內(nèi)存足夠存放那個(gè)小RDD中的數(shù)據(jù)。這種方式下，根本不會發(fā)生shuffle操作，肯定也不會發(fā)生數(shù)據(jù)傾斜。從根本上杜絕了join操作可能導(dǎo)致的數(shù)據(jù)傾斜的問題。對于join中有數(shù)據(jù)傾斜的情況，大家盡量第一時(shí)間先考慮這種方式,效果非常好。不適合的情況兩個(gè)RDD都比較大，那么這個(gè)時(shí)候，你去將其中一個(gè)RDD做成broadcast，就很笨拙了。很可能導(dǎo)致內(nèi)存不足。最終導(dǎo)致內(nèi)存溢出，程序掛掉。而且其中某些key（或者是某個(gè)key），還發(fā)生了數(shù)據(jù)傾斜。此時(shí)可以采用最后兩種方式。對于join這種操作，不光是考慮數(shù)據(jù)傾斜的問題。即使是沒有數(shù)據(jù)傾斜問題，也完全可以優(yōu)先考慮，用我們講的這種高級的reducejoin轉(zhuǎn)mapjoin的技術(shù)，不要用普通的join，去通過shuffle，進(jìn)行數(shù)據(jù)的join。完全可以通過簡單的map，使用mapjoin的方式，犧牲一點(diǎn)內(nèi)存資源。在可行的情況下，優(yōu)先這么使用。不走shuffle，直接走map，是不是性能也會高很多？這是肯定的。視頻庫網(wǎng)址：資料領(lǐng)?。阂曨l庫網(wǎng)址：資料領(lǐng)?。?Q1679806262傳智播客旗下高端IT教育品牌改變中國傳智播客旗下高端IT教育品牌6、sample采樣傾斜key單獨(dú)進(jìn)行join6.1、方案實(shí)現(xiàn)思路將發(fā)生數(shù)據(jù)傾斜的key,單獨(dú)拉出來，放到一個(gè)RDD中去。就用這個(gè)原本會傾斜的keyRDD跟其他RDD單獨(dú)去join—下，這個(gè)時(shí)候key對應(yīng)的數(shù)據(jù)可能就會分散到多個(gè)task中去進(jìn)行join操作。就不至于說是，這個(gè)key跟之前其他的key混合在一個(gè)RDD中時(shí)，肯定是會導(dǎo)致一個(gè)key對應(yīng)的所有數(shù)據(jù)都到一個(gè)task中去，就會導(dǎo)致數(shù)據(jù)傾斜。6.2、使用場景這種方案什么時(shí)候適合使用？優(yōu)先對于join，肯定是希望能夠采用上一個(gè)方案，即reducejoin轉(zhuǎn)換mapjoin。兩個(gè)RDD數(shù)據(jù)都比較大，那么就不要那么搞了。針對你的RDD的數(shù)據(jù)，你可以自己把它轉(zhuǎn)換成一個(gè)中間表，或者是直接用countByKey()的方式，你可以看一下這個(gè)RDD各個(gè)key對應(yīng)的數(shù)據(jù)量。此時(shí)如果你發(fā)現(xiàn)整個(gè)RDD就一個(gè)，或者少數(shù)幾個(gè)key對應(yīng)的數(shù)據(jù)量特別多。盡量建議，比如就是一個(gè)key對應(yīng)的數(shù)據(jù)量特別多。此時(shí)可以采用這種方案，單拉出來那個(gè)最多的key，單獨(dú)進(jìn)行join,盡可能地將key分散到各個(gè)task上去進(jìn)行join操作。什么時(shí)候不適用呢？如果一個(gè)RDD中，導(dǎo)致數(shù)據(jù)傾斜的key特別多。那么此時(shí)，最好還是不要這樣了。還是使用我們最后一個(gè)方案，終極的join數(shù)據(jù)傾斜的解決方案。就是說，咱們單拉出來了一個(gè)或者少數(shù)幾個(gè)可能會產(chǎn)生數(shù)據(jù)傾斜的key，然后還可以進(jìn)行更加優(yōu)化的一個(gè)操作。對于那個(gè)key，從另外一個(gè)要join的表中，也過濾出來一份數(shù)據(jù)，比如可能就只有一條數(shù)據(jù)。userid2infoRDD，—個(gè)useridkey，就對應(yīng)一條數(shù)據(jù)。然后呢，采取對那個(gè)只有一條數(shù)據(jù)的RDD，進(jìn)行flatMap操作，打上100個(gè)隨機(jī)數(shù)，作為前綴，返回100條數(shù)據(jù)。單獨(dú)拉出來的可能產(chǎn)生數(shù)據(jù)傾斜的RDD，給每一條數(shù)據(jù)，都打上一個(gè)100以內(nèi)的隨機(jī)數(shù)，作為前綴。視頻庫網(wǎng)址：資料領(lǐng)?。?Q1679806262再去進(jìn)行join，是不是性能就更好了?？隙梢詫?shù)據(jù)進(jìn)行打散，去進(jìn)行join。join完以后