丨性能調(diào)優(yōu)必要性spark本身就很快為啥還需要我

在數(shù)據(jù)應(yīng)用場(chǎng)景中，ETL（EtratTrasformLoad）往往是打頭陣的那個(gè)，畢竟源數(shù)據(jù)經(jīng)過(guò)抽取和轉(zhuǎn)換才能用于探索和分析，或者是供養(yǎng)給機(jī)器學(xué)習(xí)算法進(jìn)行模型訓(xùn)練，從而挖掘出數(shù)據(jù)次的價(jià)值。我們今天要舉的兩個(gè)例子，都取自典型ETL端到端作業(yè)中常見(jiàn)的操作和計(jì)算任務(wù)。1ETLextractFields：它的輸入?yún)?shù)是Seq[Row]類型，也即數(shù)據(jù)條目序列；輸出結(jié)果的返回類型是Seq[(String,Int)]，也就是（String,Int）對(duì)兒的序列；函數(shù)的計(jì)算邏輯是從數(shù)據(jù)條目中抽取索引為2的字符串和索引為4的整型。應(yīng)該說(shuō)這個(gè)業(yè)務(wù)需求相當(dāng)簡(jiǎn)單明了，實(shí)現(xiàn)起來(lái)簡(jiǎn)直是小菜一碟。在實(shí)際開(kāi)發(fā)中，我觀察到有不少同學(xué)一上來(lái)就迅速地用下面的方式去實(shí)現(xiàn)，干脆利落，代碼寫得挺快，功能也沒(méi)問(wèn)題，UT、功能測(cè)試都能過(guò)。4varfieldsSeq[(String,5=>6fieldsfields:+(row.getString(2),78910}在上面這個(gè)函數(shù)體中，是先定義一個(gè)類型是eq[(Srin,In)]的變量fiels，變量類型和函數(shù)返回類型完全一致。然后，函數(shù)逐個(gè)遍歷輸入?yún)?shù)中的數(shù)據(jù)條目，抽取數(shù)據(jù)條目中索引是2和4的字段并且構(gòu)建二元元組，緊接著把元組追加到最初定義的變量fields中。最后，函數(shù)返回類型是eq[(Srin,nt)]的變量fields。乍看上去，這個(gè)函數(shù)似乎沒(méi)什么問(wèn)題。特殊的地方在于，盡管這個(gè)數(shù)據(jù)抽取函數(shù)很小，在復(fù)雜的ETL應(yīng)用里是非常微小的一環(huán)，但在整個(gè)ETL作業(yè)中，它會(huì)在不同地方被頻繁地反ETLETL想要讓ETLextractFields這個(gè)小函數(shù)會(huì)被頻繁地調(diào)用，不如我們從它下手好了，看看有沒(méi)有可能給它“減個(gè)肥、瘦個(gè)身”。重新審視函數(shù)extractFields的類型之后，我們不難發(fā)現(xiàn)，這個(gè)函數(shù)從頭到尾無(wú)非是從Seq[Row]到Seq[(String,Int)]的轉(zhuǎn)換，函數(shù)體的邏輯就是字段提取，只要從Seq[Row]可以得到Seq[(String,Int)]，目的就達(dá)到了。要達(dá)成這兩種數(shù)據(jù)類型之間的轉(zhuǎn)換，除了利用上面這種開(kāi)發(fā)者信手拈來(lái)的過(guò)程式編程，我們還可以用函數(shù)式的編程范式。函數(shù)式編程的原則之一就是盡可能地在函數(shù)體中避免副作用（ieeffet），副作用指的是函數(shù)對(duì)于狀態(tài)的修改和變更，比如上例中exraFiels函數(shù)對(duì)于fiels變量不停地執(zhí)行追加操作就屬于副作用。基于這個(gè)想法，我們就有了第二種實(shí)現(xiàn)方式，如下所示。與第一種實(shí)現(xiàn)相比，它最大的區(qū)別在于去掉了fields變量。之后，為了達(dá)到同樣的效果，我們?cè)谳斎雲(yún)?shù)eqRow]上直接調(diào)用map操作逐一地提取特定字段并構(gòu)建元組，最后通過(guò)toSeq將映射轉(zhuǎn)換為序列，干凈利落，一氣呵成。//實(shí)現(xiàn)方案2——valextractFields:Seq[Row]=>Seq[(String,Int)]=(rows:Seq[Row])rows.map(row=>(row.getString(2),56你可能會(huì)問(wèn)：“兩份代碼實(shí)現(xiàn)無(wú)非是差了個(gè)中間變量而已，能有多大差別呢？看上去不過(guò)是代碼更簡(jiǎn)潔了而已。”事實(shí)上，我基于第二份代碼把ETL作業(yè)推上線后，就驚奇地發(fā)現(xiàn)端到端執(zhí)行性能提升了一倍！從原來(lái)的兩個(gè)小時(shí)縮短到一個(gè)小時(shí)。兩份功能完全一樣的代碼，在分布式環(huán)境中的執(zhí)行性能竟然有著成倍的差別。因此你看，在日常的開(kāi)發(fā)工作中，僅僅專注于業(yè)務(wù)功能實(shí)現(xiàn)還是不夠的，任何一個(gè)可以進(jìn)行調(diào)優(yōu)的小環(huán)節(jié)咱們都不能放過(guò)。你也許會(huì)說(shuō)：“你這個(gè)例子只是個(gè)例吧？更何況，這個(gè)例子里的優(yōu)化，僅僅是編程范式的調(diào)整，看上去和Sark似乎也沒(méi)什么關(guān)系啊！”不要緊，我們?cè)賮?lái)看第二個(gè)例子。第二個(gè)例子會(huì)稍微復(fù)雜一些，我們先來(lái)把業(yè)務(wù)需求和數(shù)據(jù)關(guān)系交代清楚。代代123456789(startDate,e.g.("2021-01-01","2021-01-valpairDF:DataFrame=(dim1,dim2,dim3,eventDate,e.g.("X","Y","Z","2021-01-15",valfactDF:DataFrame=//StoragerootvalrootPath:String=在這個(gè)案例中，我們有兩份數(shù)據(jù)，分別是pairDF和facDF，數(shù)據(jù)類型都是DaaFrame。第一份數(shù)據(jù)airDF的Shema包含兩個(gè)字段，分別是開(kāi)始日期和結(jié)束日期。第二份數(shù)據(jù)的字段較多，不過(guò)最主要的字段就兩個(gè)，一個(gè)是Eetae日期，另一個(gè)是業(yè)務(wù)關(guān)心的統(tǒng)計(jì)量，取名為ale。其他維度如im1、im2、im3主要用于數(shù)據(jù)分組，具體含義并不重要。從數(shù)據(jù)量來(lái)看，pairDF的數(shù)據(jù)量很小，大概幾百條記錄，facDF數(shù)據(jù)量很大，有上千萬(wàn)行。3步：pairDFfactDFEventdatedim1、dim2、dim3Eventdate4factDF將最終的統(tǒng)計(jì)結(jié)果落盤到AmazonS3代代123456789//實(shí)現(xiàn)方案1——defcreateInstance(factDF:DataFrame,startDate:String,endDate:String):DatvalinstanceDF=factDF.filter(col("eventDate")>lit(startDate)&&col("eventDate")<=.groupBy("dim1","dim2","dim3",.agg(sum("value")as"sum_value")}case(startDate:String,endDate:String)valinstance=createInstance(factDF,startDate,valoutPath=s"${rootPath}/endDate=${endDate}/startDate=${startDate}"}首先，他們是以factDF、開(kāi)始時(shí)間和結(jié)束時(shí)間為形參定義createInstance函數(shù)。在函數(shù)體中，先根據(jù)Eventdate對(duì)factDF進(jìn)行過(guò)濾，然后從4個(gè)維度分組匯總統(tǒng)計(jì)量，最后將匯總結(jié)果返回。定義完createInstance函數(shù)之后，收集pairDF到Driver端并逐條遍歷每一個(gè)時(shí)間對(duì)，然后以factDF、開(kāi)始時(shí)間、結(jié)束時(shí)間為實(shí)參調(diào)用createInstance函數(shù)，來(lái)獲取滿足過(guò)濾要求的匯總結(jié)果。最后，以Parquet的形式將結(jié)果落盤。同樣地，這段代碼從功能的角度來(lái)說(shuō)沒(méi)有任何問(wèn)題，而且從線上的結(jié)果來(lái)看，數(shù)據(jù)的處理邏輯也完全符合預(yù)期。不過(guò)，端到端的執(zhí)行性能可以說(shuō)是慘不忍睹，在16臺(tái)機(jī)型為C5.4xlareAWSEC2的分布式運(yùn)行環(huán)境中，基于上面這份代碼的ETL作業(yè)花費(fèi)了半個(gè)小時(shí)才執(zhí)行完畢。沒(méi)有對(duì)比就沒(méi)有，在同一份數(shù)據(jù)集之上，采用下面的第二種實(shí)現(xiàn)方式，僅用2臺(tái)同樣機(jī)型的EC2就能讓ETL作業(yè)在15分鐘以內(nèi)完成端到端的計(jì)算任務(wù)。兩份代碼的業(yè)務(wù)功能代代1234567//實(shí)現(xiàn)方案2——valinstances=.join(pairDF,factDF("eventDate")>pairDF("startDate")&&.groupBy("dim1","dim2","dim3","eventDate","startDate",.agg(sum("value")asinstances.write.partitionBy("endDate",我們都知道，觸發(fā)Spark延遲計(jì)算的Actions算子主要有兩類：一類是將分布式計(jì)算結(jié)果直接落盤的操作，如DataFrame的write、RDD的saveAsTextFile等；另一類是將分布式結(jié)果收集到Driver端的操作，如first、take、collect。顯然，對(duì)于第二類算子來(lái)說(shuō)，Drier有可能形成單點(diǎn)瓶頸，尤其是用collect算子去全量收集較大的結(jié)果集時(shí)，更容易出現(xiàn)性能問(wèn)題。因此，在第一種實(shí)現(xiàn)方式中，我們很容易就能嗅到collect這里的調(diào)用，味道很差。盡管collect這里味道不好，但在我們的場(chǎng)景里，pairDF畢竟是一份很小的數(shù)據(jù)集，才幾百條數(shù)據(jù)記錄而已，全量搜集到Driver端也不是什么大問(wèn)題。最的是collect后面的foreach。要知道，factDF是一份龐大的分布式數(shù)據(jù)集，盡管createInstance的邏輯僅僅是對(duì)factDF進(jìn)行過(guò)濾、匯總并落盤，但是createInstance函數(shù)在foreach中會(huì)被調(diào)用幾百次，pairDF中有多少個(gè)時(shí)間對(duì)，createInstance就會(huì)被調(diào)SparkDAGAction導(dǎo)致整條DAG從頭到尾重新執(zhí)行。明白了這一點(diǎn)之后，我們?cè)賮?lái)仔細(xì)觀察這份代碼，你品、你細(xì)品，目不轉(zhuǎn)睛地盯著foreach和eaeInstane中的facDF，你會(huì)驚訝地發(fā)現(xiàn)：有著上千萬(wàn)行數(shù)據(jù)的fa被反復(fù)掃描了幾百次！而且，是全量掃描喲！嚇不嚇人？可不可怕？這么分析下來(lái)，ETL作業(yè)端到端執(zhí)行效率低下的始作俑者，是不是就無(wú)遺了？反觀第二份代碼，factDF和pairDF用pairDF.startDate<factDF.eventDate<=pairDF.endDate的不等式條件進(jìn)行數(shù)據(jù)關(guān)聯(lián)。在Spark中，不等式Join的實(shí)現(xiàn)方式是NestedLoopJoin。盡管NestedLoopJoin是所有Join實(shí)現(xiàn)方式（MergeJoin，HashJoin，BroadcastJoin等）中性能的一種，而且這種Join方式?jīng)]有任何優(yōu)化空間，但factDF與pairDF今天，我們分析了兩個(gè)案例，這兩個(gè)案例都來(lái)自數(shù)據(jù)應(yīng)用的ETL場(chǎng)景。第一個(gè)案例講的是，在函數(shù)被頻繁調(diào)用的情況下，函數(shù)里面一個(gè)簡(jiǎn)單變量所引入的性能開(kāi)銷被成倍地放大。第二個(gè)例子講的是，不恰當(dāng)?shù)膶?shí)現(xiàn)方式導(dǎo)致海量數(shù)據(jù)被反復(fù)地掃描成百上千次。通過(guò)對(duì)這兩個(gè)案例進(jìn)行分析和探討，我們發(fā)現(xiàn)，對(duì)于Spark的應(yīng)用開(kāi)發(fā)，絕不僅僅是完成業(yè)務(wù)功能實(shí)現(xiàn)就高枕無(wú)憂了。Spark天生的執(zhí)行效率再高，也需要你針對(duì)具體的應(yīng)用場(chǎng)景而性能調(diào)優(yōu)的收益顯而易見(jiàn)：一來(lái)可以節(jié)約成本，尤其是按需的云上成本，更短的執(zhí)行時(shí)間意味著更少的花銷；二來(lái)可以提升開(kāi)發(fā)的迭代效率，尤其是對(duì)于從事數(shù)據(jù)分析、數(shù)據(jù)科學(xué)、機(jī)器學(xué)習(xí)的同學(xué)來(lái)說(shuō)，更高的執(zhí)行效率可以更快地獲取數(shù)據(jù)洞察，更快地找到模型收斂的最優(yōu)解。因此你看，性能調(diào)優(yōu)不是一件錦上添花的事情，而是開(kāi)發(fā)者必須要掌握的一項(xiàng)傍身技能。那么，對(duì)于Spark的性能調(diào)優(yōu)，你準(zhǔn)備好了嗎？生活不止眼前的茍且，讓我們來(lái)一場(chǎng)說(shuō)走 不得售賣。頁(yè)面已增加防盜追蹤，將依法其上一 開(kāi)篇詞|Spark性能調(diào)優(yōu)，你該掌握這些“套路下一 02|性能調(diào)優(yōu)的本質(zhì)：調(diào)優(yōu)的五花八門，該從哪里入手2 83 2Case12 2可以先將pairdfcollect到driver，再將數(shù)組按照startdate排序，然后再將其廣播。然后在fevent22021-03-valadArr=ArrayBuffer[Map[String,String]]()if(ads!=null)valadnum=ads.lengthfori0untiladnum)…沒(méi)有看到哪里定義分布式數(shù)據(jù)集，所有計(jì)算看上去是基于(0untilads.size())這個(gè)List，那么后1 12021-03-1 12021-03-請(qǐng)問(wèn)磊哥，spark里面nestedloopjoin和cartesianproductjion作者回復(fù):nlj是一種join實(shí)現(xiàn)方式哈，和hashjoin、sortmergejoin一樣，是一種join實(shí)現(xiàn)機(jī)制。cartesianjoin是一種join形式，和inner、left、right對(duì)等。每一種join形式，都可以用多種1 12021-03-編輯回復(fù):上傳到了B站 2021-03- 只用Scala嗎？