05調(diào)度系統(tǒng)數(shù)據(jù)不動代碼到底意思

05|調(diào)度系統(tǒng)：“數(shù)據(jù)不動代碼動”2021-03-24 進(jìn)入課來提升CPU利用率，控制任務(wù)并行度的參數(shù)是Spark的配置項：spark.default.parallelism。增加并行度確實能夠充分利用閑置的CPU線程，但是，parallelism數(shù)值也不宜過大，過大反而會引入過多的調(diào)度開銷，得不償失。數(shù)值一個數(shù)值的嘗試，parallelismLabel在機(jī)器學(xué)習(xí)應(yīng)用中，特征工程幾乎占據(jù)了算法同學(xué)80%的時間和精力，畢竟，一份質(zhì)量優(yōu)良的訓(xùn)練樣本限定了模型效果的上限和天花板，我們要講的案例就來自特征工程中一個典型的處理場景：LaelEcoing（標(biāo)簽編碼）。什么是Laelencoing呢？模型特征按照是否連續(xù)可以分為兩類：連續(xù)性數(shù)值特征和離散型特征，離散型特征往往以字符串的形式存在，比如用戶興趣特征就包括體育、政治、軍事和娛樂等。對于很多機(jī)器學(xué)習(xí)算法來說，字符串類型的數(shù)據(jù)是不能直接消費的，需要轉(zhuǎn)換為數(shù)值才行，例如把體育、政治、軍事、娛樂映射為0、1、2、3，這個過程在機(jī)器學(xué)習(xí)領(lǐng)域有個術(shù)語就叫Laelencoing。我們這一講的案例，就是要對用戶興趣特征做Laelecoing，簡單來說就是以固定的模板把字符串轉(zhuǎn)換為數(shù)值，然后將千億條樣本中的用戶興趣轉(zhuǎn)換為對應(yīng)的索引值。固定模板是離線模型訓(xùn)練與線上模型服務(wù)之間的文件接口，內(nèi)容僅包含用戶興趣這一列，字符串已按事先約定好的規(guī)則進(jìn)行排序。我們需要注意的是，用戶興趣包含4個層級，因此這個模板文件較大，記錄數(shù)達(dá)到萬級別。那具體怎么轉(zhuǎn)換呢？例如，我們可以將用戶興趣“體育-籃球-NBA-湖人”映射為0，將興趣“軍事-武器-步槍-A47”映射為1，以此類推。應(yīng)該說，需求還是相當(dāng)明確的，我身邊的同學(xué)們拿到需求之后，奔兒都沒打，以迅雷不及掩耳之勢就實現(xiàn)了如下的處理函數(shù)。11實現(xiàn)方式deffindIndex(templatePath:String,interest:String):Int{valsource=Source.fromFile(filePath,"UTF-8")vallines=source.getLines().toArrayvalsearchMap=lines.zip(0untilsearchMap.getOrElse(interest,-findIndex(filePath,"體育-籃球-NBA-湖人我們可以看到這個函數(shù)有兩個形參，一個是模板文件路徑，另一個是訓(xùn)練樣本中的用戶興趣。處理函數(shù)首先讀取模板文件，然后根據(jù)文件中排序的字符串構(gòu)建一個從興趣到索引的Map映射，最后在這個Map中查找第二個形參傳入的用戶興趣，如果能找到則返回對應(yīng)的索引，找不到的話則返回-1。這段代碼看上去似乎沒什么問題，同學(xué)們基于上面的函數(shù)對千億樣本做Labelencoding，在20臺機(jī)型為C5.4xlargeAWSEC2的分布式集群中花費了5個小時。坦白說，這樣的實現(xiàn)方式valfindIndex:(String)=>(String)=>Int{(filePath)valsource=Source.fromFile(filePath,"UTF-8")vallines=source.getLines().toArrayvalsearchMap=lines.zip(0until(interest)=>searchMap.getOrElse(interest,-valpartFunc=同學(xué)們基于第二種方式對相同的數(shù)據(jù)集做Labelencoding之后，在10臺同樣機(jī)型的分布式集群中花了不到20分鐘就把任務(wù)跑完了。可以說，執(zhí)行性能的提升是顯而易見的。那我們可以看到，相比于第一份代碼，第二份代碼的函數(shù)體內(nèi)沒有任何變化，還是先讀取模板文件、構(gòu)建Map映射、查找用戶興趣，最后返回索引。最大的區(qū)別就是第二份代碼對高階函數(shù)的使用，具體來說有2點：處理函數(shù)定義為高階函數(shù)，形參是模板文件路徑，返回結(jié)果是從用戶興趣到索引的函數(shù)；封裝千億樣本的Dataset所調(diào)用的函數(shù)，不是第一份代碼中的findIndex，而是用模板文件調(diào)用findIndex得到的partFunc，partFunc是形參為興趣、結(jié)果為索引的普通標(biāo)Spark調(diào)度系統(tǒng)的核心職責(zé)是，先將用戶構(gòu)建的DAG轉(zhuǎn)化為分布式任務(wù)，結(jié)合分布式集群資源的可用性，基于調(diào)度規(guī)則依序把分布式任務(wù)分發(fā)到執(zhí)行器。這個過程聽上去就夠復(fù)這里，我們先對內(nèi)存計算的第二層含義做個簡單地回顧，它指的是同一tage中的所有操作會被捏合為一個函數(shù)，這個函數(shù)一次性會被地應(yīng)用到輸入數(shù)據(jù)上，并且一次性地產(chǎn)生計算結(jié)果。升級之前的土豆加工流程DAG被切分為3個執(zhí)行階段Stage，它們分別是Stage0、Stage1、Stage2。其中，Stage0產(chǎn)出即食薯片，Stage1分發(fā)調(diào)味品，Stage2則產(chǎn)Stage0，Stage03是清洗、切片和烘焙。這3個環(huán)節(jié)需要3種不同的設(shè)備，即清洗機(jī)、切片機(jī)和烤箱。工坊有3條流水線，每種設(shè)備都需要3套，在成本方面要花不少錢呢，因此工坊老板一直此時，工頭兒建議：“老板，我聽說市場上有一種可編程的土豆加工設(shè)備，它是個黑盒子并且只有輸入口和輸出口，從外面看不見里面的操作流程。不過黑盒子受程序控制，給定輸入口的食材，我們可以編寫程序控制黑盒子的輸出。有了這個可編程設(shè)備，咱們不但省了錢，將來還可以靈活地擴(kuò)充產(chǎn)品線。比方想生產(chǎn)各種風(fēng)味的薯條或是土豆泥，只需要更換一份程序加載到黑盒子里就行啦！”于是，工坊的加工流水線就變成了如下的樣子。工人們的工作也從按照DAG流程圖的關(guān)鍵步驟，在流水線上安裝相應(yīng)的設(shè)備，變成了把關(guān)鍵步驟編寫相應(yīng)的程序加載到黑盒內(nèi)。這樣一來，這家工坊的生產(chǎn)力也從作坊式的生產(chǎn)方式，升級到了現(xiàn)代化流水線的作業(yè)模式。那么，這個故事跟我們今天要講的調(diào)度系統(tǒng)有什么關(guān)系呢？事實上，Spark作流程包含如下5個步驟：將DAG創(chuàng)建分布式任務(wù)Tasks和任務(wù)組依序?qū)⒎植际饺蝿?wù)分發(fā)到執(zhí)行器Executor4現(xiàn)在，你可能會覺得用故事來記這幾個步驟好像多此一舉，但當(dāng)我們學(xué)完了所有的原理之后，再回過頭來把故事的主線串聯(lián)起來，你就會驚喜地發(fā)現(xiàn)，所有的原理你都能輕松地記住和理解，這可比死記硬背的效率要高得多。接下來，我們深入到流程中的每一步去探究Spark調(diào)度系統(tǒng)是如何工作的。不過在此之Sark調(diào)度系統(tǒng)包含3個核心組件，分別是DAGceder、TaskSeder和cedeBacend。這3個組件都運行在Dier進(jìn)程中，它們通力合作將用戶構(gòu)建的DAG轉(zhuǎn)化為分布式任務(wù)，再把這些任務(wù)分發(fā)給集群中的Eecutors去執(zhí)行。不過，它們的名字都包含ceder，光看名字還真是丈二和尚摸不著頭腦，所以我把它們和調(diào)度系統(tǒng)流程中5個步驟的對應(yīng)關(guān)系總結(jié)在了下表中，你可以看一看。DAGScheduler的主要職責(zé)有二：一是把用戶DAG拆分為Stages，如果你不記得這個過程可以回顧一下上一講的內(nèi)容；二是在Stage內(nèi)創(chuàng)建計算任務(wù)Tasks，戶通過組合不同算子實現(xiàn)的數(shù)據(jù)轉(zhuǎn)換邏輯。然后，執(zhí)行器Executors接收到Tasks，會將不過，如果我們給集群中處于繁忙或者是飽和狀態(tài)的Executors分發(fā)了任務(wù)，執(zhí)行效果會大打折扣。因此，在分發(fā)任務(wù)之前，調(diào)度系統(tǒng)得先判斷哪些節(jié)點的計算資源空閑，然后再SchedulerBackend就是用來干這個事的，它是對于資源調(diào)度器的封裝與抽象，為了支持多樣的資源調(diào)度模式如Standalone、YARN和Mesos，SchedulerBackend提供了對應(yīng)的實現(xiàn)類。在運行時，Spark根據(jù)用戶提供的MasterURL，來決定實例化哪種實現(xiàn)類的對象。MasterURL就是你通過各種方式指定的資源管理器，如--masterspark://ip:host（Standalone模式）、--masteryarn（YARN模式）對于集群中可用的計算資源，SchedulerBackend會用一個叫做ExecutorDataMap的數(shù)據(jù)結(jié)構(gòu)，來記錄每一個計算節(jié)點中Executors的資源狀態(tài)。ExecutorDataMap是一種HashMapKeyExecutor的字符串，ValueExecutorData據(jù)結(jié)構(gòu)，ExecutorData用于封裝Executor的資源狀態(tài)，如RPC地址、主機(jī)地址、可用CPU核數(shù)和滿配CPU核數(shù)等等，它相當(dāng)于是對Executor做的“資源畫像”?？偟膩碚f，對內(nèi)，SchedulerBackend用ExecutorData對Executor進(jìn)行資源畫像；對外，SchedulerBackend以WorkerOffer為粒度提供計算資源，WorkerOffer封裝了ExecutorID、主機(jī)地址和CPU核數(shù)，用來表示一份可用于調(diào)度任務(wù)的空閑資源。顯然，基于Executor資源畫像，SchedulerBackend可以同時提供多個WorkerOffer用于分布式任務(wù)調(diào)度。WorkerOfferOffer工作機(jī)會，結(jié)合Spark調(diào)度系統(tǒng)的上下文，就變成了使用硬件資源的機(jī)會。好了，到此為止，要調(diào)度的計算任務(wù)有了，就是DAGScheduler通過Stages創(chuàng)建的Tasks；SchedulerBackendWorkerOffer。如果從供需的角度看待任務(wù)調(diào)度，DAGScheduler就是需求端，SchedulerBackend就是供給端。左邊有需求，右邊有供給，如果把Spark要有個中介來幫它們對接意愿、撮合交易，從而最大限度地提升資源配置的效率。在Spark調(diào)度系統(tǒng)中，這個中介就是TaskScheduler。TaskScheduler的職責(zé)是，基于既定的規(guī)則顯然，TaskSchedulerTaskScheduler兩個層次，一個是不同Stages之間的調(diào)度優(yōu)先級，一個是Stages內(nèi)不同任務(wù)之間的調(diào)首先，對于兩個或多個Saes，如果它們彼此之間不存在依賴關(guān)系、互相獨立，在面對同一份可用計算資源的時候，它們之間就會存在競爭關(guān)系。這個時候，先調(diào)度誰、或者說誰優(yōu)先享受這份計算資源，大家就得基于既定的規(guī)則和協(xié)議照章辦事了。StagesTaskScheduler2。照被創(chuàng)建的時間順序來依次消費可用計算資源。這就好比在二手房交易市場中，兩個人同時看中一套房子，不管兩個人各自愿意出多少錢，誰最先交定金，中介就優(yōu)先給誰和賣家撮合交易。交?！睕]錯，考慮到開發(fā)者的意愿度，TaskScheduler提供了FAIR公平調(diào)度模式。在這種模式下，哪個Stages優(yōu)先被調(diào)度，取決于用戶在配置文件fairscheduler.xml中的定在配置文件中，Sak允許用戶定義不同的調(diào)度池，每個調(diào)度池可以指定不同的調(diào)度優(yōu)先級，用戶在開發(fā)過程中可以關(guān)聯(lián)不同作業(yè)與調(diào)度池的對應(yīng)關(guān)系，這樣不同Saes的調(diào)度就直接和開發(fā)者的意愿掛鉤，也就能享受不同的優(yōu)先級待遇。對應(yīng)到二手房交易的例子中，如果第二個人樂意付30%的高溢價，中介自然樂意優(yōu)先撮合他與賣家的交易。StagesStages調(diào)度優(yōu)先級，Stages內(nèi)部的任務(wù)調(diào)度相對來說簡單得多。當(dāng)TaskScheduler接收到來自SchedulerBackend的WorkerOffer后，TaskScheduler級別要求的任務(wù)進(jìn)行分發(fā)。眾所周知，本地性級別有4種：Processlocal<NodelocalRacklocalAny。從左到右分別是進(jìn)程本地性、節(jié)點本地性、機(jī)架本地性和跨機(jī)架本進(jìn)程本地性表示計算任務(wù)所需的輸入數(shù)據(jù)就在某一個Eeutor進(jìn)程內(nèi)，因此把這樣的計算任務(wù)調(diào)度到目標(biāo)進(jìn)程內(nèi)最劃算。同理，如果數(shù)據(jù)源還未加載到Eecutor進(jìn)程，而是存儲在某一計算節(jié)點的磁盤中，那么把任務(wù)調(diào)度到目標(biāo)節(jié)點上去，也是一個不錯的選擇。再次，如果我們無法確定輸入源在哪臺機(jī)器，但可以肯定它一定在某個機(jī)架上，本地性級別就會退化到Racklcal。DAGScheduler劃分Stages、創(chuàng)建分布式任務(wù)的過程中，會為每一個任務(wù)指定本地性級別，本地性級別中會記錄該任務(wù)有意向的計算節(jié)點地址，甚至是Executor進(jìn)程ID。換句話說，任務(wù)自帶調(diào)度意愿，它通過本地性級別告訴TaskScheduler自己更樂意被調(diào)度到既然計算任務(wù)的個人意愿這么強(qiáng)烈，TaskScheduler作為中間商，肯定要優(yōu)先滿足人家的意愿。這就像一名碼農(nóng)想要租西二旗的房子，但是房產(chǎn)中介App推送的結(jié)果都是東三環(huán)國由此可見，Spak調(diào)度系統(tǒng)的原則是盡可能地讓數(shù)據(jù)呆在原地、保持不動，同時盡可能地把承載計算任務(wù)的代碼分發(fā)到離數(shù)據(jù)最近的地方，從而最大限度地降低分布式系統(tǒng)中的網(wǎng)絡(luò)開銷。畢竟，分發(fā)代碼的開銷要比分發(fā)數(shù)據(jù)的代價低太多，這也正是“數(shù)據(jù)不動代碼動”這個說法的由來?？偟膩碚f，TaskScheduler根據(jù)本地性級別遴選出待計算任務(wù)之后，先對這些任務(wù)進(jìn)行序列化。然后，交給SchedulerBackend，SchedulerBackend根據(jù)ExecutorData中記錄的RPC地址和主機(jī)地址，再將序列化的任務(wù)通過網(wǎng)絡(luò)分發(fā)到目的主機(jī)的Executor中去。最后，Executor接收到任務(wù)之后，把任務(wù)交由內(nèi)置的線程池，線程池中的多線程則并發(fā)地在不同Labelencoding例中，2種實現(xiàn)方式的差別到底在哪兒了。我們先來回顧案例中處理函數(shù)的主要計算步2種實現(xiàn)方式的本質(zhì)區(qū)別在于，函數(shù)中2個計算步驟的分布式計算過程不同。在第1種實現(xiàn)方式中，函數(shù)是一個接收兩個形參的普通標(biāo)量函數(shù)，Dataset上做Labelencoding。在Spark任務(wù)調(diào)度流程中，該函數(shù)在Driver端交由DAGScheduler打包為Tasks，經(jīng)過TaskScheduler調(diào)度給SchedulerBackend，最后由SchedulerBackend分發(fā)到集群中的ExecutorsExecutors2，2Driver參的標(biāo)量函數(shù)，這個標(biāo)量函數(shù)內(nèi)攜帶了剛剛建好的映射字典。最后，Dataset函數(shù)作用于千億樣本之上做Labelencoding。發(fā)現(xiàn)區(qū)別了嗎？在第2種實現(xiàn)中，函數(shù)的第一步計算只在Dier端計算一次，分發(fā)給集群中所有Eeutors的任務(wù)中封裝的是攜帶了字典的標(biāo)量函數(shù)。然后在Execuors端，Eecutors在各自的數(shù)據(jù)分片上調(diào)用該函數(shù)，省去了掃描模板文件、建立字典的開銷。最后，我們只需要把樣本中的用戶興趣傳遞進(jìn)去，函數(shù)就能以O(shè)1)的查詢效率返回數(shù)值結(jié)果。對于一個有著成百上千Executors的分布式集群來說，這2種不同的實現(xiàn)方式帶來的性能差異還是相當(dāng)可觀的。因此，如果你能把Spark調(diào)度系統(tǒng)的工作原理牢記于心，我相信在代碼開發(fā)或是review的過程中，你都能夠意識到第一個計算步驟會帶來的性能問題。這種開今天這一講，我們先通過一個機(jī)器學(xué)的案例對比了2種實現(xiàn)方式的性能差異，知道了對于調(diào)度系統(tǒng)一知半解，很有可能在開發(fā)過程中引入潛在的性能隱患。為此，我梳理了調(diào)度系統(tǒng)工作流程的5個主要步驟：將DAG創(chuàng)建分布式任務(wù)Tasks和任務(wù)組依序?qū)⒎植际饺蝿?wù)分發(fā)到執(zhí)行器5SparkSparkDAG集群資源的可用性，基于調(diào)度規(guī)則依序把分布式任務(wù)分發(fā)到執(zhí)行器Executors；Spark承載計算任務(wù)的代碼分發(fā)到離數(shù)據(jù)最近的地方（Executors），從而最大限D(zhuǎn)AGScheduler在創(chuàng)建Tasks在計算與存儲分離的云計算環(huán)境中，Nodelocal?上一 04|DAG與流水線：到底啥叫“內(nèi)存計算下一 06|存儲系統(tǒng)：空間換時間，還是時間換空間2021-03-2 32021-03-老師正例這個，先建map，再broadcastmap22021-03-Source.fromFile用于讀本地文件，所以用spark讀文件不是應(yīng)該用sc.textFile來從hdfs文件，建字典；2在千億樣本上查找字典。3 22021-03-12021-03-Partialfunctions來舉例，目的是為了講調(diào)度系統(tǒng)。Partialfunctions不是關(guān)鍵哈，任務(wù)調(diào)度過程中，分發(fā)了哪些東西到executors，這個才重41Partialfunctions只是方便舉例，我順手就拿過來用了，它只是一種實現(xiàn)形式哈～412021-03-自己應(yīng)該調(diào)度到哪個節(jié)點，甚至是哪個executors。最后schedulerBackend會把task代碼，分發(fā)1作者回復(fù):本地性是2021-03-老師，請教一個問題，在GraphEmbedding中，基于用戶行為的構(gòu)造的物品概率轉(zhuǎn)移矩構(gòu)造by用戶的物品轉(zhuǎn)移序列樣本。有了樣本其實就可以拿來訓(xùn)練了。但是這里有dataskew的風(fēng)merge重新merge成一個序列（byuser）。當(dāng)然這里你可能需要定義udf來專門干merge這個2 2021-03-等。rdddataframe.apply的計算邏輯，才會分發(fā)到executors去執(zhí)行。2021-03-2021-03-driver讀進(jìn)內(nèi)存、構(gòu)建字典，最后從driver22021-03-第一題：因為是為每個parttonta，所以在建立ta之前，都會獲取每個tton的位置偏好信息。首先判斷rddrddId+tInx組合成oId判斷。如果沒有，判斷rLocatons，看起來是判斷是否chpont