丨廣播變量克制shuffle如何一招制勝

接下來，咱們借助一個小例子，來講一講廣播變量的含義與作用。這個例子和WordCout有關(guān)，它可以說是分布式編程里的 oworld了，WordCout就是用來統(tǒng)計文件中全部單詞的，你肯定已經(jīng)非常熟悉了，所以，我們例子中的需求增加了一點難度，我們要對指定列表中給定的單詞計數(shù)。1234valdict=List(“spark”,valwords==words.filter(word=>.map((_,1)).reduceByKey(_+按照這個需求，同學(xué)小A實現(xiàn)了如上的代碼，一共有4行，我們逐一來看。第1行在Driver端給定待查單詞列表dict；第2行以textFileAPI分布式文件，內(nèi)容包含一列，的是常見的單詞；第3行用列表dict中的單詞過濾分布式文件內(nèi)容，只保留dict中給定的單詞；第4行調(diào)用reduceByKey對單詞進行累加計數(shù)。學(xué)習(xí)過調(diào)度系統(tǒng)之后，我們知道，第一行代碼定義的dict列表連帶后面的3行代碼會一同打包到Task里面去。這個時候，Task就像是一架架小飛機，攜帶著這些“行李”，飛往集群中不同的Executors。對于這些“行李”來說，代碼的“負重”較輕，可以忽略不你可能會說：“也還好吧，dict個例子中的并行度是10000，那么，Driver端需要通過網(wǎng)絡(luò)分發(fā)總共10000份dict拷貝。這個時候，集群內(nèi)所有的Eecutors需要消耗大量內(nèi)存來這10000份的拷貝，對寶貴的網(wǎng)絡(luò)和內(nèi)存資源來說，這已經(jīng)是一筆不小的浪費了。更何況，如果換做一個更大的數(shù)據(jù)結(jié)構(gòu)，Task分發(fā)所引入的網(wǎng)絡(luò)與內(nèi)存開銷會更可怕。換句話說，統(tǒng)計計數(shù)的業(yè)務(wù)邏輯還沒有開始執(zhí)行，Spark但是，在著手優(yōu)化之前，我們不妨先來想，現(xiàn)有的問題是什么，我們要達到的目的是什么。結(jié)合剛剛的分析，我們不難發(fā)現(xiàn)，WrdCunt的痛點在于，數(shù)據(jù)結(jié)構(gòu)的分發(fā)和受制于并行，并且是以Task為粒度的，因此往往頻次過高。痛點明確了，調(diào)優(yōu)的目的也就清晰了，我們需要降低數(shù)據(jù)結(jié)構(gòu)分發(fā)的頻次。要達到這個目的，我們首先想到的就是降低并行度。不過，牽一發(fā)而動全身，并行度一旦調(diào)整，其他與CPU、內(nèi)存有關(guān)的配置項都要跟著適配，這難免把調(diào)優(yōu)變復(fù)雜了。實際上，要降低數(shù)據(jù)結(jié)構(gòu)的分發(fā)頻次，我們還可以考慮廣播變量。廣播變量是一種分發(fā)機制，它封裝目標(biāo)數(shù)據(jù)結(jié)構(gòu)，以Executors為粒度去做數(shù)據(jù)分發(fā)。換句話說，在廣播變量的工作機制下，數(shù)據(jù)分發(fā)的頻次等同于集群中的Executors個數(shù)。通常來說，集群中的Executors數(shù)量都遠遠小于Task數(shù)量，相差兩到三個數(shù)量級是常有的事。那么，對于第一版的WordCount實現(xiàn)，如果我們使用廣播變量的話，會有哪些代碼的改動很簡單，主要有兩個改動：第一個改動是用broadcast封裝dict列表，第二個改動是在dict列表的地方改用broadcast.value替代。12345valdict=List(“spark”,valbc=valwords==words.filter(word=>.map((_,1)).reduceByKey(_+在廣播變量的運行機制下，封裝成廣播變量的數(shù)據(jù)，由Driver端以Executors為粒度分發(fā)，每一個Executors接收到廣播變量之后，將其交給BlockManager管理。由于廣播變量攜帶的數(shù)據(jù)已經(jīng)通過專門的途徑到BlockManager中，因此分發(fā)到Executors的Task不需要再攜帶同樣的數(shù)據(jù)。這個時候，你可以把廣播變量想象成一架架貨機，專門為Task這些小飛機運送“大件行李”。Driver與每一個Executors之間都開通一條這樣的貨機航線，統(tǒng)一運載負重較大的“數(shù)據(jù)行李”。有了貨機來幫忙，Task小飛機只需要攜帶那些負重較輕的代碼就好了。等這些Task小飛機在Executors著陸，它們就可以到Executors的公用倉庫BlockManager里去提取它們的“大件行李”?？傊?，在廣播變量的機制下，dit列表數(shù)據(jù)需要分發(fā)和的次數(shù)銳減。我們假設(shè)集群中有20個Eecuors，不過任務(wù)并行度還是10000，那么，Drier需要通過網(wǎng)絡(luò)分發(fā)的it列表拷貝就會由原來的10000份減少到20份。同理，集群范圍內(nèi)所有Eeutors需要的it拷貝，也由原來的10000份，減少至20份。這個時候，引入廣播變量后的開銷只是原來Task分發(fā)的1/500！Driver我們來看這樣一個例子。在電子商務(wù)領(lǐng)域中，開發(fā)者往往用事實表來類數(shù)據(jù)，用維度表來像物品、用戶這樣的描述性數(shù)據(jù)。事實表的特點是規(guī)模龐大，數(shù)據(jù)體量隨著業(yè)務(wù)的發(fā)展不斷地快速增長。維度表的規(guī)模要比事實表小很多，數(shù)據(jù)體量的變化也相對穩(wěn)定。假設(shè)用戶維度數(shù)據(jù)以Parquet文件格式在HDFS文件系統(tǒng)中，業(yè)務(wù)部門需要我們valuserFile:String=valdf:DataFrame=valbc_df:Broadcast[DataFrame]=首先，我們用ParquetAPIHDFS分布式數(shù)據(jù)文件生成DataFrame，然后用broadcastDataFrame。從代碼上來看，這種實現(xiàn)方式和封裝普通變量沒有太大差別，它們都調(diào)用了broadcastAPI，只是傳入的參數(shù)不同。DriverDriver到各個Executors，再讓Executors把數(shù)據(jù)緩存到BlockManager就好了。與普通變量相比，分布式數(shù)據(jù)集的數(shù)據(jù)源不在Driver端，而是來自所有的Executors。Executors中的每個分布式任務(wù)負責(zé)生產(chǎn)全量數(shù)據(jù)集的一部分，也就是圖中不同的數(shù)據(jù)分區(qū)。因此，步驟1就是Driver從所有的Executors拉取這些數(shù)據(jù)分區(qū)，然后在本地構(gòu)建全量數(shù)據(jù)。步驟2與從普通變量創(chuàng)建廣播變量的過程類似。Driver把匯總好的全量數(shù)據(jù)分發(fā)給各個Executors，Executors將接收到的全量數(shù)據(jù)緩存到系統(tǒng)的BlockManager中。不難發(fā)現(xiàn)，相比從普通變量創(chuàng)建廣播變量，從分布式數(shù)據(jù)集創(chuàng)建廣播變量的網(wǎng)絡(luò)開銷更大。原因主要有二：一是，前者比后者多了一步網(wǎng)絡(luò)通信；二是，前者的數(shù)據(jù)體量通常比后者大很多。你可能會問：“Driver從Executors拉取DataFrame的數(shù)據(jù)分片，揉成一份全量數(shù)據(jù)，然后再廣去，拋開網(wǎng)絡(luò)開銷不說，來來回回得費這么大勁，圖啥呢？”這是一個好問題，因為以廣播變量的形式緩存分布式數(shù)據(jù)集，正是克制Shuffle锏。Shuffle為什么這么說呢？我還是拿電子商務(wù)場景舉例。有了用戶的數(shù)據(jù)之后，為了分析不同用戶的購物習(xí)慣，業(yè)務(wù)部門要求我們對表和用戶表進行數(shù)據(jù)關(guān)聯(lián)。這樣的數(shù)據(jù)關(guān)聯(lián)需求在數(shù)據(jù)分析領(lǐng)域還是相當(dāng)普遍的。valtransactionsDF:DataFrame=valuserDF:DataFrame=transactionsDF.join(userDF,Seq(“userID”),因為需求非常明確，同學(xué)小A立即調(diào)用Parquet數(shù)據(jù)源API，分布式文件，創(chuàng)表和用戶表的DataFrame，然后調(diào)用DataFrame的Join方法，以userID作為Joinkeys，用內(nèi)關(guān)聯(lián)（InnerJoin）的方式完成了兩表的數(shù)據(jù)關(guān)聯(lián)。在分布式環(huán)境中，表和用戶表想要以userID為Joinkeys進行關(guān)聯(lián)，就必須要確保一個前提：記錄和與之對應(yīng)的用戶信息在同一個Executors內(nèi)。也就是說，如果用戶黃小乙的購物信息都在Executor0，而個人屬性信息緩存在Executor2，那么，在分布在不進行任何調(diào)優(yōu)的情況下，Spark默認采用ShuffleJoin的方式來做到這一點。ShuffleJoin的過程主要有兩步。第一步就是對參與關(guān)聯(lián)的左右表分別進行Shuffle，Shuffle的分區(qū)規(guī)則是先對Joinkeys計算哈希值，再把哈希值對分區(qū)數(shù)取模。由于左右表的分區(qū)數(shù)是一致的，因此Shuffle過后，一定能夠保證userID相同的記錄和用戶數(shù)據(jù)坐落在同一個Executors內(nèi)。ShuffleJoinhffle完成之后，第二步就是在同一個Excutrs內(nèi)，Rducetask就可以對userID一致的記錄進行關(guān)聯(lián)操作。但是，由于表是事實表，數(shù)據(jù)體量異常龐大，對TB級別的數(shù)據(jù)進行hffle，想想都覺得可怕！因此，上面對兩個DaaFrame直接關(guān)聯(lián)的代碼，還有很大的調(diào)優(yōu)空間。我們該怎么做呢？話句話說，對于分布式環(huán)境中的數(shù)據(jù)關(guān)聯(lián)來說，要想確保記錄和與之對應(yīng)的用戶信息在同一個Eeutors中，我們有沒有其他辦法呢？克制Shuffle代代123456importvaltransactionsDF:DataFrame=_valuserDF:DataFrame=_valbcUserDF=77transactionsDF.join(bcUserDF,Seq(“userID”),8Drier從所有Eecutors收集erDF所屬的所有數(shù)據(jù)分片，在本地匯總用戶數(shù)據(jù)，然后給每一個Eecutors都發(fā)送一份全量數(shù)據(jù)的拷貝。既然每個Eeutors都有userDF的全量數(shù)據(jù)，這個時候，表的數(shù)據(jù)分區(qū)待在原地、保持不動，就可以輕松地關(guān)聯(lián)到一致的用戶數(shù)據(jù)。如此一來，我們不需要對數(shù)據(jù)體量巨大的表進行Shuffle，同樣可以在分布式環(huán)境中，完成兩張表的數(shù)據(jù)關(guān)聯(lián)。BroadcastJoin將小表廣播，避免大表利用廣播變量，我們成功地避免了海量數(shù)據(jù)在集群內(nèi)的、分發(fā)，節(jié)省了原本由ufle引入的磁盤和網(wǎng)絡(luò)開銷，大幅提升運行時執(zhí)行性能。當(dāng)然，采用廣播變量優(yōu)化也是有成本的，畢竟廣播變量的創(chuàng)建和分發(fā)，也是會帶來網(wǎng)絡(luò)開銷的。但是，相比大表的全網(wǎng)分發(fā)，小表的網(wǎng)絡(luò)開銷幾乎可以忽略不計。這種小投入、大產(chǎn)出，用極小的成本去博取高額的性能收益，真可以說是“四兩撥千斤”！在數(shù)據(jù)關(guān)聯(lián)場景中，廣播變量是克制Shuffle的锏。掌握了它，我們就能以極小的成第一種，從普通變量創(chuàng)建廣播變量。在廣播變量的運行機制下，普通變量的數(shù)據(jù)封裝成廣播變量，由Driver端以Executors為粒度進行分發(fā)，每一個Executors接收到廣播變量之后，將其交由BlockManager管理。Driver需要從所有的ExecutorsDriverExecutors，Executors存到系統(tǒng)的BlockManager中。ShuffleJoinsBroadcastJoins，就可以用小表廣播來代替大表的全網(wǎng)分發(fā)，真正做到克制Shuffle。Spark廣播機制現(xiàn)有的實現(xiàn)方式是存在隱患的，在數(shù)據(jù)量較大的情況下，Driver可能會成為瓶頸，你能想到更好的方式來重新實現(xiàn)Spark的廣播機制嗎？（提示：在什么情況下，不適合把ShuffleJoins轉(zhuǎn)換為BroadcastJoins？ 不得售賣。頁面已增加防盜追蹤，將依法其上一 11|Shuffle的工作原理：為什么說Shuffle是一時無兩的性能下一 13|廣播變量（二）：有哪些途徑讓SparkSQL選擇Broadcast絡(luò)分發(fā)多次，已經(jīng)遠超出了shufflejoin需要傳輸?shù)臄?shù)據(jù)作者回復(fù):9??哈～但是，據(jù)我觀察，這部分代碼尚未生效。細節(jié)可以參考這個ticket：【Executorsidebroadcastforbroadcastjoins】，看上去還是進先是調(diào)用 其中 newOutOfMemoryError("Notenoughmemorytobuildandbroadcastthetabletoall"+"workernodes.Asaworkaround,youcaneitherdisablebroadcastbysetting"+s"${SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key}to-1orincreasethespark"+s"drivermemorybysetting${SparkLauncher.DRIVER_MEMORY}toahighervalue.").initCause(oe.ge2 3廣播join不是默認開啟的嗎，好像小表默認10M；還需像文中代碼valbcUserDFbroahlerD相的錄和用數(shù)據(jù)坐在同一個ors內(nèi)?！眔pD的分?jǐn)?shù)、或者說并行度，確實是由D文件系統(tǒng)決定的；但是，he過后，每個分布式數(shù)據(jù)集的并行度，就由數(shù)pr..hl.priions來決，這個咱在配置哪一講喲~果你沒有手工用riion或是olehle過后（Re階段）默認確實是開啟的，默認值確實也是10MB，但是，這個10MB太太太太太太（xN）小了！12021-05-orpr.o.rs給or指定的rororor理能力其實的地方于，不的者、不同上下文并發(fā)和行這兩詞，所以，回你的問，其實什么不康的~100并行度意味著000個區(qū)的分式數(shù)據(jù)集，這應(yīng)該不見到。外10個的集群其實也不小了~不過你的200有到，不知是200行度，是 集群發(fā)。如果是 集群發(fā)的話這個和100r不上。意味著的每個ore要20線程，哈，目前沒有這給力的P