轉自(zi)//tech.meituan.com/spark-tuning-pro.html，感(gan)謝原作(zuo)者的(de)貢獻

前(qian)言(yan)

繼講解了每個Spark開發(fa)(fa)人(ren)員都必(bi)須熟知(zhi)的開發(fa)(fa)調(diao)(diao)優(you)與資源調(diao)(diao)優(you)之(zhi)后(hou)，本文作為《Spark性能優(you)化指南》的高級(ji)篇，將深入(ru)分析數據(ju)傾斜(xie)調(diao)(diao)優(you)與shuffle調(diao)(diao)優(you)，以解決更加(jia)棘手的性能問(wen)題。

數據(ju)傾斜調(diao)優

調優(you)概述

有(you)的(de)時(shi)候，我們可(ke)能會(hui)遇到(dao)大數據計算中(zhong)一個最棘手的(de)問題——數據傾(qing)斜，此時(shi)Spark作(zuo)業的(de)性(xing)能會(hui)比期望差很多。數據傾(qing)斜調(diao)優，就是使(shi)用各種技術方案解決不同(tong)類型的(de)數據傾(qing)斜問題，以保(bao)證Spark作(zuo)業的(de)性(xing)能。

數據(ju)傾斜發生時的現象

• 絕(jue)大多(duo)數task執(zhi)(zhi)行得都非常快，但個(ge)(ge)別task執(zhi)(zhi)行極慢。比如，總共有1000個(ge)(ge)task，997個(ge)(ge)task都在(zai)1分鐘之內執(zhi)(zhi)行完(wan)了(le)，但是剩(sheng)余兩三個(ge)(ge)task卻要(yao)一兩個(ge)(ge)小時。這種情況很常見。

• 原(yuan)本能夠正常執行的(de)(de)Spark作業，某天(tian)突然報出(chu)OOM（內存(cun)溢出(chu)）異常，觀察異常棧，是(shi)我(wo)們寫(xie)的(de)(de)業務(wu)代碼造成的(de)(de)。這種情況比較少見。

數據傾斜發(fa)生的原理

數(shu)據(ju)(ju)傾(qing)斜的(de)原理(li)(li)很簡單：在進(jin)(jin)行shuffle的(de)時(shi)候，必須將(jiang)各個節(jie)(jie)點(dian)上相同的(de)key拉(la)取到(dao)某個節(jie)(jie)點(dian)上的(de)一(yi)個task來進(jin)(jin)行處理(li)(li)，比(bi)如(ru)(ru)按照key進(jin)(jin)行聚合或join等操作。此時(shi)如(ru)(ru)果某個key對應的(de)數(shu)據(ju)(ju)量特別大(da)的(de)話(hua)，就(jiu)會(hui)(hui)發生數(shu)據(ju)(ju)傾(qing)斜。比(bi)如(ru)(ru)大(da)部分key對應10條數(shu)據(ju)(ju)，但(dan)(dan)是個別key卻對應了(le)100萬條數(shu)據(ju)(ju)，那(nei)(nei)么大(da)部分task可能就(jiu)只會(hui)(hui)分配到(dao)10條數(shu)據(ju)(ju)，然后1秒(miao)鐘(zhong)就(jiu)運(yun)行完(wan)了(le)；但(dan)(dan)是個別task可能分配到(dao)了(le)100萬數(shu)據(ju)(ju)，要(yao)運(yun)行一(yi)兩個小時(shi)。因此，整個Spark作業的(de)運(yun)行進(jin)(jin)度是由運(yun)行時(shi)間最長的(de)那(nei)(nei)個task決定的(de)。

因(yin)此(ci)出(chu)現數(shu)據(ju)傾(qing)斜的時候，Spark作(zuo)業看(kan)起來會運(yun)行得非常緩慢(man)，甚(shen)至可能因(yin)為某個task處(chu)理的數(shu)據(ju)量過大導(dao)致(zhi)內存(cun)溢(yi)出(chu)。

下(xia)圖就是(shi)一個很(hen)清(qing)晰的(de)例(li)子：hello這(zhe)個key，在(zai)三個節點上對(dui)應了總共7條(tiao)數(shu)據(ju)，這(zhe)些數(shu)據(ju)都(dou)會被拉取到同一個task中進行處(chu)理(li)；而world和you這(zhe)兩個key分別才對(dui)應1條(tiao)數(shu)據(ju)，所以另外(wai)兩個task只要分別處(chu)理(li)1條(tiao)數(shu)據(ju)即可。此(ci)時第(di)一個task的(de)運(yun)行時間可能是(shi)另外(wai)兩個task的(de)7倍，而整個stage的(de)運(yun)行速(su)度也由運(yun)行最慢的(de)那個task所決(jue)定。

如(ru)何定位導致數據傾(qing)斜的代碼

數(shu)據傾斜(xie)只會(hui)發生在shuffle過(guo)程中。這(zhe)里給大家羅(luo)列一些(xie)常用的(de)并且可(ke)能會(hui)觸發shuffle操作的(de)算子(zi)：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。出現數(shu)據傾斜(xie)時，可(ke)能就是你的(de)代碼(ma)中使用了這(zhe)些(xie)算子(zi)中的(de)某一個所導(dao)致的(de)。

某個task執行特別(bie)慢(man)的情況

首先要看的，就(jiu)是數據傾斜發生(sheng)在第幾個stage中。

如果是(shi)(shi)(shi)用yarn-client模(mo)式(shi)提(ti)交，那么本地是(shi)(shi)(shi)直(zhi)接可以(yi)(yi)看到(dao)log的，可以(yi)(yi)在log中找到(dao)當前運(yun)行到(dao)了(le)第幾個stage；如果是(shi)(shi)(shi)用yarn-cluster模(mo)式(shi)提(ti)交，則可以(yi)(yi)通過Spark Web UI來查看當前運(yun)行到(dao)了(le)第幾個stage。此外，無論(lun)是(shi)(shi)(shi)使(shi)用yarn-client模(mo)式(shi)還是(shi)(shi)(shi)yarn-cluster模(mo)式(shi)，我們(men)都可以(yi)(yi)在Spark Web UI上(shang)深入看一下當前這個stage各個task分(fen)(fen)配(pei)(pei)的數據量，從而進一步確(que)定是(shi)(shi)(shi)不是(shi)(shi)(shi)task分(fen)(fen)配(pei)(pei)的數據不均勻導致(zhi)了(le)數據傾斜。

比如下圖(tu)中(zhong)，倒數(shu)(shu)(shu)第三列(lie)顯(xian)(xian)示了(le)每(mei)個(ge)task的(de)(de)(de)運(yun)(yun)行(xing)(xing)時(shi)(shi)間。明(ming)顯(xian)(xian)可(ke)以看(kan)到(dao)，有的(de)(de)(de)task運(yun)(yun)行(xing)(xing)特別(bie)快(kuai)，只(zhi)需要(yao)幾(ji)秒鐘就可(ke)以運(yun)(yun)行(xing)(xing)完；而有的(de)(de)(de)task運(yun)(yun)行(xing)(xing)特別(bie)慢，需要(yao)幾(ji)分(fen)鐘才能運(yun)(yun)行(xing)(xing)完，此(ci)時(shi)(shi)單從運(yun)(yun)行(xing)(xing)時(shi)(shi)間上看(kan)就已經(jing)能夠確定發(fa)生數(shu)(shu)(shu)據(ju)(ju)傾斜了(le)。此(ci)外，倒數(shu)(shu)(shu)第一列(lie)顯(xian)(xian)示了(le)每(mei)個(ge)task處(chu)(chu)理的(de)(de)(de)數(shu)(shu)(shu)據(ju)(ju)量，明(ming)顯(xian)(xian)可(ke)以看(kan)到(dao)，運(yun)(yun)行(xing)(xing)時(shi)(shi)間特別(bie)短(duan)的(de)(de)(de)task只(zhi)需要(yao)處(chu)(chu)理幾(ji)百KB的(de)(de)(de)數(shu)(shu)(shu)據(ju)(ju)即可(ke)，而運(yun)(yun)行(xing)(xing)時(shi)(shi)間特別(bie)長的(de)(de)(de)task需要(yao)處(chu)(chu)理幾(ji)千KB的(de)(de)(de)數(shu)(shu)(shu)據(ju)(ju)，處(chu)(chu)理的(de)(de)(de)數(shu)(shu)(shu)據(ju)(ju)量差了(le)10倍。此(ci)時(shi)(shi)更加能夠確定是發(fa)生了(le)數(shu)(shu)(shu)據(ju)(ju)傾斜。

知(zhi)道數據傾(qing)斜發生(sheng)在哪一(yi)(yi)(yi)個stage之后，接著我們(men)(men)就需(xu)要根據stage劃分(fen)(fen)原理，推(tui)算出(chu)來發生(sheng)傾(qing)斜的(de)(de)那(nei)(nei)個stage對應代碼(ma)(ma)中(zhong)的(de)(de)哪一(yi)(yi)(yi)部(bu)分(fen)(fen)，這(zhe)部(bu)分(fen)(fen)代碼(ma)(ma)中(zhong)肯(ken)定會(hui)有一(yi)(yi)(yi)個shuffle類算子(zi)(zi)。精準(zhun)推(tui)算stage與代碼(ma)(ma)的(de)(de)對應關系，需(xu)要對Spark的(de)(de)源碼(ma)(ma)有深(shen)入的(de)(de)理解，這(zhe)里我們(men)(men)可(ke)以介(jie)紹一(yi)(yi)(yi)個相對簡(jian)單實(shi)用的(de)(de)推(tui)算方(fang)法(fa)：只要看(kan)到(dao)Spark代碼(ma)(ma)中(zhong)出(chu)現(xian)了(le)一(yi)(yi)(yi)個shuffle類算子(zi)(zi)或者是Spark SQL的(de)(de)SQL語(yu)句中(zhong)出(chu)現(xian)了(le)會(hui)導致shuffle的(de)(de)語(yu)句（比如group by語(yu)句），那(nei)(nei)么(me)就可(ke)以判(pan)定，以那(nei)(nei)個地方(fang)為(wei)界限劃分(fen)(fen)出(chu)了(le)前后兩(liang)個stage。

這(zhe)里我們就以(yi)(yi)Spark最(zui)基(ji)礎(chu)的(de)(de)入門程(cheng)序——單詞計數(shu)來舉例(li)，如何用最(zui)簡(jian)單的(de)(de)方法大致推算(suan)出一個stage對應(ying)的(de)(de)代碼。如下示(shi)例(li)，在整個代碼中，只有一個reduceByKey是會(hui)發(fa)生shuffle的(de)(de)算(suan)子(zi)，因此就可(ke)以(yi)(yi)認(ren)為(wei)，以(yi)(yi)這(zhe)個算(suan)子(zi)為(wei)界限，會(hui)劃(hua)分出前后(hou)兩個stage。

• stage0，主要是執(zhi)行從(cong)textFile到(dao)map操作(zuo)，以及執(zhi)行shuffle write操作(zuo)。shuffle write操作(zuo)，我們可以簡單理解為(wei)對pairs RDD中的數(shu)據進行分區操作(zuo)，每個task處理的數(shu)據中，相同的key會寫入同一個磁盤文件內。
• stage1，主要是(shi)執行(xing)(xing)(xing)(xing)從reduceByKey到(dao)collect操(cao)作，stage1的(de)各(ge)個task一開始(shi)運行(xing)(xing)(xing)(xing)，就(jiu)(jiu)會首先執行(xing)(xing)(xing)(xing)shuffle read操(cao)作。執行(xing)(xing)(xing)(xing)shuffle read操(cao)作的(de)task，會從stage0的(de)各(ge)個task所(suo)在(zai)(zai)節點拉取屬(shu)于自(zi)己(ji)處(chu)理(li)的(de)那(nei)些key，然(ran)后對同(tong)一個key進(jin)行(xing)(xing)(xing)(xing)全局性(xing)的(de)聚合(he)或join等操(cao)作，在(zai)(zai)這里(li)就(jiu)(jiu)是(shi)對key的(de)value值進(jin)行(xing)(xing)(xing)(xing)累(lei)加。stage1在(zai)(zai)執行(xing)(xing)(xing)(xing)完reduceByKey算子(zi)之后，就(jiu)(jiu)計算出了最終的(de)wordCounts RDD，然(ran)后會執行(xing)(xing)(xing)(xing)collect算子(zi)，將所(suo)有數據(ju)拉取到(dao)Driver上，供我(wo)們遍(bian)歷和打印(yin)輸出。

val conf = new SparkConf()
val sc = new SparkContext(conf)

val lines = sc.textFile("hdfs://...")
val words = lines.flatMap(_.split(" "))
val pairs = words.map((_, 1))
val wordCounts = pairs.reduceByKey(_ + _)

wordCounts.collect().foreach(println(_))

通過對單(dan)詞(ci)計數程序的(de)(de)(de)分(fen)析，希望能夠讓大(da)家了解最基(ji)本(ben)的(de)(de)(de)stage劃(hua)分(fen)的(de)(de)(de)原理，以(yi)及stage劃(hua)分(fen)后shuffle操作是(shi)如(ru)何在(zai)兩個stage的(de)(de)(de)邊(bian)界處執(zhi)行(xing)的(de)(de)(de)。然(ran)后我(wo)們(men)(men)就知道如(ru)何快速定位出(chu)發生數據(ju)(ju)傾(qing)斜的(de)(de)(de)stage對應代碼(ma)的(de)(de)(de)哪一個部(bu)分(fen)了。比(bi)如(ru)我(wo)們(men)(men)在(zai)Spark Web UI或者本(ben)地(di)log中(zhong)(zhong)發現(xian)，stage1的(de)(de)(de)某(mou)幾個task執(zhi)行(xing)得(de)特別(bie)慢，判定stage1出(chu)現(xian)了數據(ju)(ju)傾(qing)斜，那(nei)(nei)么就可以(yi)回到代碼(ma)中(zhong)(zhong)定位出(chu)stage1主要(yao)包括(kuo)了reduceByKey這(zhe)個shuffle類算子，此時基(ji)本(ben)就可以(yi)確(que)定是(shi)由educeByKey算子導致的(de)(de)(de)數據(ju)(ju)傾(qing)斜問題。比(bi)如(ru)某(mou)個單(dan)詞(ci)出(chu)現(xian)了100萬(wan)次(ci)，其他(ta)單(dan)詞(ci)才出(chu)現(xian)10次(ci)，那(nei)(nei)么stage1的(de)(de)(de)某(mou)個task就要(yao)處理100萬(wan)數據(ju)(ju)，整個stage的(de)(de)(de)速度就會被(bei)這(zhe)個task拖慢。

某個task莫名(ming)其妙內存溢出的情況

這種情況下去(qu)定位出(chu)問題(ti)的(de)(de)代碼就(jiu)(jiu)比較容易了。我們(men)建議直接(jie)看yarn-client模(mo)式(shi)下本地log的(de)(de)異常棧(zhan)(zhan)，或(huo)者(zhe)是通過YARN查看yarn-cluster模(mo)式(shi)下的(de)(de)log中(zhong)的(de)(de)異常棧(zhan)(zhan)。一般(ban)來說(shuo)，通過異常棧(zhan)(zhan)信息就(jiu)(jiu)可以定位到你的(de)(de)代碼中(zhong)哪一行發生(sheng)了內存溢出(chu)。然后在那(nei)行代碼附近找找，一般(ban)也(ye)會有(you)shuffle類(lei)算(suan)子，此時很可能(neng)就(jiu)(jiu)是這個算(suan)子導致了數(shu)據傾斜。

但是(shi)大家要注意的(de)(de)(de)(de)是(shi)，不能單純(chun)靠偶然的(de)(de)(de)(de)內存溢(yi)出(chu)就(jiu)判定發生了數據(ju)(ju)傾(qing)斜(xie)。因為自己編寫(xie)的(de)(de)(de)(de)代碼的(de)(de)(de)(de)bug，以及(ji)偶然出(chu)現的(de)(de)(de)(de)數據(ju)(ju)異常，也可(ke)能會導致(zhi)(zhi)內存溢(yi)出(chu)。因此還是(shi)要按照上面(mian)所講(jiang)的(de)(de)(de)(de)方法，通過Spark Web UI查(cha)看(kan)報(bao)錯的(de)(de)(de)(de)那個stage的(de)(de)(de)(de)各(ge)個task的(de)(de)(de)(de)運(yun)行時間以及(ji)分(fen)配(pei)的(de)(de)(de)(de)數據(ju)(ju)量，才能確定是(shi)否是(shi)由于數據(ju)(ju)傾(qing)斜(xie)才導致(zhi)(zhi)了這(zhe)次內存溢(yi)出(chu)。

查看導致數(shu)據(ju)傾斜(xie)的key的數(shu)據(ju)分布情況

知道了(le)數據(ju)傾斜發生在哪里之后(hou)，通常需要分(fen)析一(yi)下(xia)那個執行了(le)shuffle操作并且(qie)導(dao)致了(le)數據(ju)傾斜的(de)(de)(de)RDD/Hive表，查看一(yi)下(xia)其中key的(de)(de)(de)分(fen)布情(qing)況(kuang)。這主(zhu)要是為之后(hou)選擇(ze)哪一(yi)種技術方案提供依(yi)據(ju)。針對不(bu)同的(de)(de)(de)key分(fen)布與不(bu)同的(de)(de)(de)shuffle算(suan)子組合起來(lai)的(de)(de)(de)各(ge)種情(qing)況(kuang)，可能需要選擇(ze)不(bu)同的(de)(de)(de)技術方案來(lai)解決。

此時根據你執(zhi)行操作的情(qing)況不同(tong)，可以(yi)有很多(duo)種(zhong)查看key分布的方(fang)式(shi)：

1. 如果(guo)是Spark SQL中(zhong)的(de)group by、join語句導致(zhi)的(de)數(shu)據傾斜，那么就查詢一下SQL中(zhong)使用的(de)表的(de)key分布情況。
2. 如果是對Spark RDD執行shuffle算子導(dao)致的數據傾斜，那么可以在Spark作業中加入查看(kan)key分(fen)布(bu)(bu)的代碼，比如RDD.countByKey()。然后對統計出(chu)(chu)來的各個(ge)key出(chu)(chu)現的次(ci)數，collect/take到(dao)客戶端打印一(yi)下(xia)，就可以看(kan)到(dao)key的分(fen)布(bu)(bu)情況。

舉例(li)來(lai)說(shuo)(shuo)，對于(yu)上(shang)面所說(shuo)(shuo)的(de)單詞計(ji)數(shu)(shu)程序，如果確(que)定(ding)了是stage1的(de)reduceByKey算子(zi)導致了數(shu)(shu)據傾斜，那么就(jiu)應該看看進行reduceByKey操作的(de)RDD中(zhong)的(de)key分布情況，在(zai)這個(ge)例(li)子(zi)中(zhong)指(zhi)的(de)就(jiu)是pairs RDD。如下(xia)示例(li)，我們可以先對pairs采樣10%的(de)樣本數(shu)(shu)據，然后(hou)使用(yong)countByKey算子(zi)統(tong)計(ji)出每個(ge)key出現(xian)的(de)次(ci)數(shu)(shu)，最后(hou)在(zai)客戶端(duan)遍歷和打(da)印樣本數(shu)(shu)據中(zhong)各個(ge)key的(de)出現(xian)次(ci)數(shu)(shu)。

val sampledPairs = pairs.sample(false, 0.1)
val sampledWordCounts = sampledPairs.countByKey()
sampledWordCounts.foreach(println(_))

數(shu)據(ju)傾斜的(de)解(jie)決方案

解決方案(an)一：使(shi)用Hive ETL預處(chu)理數據

方案適用場景：導致數據傾斜的是Hive表。如果該Hive表中的數據本身很不均勻（比如某個key對應了100萬數據，其他key才對應了10條數據），而且業務場景需要頻繁使用Spark對Hive表執行某個分析操作，那么比較適合使用這種技術方案。

方案實現思路：此時可以評估一下，是否可以通過Hive來進行數據預處理（即通過Hive ETL預先對數據按照key進行聚合，或者是預先和其他表進行join），然后在Spark作業中針對的數據源就不是原來的Hive表了，而是預處理后的Hive表。此時由于數據已經預先進行過聚合或join操作了，那么在Spark作業中也就不需要使用原先的shuffle類算子執行這類操作了。

方案實現原理：這種方案從根源上解決了數據傾斜，因為徹底避免了在Spark中執行shuffle類算子，那么肯定就不會有數據傾斜的問題了。但是這里也要提醒一下大家，這種方式屬于治標不治本。因為畢竟數據本身就存在分布不均勻的問題，所以Hive ETL中進行group by或者join等shuffle操作時，還是會出現數據傾斜，導致Hive ETL的速度很慢。我們只是把數據傾斜的發生提前到了Hive ETL中，避免Spark程序發生數據傾斜而已。

方案優點：實現起來簡單便捷，效果還非常好，完全規避掉了數據傾斜，Spark作業的性能會大幅度提升。

方案缺點：治標不治本，Hive ETL中還是會發生數據傾斜。

方案實踐經驗：在一些Java系統與Spark結合使用的項目中，會出現Java代碼頻繁調用Spark作業的場景，而且對Spark作業的執行性能要求很高，就比較適合使用這種方案。將數據傾斜提前到上游的Hive ETL，每天僅執行一次，只有那一次是比較慢的，而之后每次Java調用Spark作業時，執行速度都會很快，能夠提供更好的用戶體驗。

項目實踐經驗：在美團·點評的交互式用戶行為分析系統中使用了這種方案，該系統主要是允許用戶通過Java Web系統提交數據分析統計任務，后端通過Java提交Spark作業進行數據分析統計。要求Spark作業速度必須要快，盡量在10分鐘以內，否則速度太慢，用戶體驗會很差。所以我們將有些Spark作業的shuffle操作提前到了Hive ETL中，從而讓Spark直接使用預處理的Hive中間表，盡可能地減少Spark的shuffle操作，大幅度提升了性能，將部分作業的性能提升了6倍以上。

解決(jue)方案二(er)：過濾少數導致傾斜的key

方案適用場景：如果發現導致傾斜的key就少數幾個，而且對計算本身的影響并不大的話，那么很適合使用這種方案。比如99%的key就對應10條數據，但是只有一個key對應了100萬數據，從而導致了數據傾斜。

方案實現思路：如果我們判斷那少數幾個數據量特別多的key，對作業的執行和計算結果不是特別重要的話，那么干脆就直接過濾掉那少數幾個key。比如，在Spark SQL中可以使用where子句過濾掉這些key或者在Spark Core中對RDD執行filter算子過濾掉這些key。如果需要每次作業執行時，動態判定哪些key的數據量最多然后再進行過濾，那么可以使用sample算子對RDD進行采樣，然后計算出每個key的數量，取數據量最多的key過濾掉即可。

方案實現原理：將導致數據傾斜的key給過濾掉之后，這些key就不會參與計算了，自然不可能產生數據傾斜。

方案優點：實現簡單，而且效果也很好，可以完全規避掉數據傾斜。

方案缺點：適用場景不多，大多數情況下，導致傾斜的key還是很多的，并不是只有少數幾個。

方案實踐經驗：在項目中我們也采用過這種方案解決數據傾斜。有一次發現某一天Spark作業在運行的時候突然OOM了，追查之后發現，是Hive表中的某一個key在那天數據異常，導致數據量暴增。因此就采取每次執行前先進行采樣，計算出樣本中數據量最大的幾個key之后，直接在程序中將那些key給過濾掉。

解決方(fang)案三：提高shuffle操(cao)作的并行度

方案適用場景：如果我們必須要對數據傾斜迎難而上，那么建議優先使用這種方案，因為這是處理數據傾斜最簡單的一種方案。

方案實現思路：在對RDD執行shuffle算子時，給shuffle算子傳入一個參數，比如reduceByKey(1000)，該參數就設置了這個shuffle算子執行時shuffle read task的數量。對于Spark SQL中的shuffle類語句，比如group by、join等，需要設置一個參數，即spark.sql.shuffle.partitions，該參數代表了shuffle read task的并行度，該值默認是200，對于很多場景來說都有點過小。

方案實現原理：增加shuffle read task的數量，可以讓原本分配給一個task的多個key分配給多個task，從而讓每個task處理比原來更少的數據。舉例來說，如果原本有5個key，每個key對應10條數據，這5個key都是分配給一個task的，那么這個task就要處理50條數據。而增加了shuffle read task以后，每個task就分配到一個key，即每個task就處理10條數據，那么自然每個task的執行時間都會變短了。具體原理如下圖所示。

方案優點：實現起來比較簡單，可以有效緩解和減輕數據傾斜的影響。

方案缺點：只是緩解了數據傾斜而已，沒有徹底根除問題，根據實踐經驗來看，其效果有限。

方案實踐經驗：該方案通常無法徹底解決數據傾斜，因為如果出現一些極端情況，比如某個key對應的數據量有100萬，那么無論你的task數量增加到多少，這個對應著100萬數據的key肯定還是會分配到一個task中去處理，因此注定還是會發生數據傾斜的。所以這種方案只能說是在發現數據傾斜時嘗試使用的第一種手段，嘗試去用嘴簡單的方法緩解數據傾斜而已，或者是和其他方案結合起來使用。

解決方案四：兩階段聚合(he)（局部聚合(he)+全局聚合(he)）

方案適用場景：對RDD執行reduceByKey等聚合類shuffle算子或者在Spark SQL中使用group by語句進行分組聚合時，比較適用這種方案。

方案實現思路：這個方案的核心實現思路就是進行兩階段聚合。第一次是局部聚合，先給每個key都打上一個隨機數，比如10以內的隨機數，此時原先一樣的key就變成不一樣的了，比如(hello, 1) (hello, 1) (hello, 1) (hello, 1)，就會變成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接著對打上隨機數后的數據，執行reduceByKey等聚合操作，進行局部聚合，那么局部聚合結果，就會變成了(1_hello, 2) (2_hello, 2)。然后將各個key的前綴給去掉，就會變成(hello,2)(hello,2)，再次進行全局聚合操作，就可以得到最終結果了，比如(hello, 4)。

方案實現原理：將原本相同的key通過附加隨機前綴的方式，變成多個不同的key，就可以讓原本被一個task處理的數據分散到多個task上去做局部聚合，進而解決單個task處理數據量過多的問題。接著去除掉隨機前綴，再次進行全局聚合，就可以得到最終的結果。具體原理見下圖。

方案優點：對于聚合類的shuffle操作導致的數據傾斜，效果是非常不錯的。通常都可以解決掉數據傾斜，或者至少是大幅度緩解數據傾斜，將Spark作業的性能提升數倍以上。

方案缺點：僅僅適用于聚合類的shuffle操作，適用范圍相對較窄。如果是join類的shuffle操作，還得用其他的解決方案。

// 第一步，給RDD中的每個(ge)(ge)key都打上一個(ge)(ge)隨機前綴。
JavaPairRDD<String, Long> randomPrefixRdd = rdd.mapToPair(
        new PairFunction<Tuple2<Long,Long>, String, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, Long> call(Tuple2<Long, Long> tuple)
                    throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(10);
                return new Tuple2<String, Long>(prefix + "_" + tuple._1, tuple._2);
            }
        });

// 第二步，對打(da)上(shang)隨(sui)機前綴的key進行局(ju)部聚合。
JavaPairRDD<String, Long> localAggrRdd = randomPrefixRdd.reduceByKey(
        new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });

// 第三步(bu)，去除RDD中每個key的隨機前綴。
JavaPairRDD<Long, Long> removedRandomPrefixRdd = localAggrRdd.mapToPair(
        new PairFunction<Tuple2<String,Long>, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<Long, Long> call(Tuple2<String, Long> tuple)
                    throws Exception {
                long originalKey = Long.valueOf(tuple._1.split("_")[1]);
                return new Tuple2<Long, Long>(originalKey, tuple._2);
            }
        });

// 第四(si)步，對去除了隨機前(qian)綴(zhui)的RDD進行全局聚(ju)合。
JavaPairRDD<Long, Long> globalAggrRdd = removedRandomPrefixRdd.reduceByKey(
        new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });

解決方案(an)五：將reduce join轉(zhuan)為map join

方案適用場景：在對RDD使用join類操作，或者是在Spark SQL中使用join語句時，而且join操作中的一個RDD或表的數據量比較小（比如幾百M或者一兩G），比較適用此方案。

方案實現思路：不使用join算子進行連接操作，而使用Broadcast變量與map類算子實現join操作，進而完全規避掉shuffle類的操作，徹底避免數據傾斜的發生和出現。將較小RDD中的數據直接通過collect算子拉取到Driver端的內存中來，然后對其創建一個Broadcast變量；接著對另外一個RDD執行map類算子，在算子函數內，從Broadcast變量中獲取較小RDD的全量數據，與當前RDD的每一條數據按照連接key進行比對，如果連接key相同的話，那么就將兩個RDD的數據用你需要的方式連接起來。

方案實現原理：普通的join是會走shuffle過程的，而一旦shuffle，就相當于會將相同key的數據拉取到一個shuffle read task中再進行join，此時就是reduce join。但是如果一個RDD是比較小的，則可以采用廣播小RDD全量數據+map算子來實現與join同樣的效果，也就是map join，此時就不會發生shuffle操作，也就不會發生數據傾斜。具體原理如下圖所示。

方案優點：對join操作導致的數據傾斜，效果非常好，因為根本就不會發生shuffle，也就根本不會發生數據傾斜。

方案缺點：適用場景較少，因為這個方案只適用于一個大表和一個小表的情況。畢竟我們需要將小表進行廣播，此時會比較消耗內存資源，driver和每個Executor內存中都會駐留一份小RDD的全量數據。如果我們廣播出去的RDD數據比較大，比如10G以上，那么就可能發生內存溢出了。因此并不適合兩個都是大表的情況。

// 首先將(jiang)數據量比較小的RDD的數據，collect到Driver中來。
List<Tuple2<Long, Row>> rdd1Data = rdd1.collect()
// 然后使用Spark的(de)廣播功能，將(jiang)小RDD的(de)數據轉換成廣播變量，這(zhe)樣每個Executor就只有一份(fen)RDD的(de)數據。
// 可以(yi)盡可能節省內存空間，并且減少網絡傳(chuan)輸性(xing)能開銷。
final Broadcast<List<Tuple2<Long, Row>>> rdd1DataBroadcast = sc.broadcast(rdd1Data);

// 對(dui)另外一個RDD執行map類(lei)操作，而(er)不(bu)再是(shi)join類(lei)操作。
JavaPairRDD<String, Tuple2<String, Row>> joinedRdd = rdd2.mapToPair(
        new PairFunction<Tuple2<Long,String>, String, Tuple2<String, Row>>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, Tuple2<String, Row>> call(Tuple2<Long, String> tuple)
                    throws Exception {
                // 在(zai)算子(zi)函數中，通過廣播變量(liang)，獲取(qu)到本地Executor中的rdd1數據。
                List<Tuple2<Long, Row>> rdd1Data = rdd1DataBroadcast.value();
                // 可以將(jiang)rdd1的數據轉換為一(yi)個Map，便(bian)于后面進行(xing)join操作。
                Map<Long, Row> rdd1DataMap = new HashMap<Long, Row>();
                for(Tuple2<Long, Row> data : rdd1Data) {
                    rdd1DataMap.put(data._1, data._2);
                }
                // 獲取當前(qian)RDD數據的key以(yi)及(ji)value。
                String key = tuple._1;
                String value = tuple._2;
                // 從rdd1數據Map中，根據key獲取到可以join到的數據。
                Row rdd1Value = rdd1DataMap.get(key);
                return new Tuple2<String, String>(key, new Tuple2<String, Row>(value, rdd1Value));
            }
        });

// 這里得提(ti)示一下。
// 上面的(de)做法，僅(jin)僅(jin)適用于rdd1中的(de)key沒有(you)重復(fu)，全部是唯一的(de)場景。
// 如果rdd1中有(you)多個(ge)相(xiang)同的(de)key，那么就得用(yong)flatMap類的(de)操作(zuo)，在進行(xing)join的(de)時候不能用(yong)map，而(er)是得遍歷rdd1所(suo)有(you)數據進行(xing)join。
// rdd2中每(mei)條(tiao)數據都可能會返回(hui)多條(tiao)join后的數據。

解決方案六：采樣傾斜key并分拆(chai)join操(cao)作

方案適用場景：兩個RDD/Hive表進行join的時候，如果數據量都比較大，無法采用“解決方案五”，那么此時可以看一下兩個RDD/Hive表中的key分布情況。如果出現數據傾斜，是因為其中某一個RDD/Hive表中的少數幾個key的數據量過大，而另一個RDD/Hive表中的所有key都分布比較均勻，那么采用這個解決方案是比較合適的。

方案實現思路：

• 對包(bao)含少(shao)數幾(ji)個(ge)數據(ju)量(liang)過大的(de)(de)key的(de)(de)那個(ge)RDD，通過sample算子采(cai)樣出一份樣本(ben)來，然后統計(ji)一下每個(ge)key的(de)(de)數量(liang)，計(ji)算出來數據(ju)量(liang)最大的(de)(de)是哪幾(ji)個(ge)key。
• 然(ran)后將(jiang)這幾個(ge)key對應的(de)數據從原(yuan)來(lai)的(de)RDD中拆分(fen)出來(lai)，形(xing)成一(yi)個(ge)單獨的(de)RDD，并(bing)給每個(ge)key都打上n以內的(de)隨(sui)機(ji)數作為前綴，而不會導(dao)致傾斜的(de)大(da)部分(fen)key形(xing)成另外(wai)一(yi)個(ge)RDD。
• 接(jie)著將(jiang)(jiang)需(xu)要join的(de)另一個RDD，也(ye)過(guo)濾出來那幾個傾斜key對應的(de)數據(ju)(ju)并形(xing)成(cheng)一個單獨的(de)RDD，將(jiang)(jiang)每條(tiao)數據(ju)(ju)膨脹成(cheng)n條(tiao)數據(ju)(ju)，這n條(tiao)數據(ju)(ju)都按順序附(fu)加(jia)一個0~n的(de)前(qian)綴，不會導致傾斜的(de)大部分key也(ye)形(xing)成(cheng)另外一個RDD。
• 再將(jiang)(jiang)附加了(le)隨機前綴的(de)(de)獨(du)立RDD與另一(yi)個(ge)膨(peng)脹(zhang)n倍的(de)(de)獨(du)立RDD進行join，此(ci)時就可以將(jiang)(jiang)原先相同的(de)(de)key打散成n份，分散到多個(ge)task中去進行join了(le)。
• 而(er)另外兩個普通的(de)RDD就(jiu)照常join即可。
• 最后將兩次join的結(jie)果使用union算子合并起(qi)來即可，就是最終的join結(jie)果。

方案實現原理：對于join導致的數據傾斜，如果只是某幾個key導致了傾斜，可以將少數幾個key分拆成獨立RDD，并附加隨機前綴打散成n份去進行join，此時這幾個key對應的數據就不會集中在少數幾個task上，而是分散到多個task進行join了。具體原理見下圖。

方案優點：對于join導致的數據傾斜，如果只是某幾個key導致了傾斜，采用該方式可以用最有效的方式打散key進行join。而且只需要針對少數傾斜key對應的數據進行擴容n倍，不需要對全量數據進行擴容。避免了占用過多內存。

方案缺點：如果導致傾斜的key特別多的話，比如成千上萬個key都導致數據傾斜，那么這種方式也不適合。

// 首(shou)先從包含了(le)少數幾個導(dao)致數據傾(qing)斜key的rdd1中，采樣10%的樣本數據。
JavaPairRDD<Long, String> sampledRDD = rdd1.sample(false, 0.1);

// 對樣本(ben)數(shu)據RDD統(tong)計出每(mei)個(ge)key的出現次數(shu)，并按(an)出現次數(shu)降序排序。
// 對(dui)降序(xu)排序(xu)后的(de)(de)數據(ju)，取出top 1或(huo)者top 100的(de)(de)數據(ju)，也就是key最多(duo)的(de)(de)前n個數據(ju)。
// 具體取(qu)出多少個數據量最多的key，由大(da)家(jia)自己決定，我們這里就取(qu)1個作為示范。
JavaPairRDD<Long, Long> mappedSampledRDD = sampledRDD.mapToPair(
        new PairFunction<Tuple2<Long,String>, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<Long, Long> call(Tuple2<Long, String> tuple)
                    throws Exception {
                return new Tuple2<Long, Long>(tuple._1, 1L);
            }     
        });
JavaPairRDD<Long, Long> countedSampledRDD = mappedSampledRDD.reduceByKey(
        new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });
JavaPairRDD<Long, Long> reversedSampledRDD = countedSampledRDD.mapToPair( 
        new PairFunction<Tuple2<Long,Long>, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<Long, Long> call(Tuple2<Long, Long> tuple)
                    throws Exception {
                return new Tuple2<Long, Long>(tuple._2, tuple._1);
            }
        });
final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2;

// 從rdd1中分拆出(chu)導(dao)致數據傾斜的key，形成獨立(li)的RDD。
JavaPairRDD<Long, String> skewedRDD = rdd1.filter(
        new Function<Tuple2<Long,String>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<Long, String> tuple) throws Exception {
                return tuple._1.equals(skewedUserid);
            }
        });
// 從rdd1中分拆出不導致數據傾斜的普通key，形成(cheng)獨立的RDD。
JavaPairRDD<Long, String> commonRDD = rdd1.filter(
        new Function<Tuple2<Long,String>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<Long, String> tuple) throws Exception {
                return !tuple._1.equals(skewedUserid);
            } 
        });

// rdd2，就是那個所有(you)key的分布相對較為均(jun)勻的rdd。
// 這里將rdd2中，前面獲取到的key對(dui)應的數據(ju)，過濾出來，分拆成(cheng)單獨的rdd，并對(dui)rdd中的數據(ju)使用flatMap算子都擴(kuo)容100倍。
// 對擴(kuo)容的每條數據，都打上0～100的前(qian)綴(zhui)。
JavaPairRDD<String, Row> skewedRdd2 = rdd2.filter(
         new Function<Tuple2<Long,Row>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<Long, Row> tuple) throws Exception {
                return tuple._1.equals(skewedUserid);
            }
        }).flatMapToPair(new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Iterable<Tuple2<String, Row>> call(
                    Tuple2<Long, Row> tuple) throws Exception {
                Random random = new Random();
                List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();
                for(int i = 0; i < 100; i++) {
                    list.add(new Tuple2<String, Row>(i + "_" + tuple._1, tuple._2));
                }
                return list;
            }

        });

// 將rdd1中分拆(chai)出來的(de)(de)導致傾斜的(de)(de)key的(de)(de)獨立(li)rdd，每條(tiao)數據都打(da)上100以(yi)內的(de)(de)隨機(ji)前綴。
// 然(ran)后將這個rdd1中分拆出(chu)來的獨(du)立rdd，與(yu)上面(mian)rdd2中分拆出(chu)來的獨(du)立rdd，進行join。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD1 = skewedRDD.mapToPair(
        new PairFunction<Tuple2<Long,String>, String, String>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, String> call(Tuple2<Long, String> tuple)
                    throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(100);
                return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
            }
        })
        .join(skewedUserid2infoRDD)
        .mapToPair(new PairFunction<Tuple2<String,Tuple2<String,Row>>, Long, Tuple2<String, Row>>() {
                        private static final long serialVersionUID = 1L;
                        @Override
                        public Tuple2<Long, Tuple2<String, Row>> call(
                            Tuple2<String, Tuple2<String, Row>> tuple)
                            throws Exception {
                            long key = Long.valueOf(tuple._1.split("_")[1]);
                            return new Tuple2<Long, Tuple2<String, Row>>(key, tuple._2);
                        }
                    });

// 將(jiang)rdd1中分拆(chai)出來的包(bao)含(han)普通key的獨立(li)rdd，直接與rdd2進行join。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD2 = commonRDD.join(rdd2);

// 將傾(qing)斜(xie)key join后的結果與普通key join后的結果，uinon起來。
// 就是最終(zhong)的join結(jie)果(guo)。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD = joinedRDD1.union(joinedRDD2);

解決方案七：使用隨機前綴和擴容(rong)RDD進行join

方案適用場景：如果在進行join操作時，RDD中有大量的key導致數據傾斜，那么進行分拆key也沒什么意義，此時就只能使用最后一種方案來解決問題了。

方案實現思路：

• 該方案的(de)(de)實現思路(lu)基本和“解決方案六”類似，首先(xian)查(cha)看RDD/Hive表(biao)中(zhong)的(de)(de)數據(ju)分布情況，找到那個造(zao)成數據(ju)傾斜的(de)(de)RDD/Hive表(biao)，比如有(you)多個key都對應了超過1萬條數據(ju)。
• 然后將該RDD的(de)每條數據(ju)都打上(shang)一個(ge)n以(yi)內的(de)隨機前綴。
• 同時對(dui)另外一個正(zheng)常(chang)的(de)RDD進行擴(kuo)容，將每(mei)條(tiao)數據都(dou)擴(kuo)容成(cheng)n條(tiao)數據，擴(kuo)容出(chu)來的(de)每(mei)條(tiao)數據都(dou)依(yi)次打上一個0~n的(de)前(qian)綴。
• 最后(hou)將(jiang)兩(liang)個處理后(hou)的RDD進行join即可。

方案實現原理：將原先一樣的key通過附加隨機前綴變成不一樣的key，然后就可以將這些處理后的“不同key”分散到多個task中去處理，而不是讓一個task處理大量的相同key。該方案與“解決方案六”的不同之處就在于，上一種方案是盡量只對少數傾斜key對應的數據進行特殊處理，由于處理過程需要擴容RDD，因此上一種方案擴容RDD后對內存的占用并不大；而這一種方案是針對有大量傾斜key的情況，沒法將部分key拆分出來進行單獨處理，因此只能對整個RDD進行數據擴容，對內存資源要求很高。

方案優點：對join類型的數據傾斜基本都可以處理，而且效果也相對比較顯著，性能提升效果非常不錯。

方案缺點：該方案更多的是緩解數據傾斜，而不是徹底避免數據傾斜。而且需要對整個RDD進行擴容，對內存資源要求很高。

方案實踐經驗：曾經開發一個數據需求的時候，發現一個join導致了數據傾斜。優化之前，作業的執行時間大約是60分鐘左右；使用該方案優化之后，執行時間縮短到10分鐘左右，性能提升了6倍。

// 首先將其(qi)中一個key分布(bu)相對較為均勻的RDD膨脹100倍。
JavaPairRDD<String, Row> expandedRDD = rdd1.flatMapToPair(
        new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Iterable<Tuple2<String, Row>> call(Tuple2<Long, Row> tuple)
                    throws Exception {
                List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();
                for(int i = 0; i < 100; i++) {
                    list.add(new Tuple2<String, Row>(0 + "_" + tuple._1, tuple._2));
                }
                return list;
            }
        });

// 其次，將另(ling)一個有數據傾斜(xie)key的(de)(de)RDD，每條數據都打上100以內的(de)(de)隨機(ji)前綴。
JavaPairRDD<String, String> mappedRDD = rdd2.mapToPair(
        new PairFunction<Tuple2<Long,String>, String, String>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, String> call(Tuple2<Long, String> tuple)
                    throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(100);
                return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
            }
        });

// 將兩個處理后(hou)的RDD進行join即可。
JavaPairRDD<String, Tuple2<String, Row>> joinedRDD = mappedRDD.join(expandedRDD);

解(jie)決方案八：多種方案組合使用

在實踐(jian)中發現，很多(duo)情況下(xia)，如(ru)果(guo)只(zhi)是(shi)處(chu)(chu)理較為簡單(dan)的(de)(de)(de)(de)數(shu)(shu)據(ju)(ju)(ju)傾(qing)(qing)斜場(chang)景，那么使(shi)用上述方(fang)案(an)(an)中的(de)(de)(de)(de)某一種基本就可(ke)以(yi)(yi)解決(jue)。但是(shi)如(ru)果(guo)要(yao)處(chu)(chu)理一個(ge)較為復雜的(de)(de)(de)(de)數(shu)(shu)據(ju)(ju)(ju)傾(qing)(qing)斜場(chang)景，那么可(ke)能需(xu)要(yao)將多(duo)種方(fang)案(an)(an)組合起來(lai)使(shi)用。比如(ru)說，我們針(zhen)對(dui)出(chu)現了多(duo)個(ge)數(shu)(shu)據(ju)(ju)(ju)傾(qing)(qing)斜環節的(de)(de)(de)(de)Spark作業，可(ke)以(yi)(yi)先運用解決(jue)方(fang)案(an)(an)一和二，預(yu)處(chu)(chu)理一部分(fen)數(shu)(shu)據(ju)(ju)(ju)，并(bing)過濾(lv)一部分(fen)數(shu)(shu)據(ju)(ju)(ju)來(lai)緩(huan)解；其(qi)次可(ke)以(yi)(yi)對(dui)某些(xie)shuffle操(cao)作提升并(bing)行度(du)，優化(hua)其(qi)性能；最后還可(ke)以(yi)(yi)針(zhen)對(dui)不同(tong)的(de)(de)(de)(de)聚合或(huo)join操(cao)作，選擇一種方(fang)案(an)(an)來(lai)優化(hua)其(qi)性能。大家(jia)需(xu)要(yao)對(dui)這些(xie)方(fang)案(an)(an)的(de)(de)(de)(de)思路和原理都透徹理解之后，在實踐(jian)中根據(ju)(ju)(ju)各種不同(tong)的(de)(de)(de)(de)情況，靈活運用多(duo)種方(fang)案(an)(an)，來(lai)解決(jue)自己的(de)(de)(de)(de)數(shu)(shu)據(ju)(ju)(ju)傾(qing)(qing)斜問(wen)題。

shuffle調優(you)

調(diao)優概(gai)述(shu)

大(da)(da)多數(shu)Spark作(zuo)業(ye)的(de)(de)性(xing)能主要(yao)就(jiu)是消耗(hao)在了(le)shuffle環(huan)節，因為該(gai)環(huan)節包含了(le)大(da)(da)量的(de)(de)磁(ci)盤IO、序列化、網絡數(shu)據傳輸等(deng)操作(zuo)。因此(ci)(ci)，如(ru)果要(yao)讓(rang)作(zuo)業(ye)的(de)(de)性(xing)能更上一層(ceng)樓，就(jiu)有(you)必(bi)要(yao)對shuffle過(guo)程進行調(diao)優(you)(you)。但是也必(bi)須提醒大(da)(da)家(jia)的(de)(de)是，影響一個Spark作(zuo)業(ye)性(xing)能的(de)(de)因素，主要(yao)還是代碼開發、資源參(can)數(shu)以及數(shu)據傾斜，shuffle調(diao)優(you)(you)只(zhi)能在整個Spark的(de)(de)性(xing)能調(diao)優(you)(you)中占到(dao)一小部分(fen)而已。因此(ci)(ci)大(da)(da)家(jia)務(wu)必(bi)把握住調(diao)優(you)(you)的(de)(de)基本原則，千萬不(bu)要(yao)舍本逐末。下面我們就(jiu)給大(da)(da)家(jia)詳細講解(jie)shuffle的(de)(de)原理，以及相(xiang)關參(can)數(shu)的(de)(de)說明，同時給出各(ge)個參(can)數(shu)的(de)(de)調(diao)優(you)(you)建議。

ShuffleManager發展(zhan)概述

在Spark的源碼中(zhong)，負(fu)責shuffle過程的執行、計算和處理的組件主要(yao)就是(shi)ShuffleManager，也即shuffle管理器。而(er)隨(sui)著Spark的版本(ben)的發展，ShuffleManager也在不斷迭代，變得越來越先進。

在Spark 1.2以前(qian)，默認的(de)shuffle計算引擎是HashShuffleManager。該(gai)ShuffleManager而HashShuffleManager有著一個非常嚴重的(de)弊端(duan)，就是會產生大量的(de)中間磁(ci)盤文件，進而由大量的(de)磁(ci)盤IO操(cao)作影(ying)響(xiang)了性(xing)能。

因此在(zai)Spark 1.2以(yi)后的版(ban)本(ben)中(zhong)，默(mo)認的ShuffleManager改(gai)成(cheng)了SortShuffleManager。SortShuffleManager相較于(yu)HashShuffleManager來說，有了一定的改(gai)進(jin)。主要(yao)就(jiu)(jiu)在(zai)于(yu)，每(mei)個Task在(zai)進(jin)行shuffle操作時，雖然也會產生(sheng)較多(duo)的臨(lin)時磁盤文(wen)件(jian)(jian)，但是最(zui)后會將所有的臨(lin)時文(wen)件(jian)(jian)合并（merge）成(cheng)一個磁盤文(wen)件(jian)(jian)，因此每(mei)個Task就(jiu)(jiu)只(zhi)有一個磁盤文(wen)件(jian)(jian)。在(zai)下(xia)一個stage的shuffle read task拉(la)取自己的數據時，只(zhi)要(yao)根據索引讀取每(mei)個磁盤文(wen)件(jian)(jian)中(zhong)的部分數據即可。

下(xia)面我們(men)詳細分(fen)析一下(xia)HashShuffleManager和SortShuffleManager的原理。

HashShuffleManager運行原理

未經優化的HashShuffleManager

下圖說(shuo)明了未經優化的HashShuffleManager的原理。這里(li)我們先(xian)明確一個(ge)假(jia)設前提：每個(ge)Executor只有1個(ge)CPU core，也就是說(shuo)，無論這個(ge)Executor上(shang)分配多少(shao)個(ge)task線程，同一時間都只能執行一個(ge)task線程。

我們先從shuffle write開始說(shuo)起。shuffle write階(jie)段，主要就是(shi)在一個(ge)(ge)stage結束計算之后，為了下一個(ge)(ge)stage可以執行shuffle類(lei)的(de)算子（比如(ru)reduceByKey），而(er)(er)將(jiang)每(mei)個(ge)(ge)task處理(li)的(de)數(shu)據(ju)按key進行“分類(lei)”。所謂(wei)“分類(lei)”，就是(shi)對相(xiang)同的(de)key執行hash算法，從而(er)(er)將(jiang)相(xiang)同key都(dou)寫(xie)入同一個(ge)(ge)磁盤文件(jian)中(zhong)，而(er)(er)每(mei)一個(ge)(ge)磁盤文件(jian)都(dou)只屬于下游stage的(de)一個(ge)(ge)task。在將(jiang)數(shu)據(ju)寫(xie)入磁盤之前，會先將(jiang)數(shu)據(ju)寫(xie)入內(nei)存緩沖中(zhong)，當內(nei)存緩沖填滿之后，才會溢寫(xie)到磁盤文件(jian)中(zhong)去。

那么(me)每(mei)個(ge)(ge)(ge)執行(xing)shuffle write的(de)(de)task，要為(wei)下(xia)一個(ge)(ge)(ge)stage創(chuang)(chuang)建(jian)(jian)多少個(ge)(ge)(ge)磁盤文(wen)(wen)件呢(ni)？很簡單，下(xia)一個(ge)(ge)(ge)stage的(de)(de)task有多少個(ge)(ge)(ge)，當前stage的(de)(de)每(mei)個(ge)(ge)(ge)task就要創(chuang)(chuang)建(jian)(jian)多少份(fen)磁盤文(wen)(wen)件。比如下(xia)一個(ge)(ge)(ge)stage總(zong)(zong)共有100個(ge)(ge)(ge)task，那么(me)當前stage的(de)(de)每(mei)個(ge)(ge)(ge)task都要創(chuang)(chuang)建(jian)(jian)100份(fen)磁盤文(wen)(wen)件。如果當前stage有50個(ge)(ge)(ge)task，總(zong)(zong)共有10個(ge)(ge)(ge)Executor，每(mei)個(ge)(ge)(ge)Executor執行(xing)5個(ge)(ge)(ge)Task，那么(me)每(mei)個(ge)(ge)(ge)Executor上總(zong)(zong)共就要創(chuang)(chuang)建(jian)(jian)500個(ge)(ge)(ge)磁盤文(wen)(wen)件，所有Executor上會(hui)創(chuang)(chuang)建(jian)(jian)5000個(ge)(ge)(ge)磁盤文(wen)(wen)件。由此可(ke)見，未經優化(hua)的(de)(de)shuffle write操(cao)作所產生的(de)(de)磁盤文(wen)(wen)件的(de)(de)數量是極其(qi)驚人的(de)(de)。

接(jie)著我們來說說shuffle read。shuffle read，通常就(jiu)(jiu)是一個(ge)(ge)(ge)(ge)stage剛開始(shi)時要做的(de)(de)(de)事(shi)情。此(ci)時該stage的(de)(de)(de)每一個(ge)(ge)(ge)(ge)task就(jiu)(jiu)需要將(jiang)上(shang)一個(ge)(ge)(ge)(ge)stage的(de)(de)(de)計算結果中(zhong)的(de)(de)(de)所(suo)有相同key，從(cong)各(ge)個(ge)(ge)(ge)(ge)節(jie)(jie)點(dian)上(shang)通過(guo)(guo)網(wang)絡都拉取到自(zi)(zi)己所(suo)在(zai)的(de)(de)(de)節(jie)(jie)點(dian)上(shang)，然后進行key的(de)(de)(de)聚合或連(lian)接(jie)等操(cao)作(zuo)。由于shuffle write的(de)(de)(de)過(guo)(guo)程中(zhong)，task給下(xia)游(you)stage的(de)(de)(de)每個(ge)(ge)(ge)(ge)task都創建了一個(ge)(ge)(ge)(ge)磁(ci)盤文(wen)件(jian)(jian)，因(yin)此(ci)shuffle read的(de)(de)(de)過(guo)(guo)程中(zhong)，每個(ge)(ge)(ge)(ge)task只要從(cong)上(shang)游(you)stage的(de)(de)(de)所(suo)有task所(suo)在(zai)節(jie)(jie)點(dian)上(shang)，拉取屬(shu)于自(zi)(zi)己的(de)(de)(de)那(nei)一個(ge)(ge)(ge)(ge)磁(ci)盤文(wen)件(jian)(jian)即可。

shuffle read的(de)拉(la)取(qu)過(guo)程是(shi)一邊(bian)拉(la)取(qu)一邊(bian)進行聚(ju)合的(de)。每(mei)個shuffle read task都會有(you)一個自(zi)己的(de)buffer緩(huan)沖，每(mei)次都只能拉(la)取(qu)與buffer緩(huan)沖相同大小的(de)數(shu)(shu)據，然后(hou)(hou)通過(guo)內(nei)存(cun)中的(de)一個Map進行聚(ju)合等操作(zuo)。聚(ju)合完(wan)一批數(shu)(shu)據后(hou)(hou)，再拉(la)取(qu)下(xia)一批數(shu)(shu)據，并放(fang)到(dao)buffer緩(huan)沖中進行聚(ju)合操作(zuo)。以此類推，直到(dao)最(zui)后(hou)(hou)將(jiang)所有(you)數(shu)(shu)據到(dao)拉(la)取(qu)完(wan)，并得到(dao)最(zui)終的(de)結果。

優化后的HashShuffleManager

下圖說明了優(you)(you)化后的HashShuffleManager的原(yuan)理。這(zhe)里說的優(you)(you)化，是指我(wo)(wo)們可以設置(zhi)一個參數(shu)，spark.shuffle.consolidateFiles。該參數(shu)默認(ren)值為(wei)(wei)false，將其設置(zhi)為(wei)(wei)true即可開啟優(you)(you)化機制。通(tong)常來說，如果(guo)我(wo)(wo)們使(shi)用HashShuffleManager，那么都建(jian)議開啟這(zhe)個選項。

開啟consolidate機(ji)制之后，在(zai)shuffle write過程中，task就不(bu)是(shi)為下游stage的每(mei)個(ge)(ge)task創建一(yi)(yi)個(ge)(ge)磁盤文(wen)件(jian)了(le)。此時會(hui)(hui)出現shuffleFileGroup的概念，每(mei)個(ge)(ge)shuffleFileGroup會(hui)(hui)對應一(yi)(yi)批磁盤文(wen)件(jian)，磁盤文(wen)件(jian)的數(shu)量與下游stage的task數(shu)量是(shi)相同的。一(yi)(yi)個(ge)(ge)Executor上有多少(shao)個(ge)(ge)CPU core，就可以(yi)并行執(zhi)行多少(shao)個(ge)(ge)task。而第一(yi)(yi)批并行執(zhi)行的每(mei)個(ge)(ge)task都(dou)會(hui)(hui)創建一(yi)(yi)個(ge)(ge)shuffleFileGroup，并將數(shu)據寫入對應的磁盤文(wen)件(jian)內。

當(dang)Executor的(de)(de)(de)(de)CPU core執行完(wan)一批(pi)task，接著(zhu)執行下一批(pi)task時，下一批(pi)task就會復用之前已有(you)(you)的(de)(de)(de)(de)shuffleFileGroup，包(bao)括(kuo)其中(zhong)的(de)(de)(de)(de)磁(ci)盤(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)。也(ye)就是說，此(ci)時task會將(jiang)數據寫(xie)入(ru)已有(you)(you)的(de)(de)(de)(de)磁(ci)盤(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)中(zhong)，而不會寫(xie)入(ru)新的(de)(de)(de)(de)磁(ci)盤(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)中(zhong)。因此(ci)，consolidate機制(zhi)允許不同的(de)(de)(de)(de)task復用同一批(pi)磁(ci)盤(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)，這樣就可(ke)以有(you)(you)效將(jiang)多(duo)個task的(de)(de)(de)(de)磁(ci)盤(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)進(jin)行一定(ding)程度上的(de)(de)(de)(de)合并，從而大幅(fu)度減少(shao)磁(ci)盤(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)的(de)(de)(de)(de)數量，進(jin)而提升shuffle write的(de)(de)(de)(de)性能。

假設第(di)二個(ge)(ge)(ge)(ge)stage有100個(ge)(ge)(ge)(ge)task，第(di)一個(ge)(ge)(ge)(ge)stage有50個(ge)(ge)(ge)(ge)task，總共(gong)還是(shi)有10個(ge)(ge)(ge)(ge)Executor，每(mei)個(ge)(ge)(ge)(ge)Executor執行5個(ge)(ge)(ge)(ge)task。那么原本使用(yong)未經(jing)優化(hua)的(de)(de)HashShuffleManager時，每(mei)個(ge)(ge)(ge)(ge)Executor會(hui)產生500個(ge)(ge)(ge)(ge)磁(ci)(ci)(ci)盤文(wen)(wen)件，所有Executor會(hui)產生5000個(ge)(ge)(ge)(ge)磁(ci)(ci)(ci)盤文(wen)(wen)件的(de)(de)。但是(shi)此(ci)時經(jing)過(guo)優化(hua)之(zhi)后，每(mei)個(ge)(ge)(ge)(ge)Executor創(chuang)建(jian)的(de)(de)磁(ci)(ci)(ci)盤文(wen)(wen)件的(de)(de)數量的(de)(de)計算公式為：CPU core的(de)(de)數量 * 下一個(ge)(ge)(ge)(ge)stage的(de)(de)task數量。也就是(shi)說，每(mei)個(ge)(ge)(ge)(ge)Executor此(ci)時只會(hui)創(chuang)建(jian)100個(ge)(ge)(ge)(ge)磁(ci)(ci)(ci)盤文(wen)(wen)件，所有Executor只會(hui)創(chuang)建(jian)1000個(ge)(ge)(ge)(ge)磁(ci)(ci)(ci)盤文(wen)(wen)件。

SortShuffleManager運行原理

SortShuffleManager的運(yun)(yun)行(xing)機(ji)制(zhi)主(zhu)要分成兩種，一種是普通運(yun)(yun)行(xing)機(ji)制(zhi)，另一種是bypass運(yun)(yun)行(xing)機(ji)制(zhi)。當shuffle read task的數(shu)量小(xiao)于等(deng)于spark.shuffle.sort.bypassMergeThreshold參數(shu)的值時(shi)（默認為(wei)200），就會啟用bypass機(ji)制(zhi)。

普通運行機(ji)制

下圖說明(ming)了(le)普(pu)通(tong)的(de)(de)SortShuffleManager的(de)(de)原(yuan)理。在該模(mo)式下，數(shu)(shu)(shu)據(ju)(ju)(ju)會(hui)(hui)先寫(xie)入(ru)一(yi)個內(nei)存(cun)數(shu)(shu)(shu)據(ju)(ju)(ju)結(jie)構(gou)中，此時(shi)根據(ju)(ju)(ju)不(bu)同的(de)(de)shuffle算子(zi)，可(ke)能選(xuan)(xuan)用(yong)不(bu)同的(de)(de)數(shu)(shu)(shu)據(ju)(ju)(ju)結(jie)構(gou)。如(ru)果是(shi)reduceByKey這種聚(ju)合(he)類(lei)的(de)(de)shuffle算子(zi)，那(nei)么(me)會(hui)(hui)選(xuan)(xuan)用(yong)Map數(shu)(shu)(shu)據(ju)(ju)(ju)結(jie)構(gou)，一(yi)邊通(tong)過(guo)Map進行聚(ju)合(he)，一(yi)邊寫(xie)入(ru)內(nei)存(cun)；如(ru)果是(shi)join這種普(pu)通(tong)的(de)(de)shuffle算子(zi)，那(nei)么(me)會(hui)(hui)選(xuan)(xuan)用(yong)Array數(shu)(shu)(shu)據(ju)(ju)(ju)結(jie)構(gou)，直接寫(xie)入(ru)內(nei)存(cun)。接著，每寫(xie)一(yi)條數(shu)(shu)(shu)據(ju)(ju)(ju)進入(ru)內(nei)存(cun)數(shu)(shu)(shu)據(ju)(ju)(ju)結(jie)構(gou)之后，就會(hui)(hui)判斷一(yi)下，是(shi)否達到(dao)了(le)某(mou)個臨界閾(yu)值(zhi)。如(ru)果達到(dao)臨界閾(yu)值(zhi)的(de)(de)話，那(nei)么(me)就會(hui)(hui)嘗試將內(nei)存(cun)數(shu)(shu)(shu)據(ju)(ju)(ju)結(jie)構(gou)中的(de)(de)數(shu)(shu)(shu)據(ju)(ju)(ju)溢寫(xie)到(dao)磁盤，然后清空內(nei)存(cun)數(shu)(shu)(shu)據(ju)(ju)(ju)結(jie)構(gou)。

在溢寫(xie)到磁(ci)盤(pan)(pan)文(wen)(wen)件之(zhi)前，會先(xian)根據key對內存(cun)數據結構中(zhong)已有的(de)數據進行排序(xu)。排序(xu)過后，會分(fen)批(pi)將數據寫(xie)入磁(ci)盤(pan)(pan)文(wen)(wen)件。默認的(de)batch數量是(shi)10000條，也就是(shi)說(shuo)，排序(xu)好(hao)的(de)數據，會以(yi)每批(pi)1萬條數據的(de)形式分(fen)批(pi)寫(xie)入磁(ci)盤(pan)(pan)文(wen)(wen)件。寫(xie)入磁(ci)盤(pan)(pan)文(wen)(wen)件是(shi)通過Java的(de)BufferedOutputStream實(shi)現的(de)。BufferedOutputStream是(shi)Java的(de)緩沖輸出流，首(shou)先(xian)會將數據緩沖在內存(cun)中(zhong)，當內存(cun)緩沖滿(man)溢之(zhi)后再一次(ci)寫(xie)入磁(ci)盤(pan)(pan)文(wen)(wen)件中(zhong)，這樣(yang)可以(yi)減(jian)少磁(ci)盤(pan)(pan)IO次(ci)數，提升(sheng)性能。

一(yi)個task將所(suo)(suo)有(you)數(shu)據(ju)寫(xie)入內存(cun)數(shu)據(ju)結(jie)構的(de)過(guo)程(cheng)中(zhong)(zhong)，會發生(sheng)多(duo)次磁盤(pan)(pan)(pan)溢寫(xie)操作，也(ye)就(jiu)會產(chan)生(sheng)多(duo)個臨時文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)(jian)。最后會將之前所(suo)(suo)有(you)的(de)臨時磁盤(pan)(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)(jian)都進行合并，這就(jiu)是merge過(guo)程(cheng)，此時會將之前所(suo)(suo)有(you)臨時磁盤(pan)(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)(jian)中(zhong)(zhong)的(de)數(shu)據(ju)讀取出來，然后依(yi)次寫(xie)入最終(zhong)的(de)磁盤(pan)(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)(jian)之中(zhong)(zhong)。此外，由于一(yi)個task就(jiu)只(zhi)對(dui)應一(yi)個磁盤(pan)(pan)(pan)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)(jian)，也(ye)就(jiu)意味(wei)著該task為下游stage的(de)task準(zhun)備的(de)數(shu)據(ju)都在(zai)這一(yi)個文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)(jian)中(zhong)(zhong)，因此還會單獨寫(xie)一(yi)份索引文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)(jian)，其中(zhong)(zhong)標識了下游各個task的(de)數(shu)據(ju)在(zai)文(wen)(wen)件(jian)(jian)(jian)(jian)(jian)(jian)中(zhong)(zhong)的(de)start offset與(yu)end offset。

SortShuffleManager由于(yu)有一(yi)個(ge)(ge)磁(ci)盤文(wen)件(jian)merge的(de)過程，因此大大減(jian)少了(le)文(wen)件(jian)數(shu)量。比(bi)如第(di)一(yi)個(ge)(ge)stage有50個(ge)(ge)task，總共(gong)有10個(ge)(ge)Executor，每(mei)(mei)個(ge)(ge)Executor執行5個(ge)(ge)task，而第(di)二個(ge)(ge)stage有100個(ge)(ge)task。由于(yu)每(mei)(mei)個(ge)(ge)task最終只有一(yi)個(ge)(ge)磁(ci)盤文(wen)件(jian)，因此此時(shi)每(mei)(mei)個(ge)(ge)Executor上(shang)只有5個(ge)(ge)磁(ci)盤文(wen)件(jian)，所有Executor只有50個(ge)(ge)磁(ci)盤文(wen)件(jian)。

bypass運(yun)行機制

下圖(tu)說(shuo)明了bypass SortShuffleManager的原理。bypass運行機制的觸發條(tiao)件(jian)如下：

• shuffle map task數(shu)量(liang)小(xiao)于spark.shuffle.sort.bypassMergeThreshold參數(shu)的(de)值。
• 不是聚合類的shuffle算(suan)子（比如reduceByKey）。

此(ci)時(shi)(shi)task會為每個(ge)下游task都(dou)創建一(yi)個(ge)臨(lin)時(shi)(shi)磁盤文(wen)(wen)(wen)件(jian)，并(bing)(bing)將(jiang)數(shu)據(ju)按key進行hash然(ran)后(hou)根(gen)據(ju)key的hash值(zhi)，將(jiang)key寫(xie)(xie)入(ru)對應(ying)的磁盤文(wen)(wen)(wen)件(jian)之(zhi)中。當然(ran)，寫(xie)(xie)入(ru)磁盤文(wen)(wen)(wen)件(jian)時(shi)(shi)也是(shi)先寫(xie)(xie)入(ru)內存(cun)緩(huan)(huan)沖(chong)，緩(huan)(huan)沖(chong)寫(xie)(xie)滿之(zhi)后(hou)再(zai)溢(yi)寫(xie)(xie)到磁盤文(wen)(wen)(wen)件(jian)的。最(zui)后(hou)，同樣會將(jiang)所(suo)有臨(lin)時(shi)(shi)磁盤文(wen)(wen)(wen)件(jian)都(dou)合并(bing)(bing)成一(yi)個(ge)磁盤文(wen)(wen)(wen)件(jian)，并(bing)(bing)創建一(yi)個(ge)單獨的索(suo)引文(wen)(wen)(wen)件(jian)。

該過程的(de)(de)磁(ci)盤(pan)寫(xie)機(ji)(ji)制(zhi)其實跟未(wei)經(jing)優化(hua)的(de)(de)HashShuffleManager是(shi)一模(mo)一樣的(de)(de)，因(yin)為都(dou)要創建數量驚人的(de)(de)磁(ci)盤(pan)文(wen)件，只是(shi)在最(zui)后會(hui)做一個(ge)磁(ci)盤(pan)文(wen)件的(de)(de)合(he)并而已。因(yin)此少量的(de)(de)最(zui)終磁(ci)盤(pan)文(wen)件，也讓該機(ji)(ji)制(zhi)相(xiang)對未(wei)經(jing)優化(hua)的(de)(de)HashShuffleManager來說，shuffle read的(de)(de)性能會(hui)更好。

而該(gai)機制(zhi)(zhi)(zhi)與普通SortShuffleManager運行機制(zhi)(zhi)(zhi)的(de)不(bu)同(tong)在于：第一，磁盤寫機制(zhi)(zhi)(zhi)不(bu)同(tong)；第二，不(bu)會進行排(pai)序。也(ye)就是(shi)說，啟用該(gai)機制(zhi)(zhi)(zhi)的(de)最大好處在于，shuffle write過程(cheng)中(zhong)，不(bu)需要進行數(shu)據的(de)排(pai)序操作(zuo)，也(ye)就節省掉了這部(bu)分的(de)性能開銷(xiao)。

shuffle相關參數調優

以下是Shffule過程中的(de)(de)一些主要參數(shu)，這里詳細講解了(le)各(ge)個參數(shu)的(de)(de)功(gong)能(neng)、默認值以及基(ji)于(yu)實踐經驗給出(chu)的(de)(de)調優建議。

spark.shuffle.file.buffer

• 默認(ren)值：32k
• 參(can)數說(shuo)明：該參(can)數用于設置shuffle write task的BufferedOutputStream的buffer緩沖(chong)大小。將數據寫(xie)到磁(ci)(ci)盤(pan)(pan)文(wen)件(jian)之前，會(hui)先寫(xie)入(ru)buffer緩沖(chong)中(zhong)，待(dai)緩沖(chong)寫(xie)滿之后，才會(hui)溢(yi)寫(xie)到磁(ci)(ci)盤(pan)(pan)。
• 調優建議：如果作業可(ke)(ke)(ke)用的(de)(de)內存資源較為充足的(de)(de)話(hua)，可(ke)(ke)(ke)以(yi)適(shi)當(dang)增加這個(ge)參數(shu)的(de)(de)大小(xiao)（比如64k），從而減(jian)少shuffle write過(guo)程中溢寫磁盤文件的(de)(de)次數(shu)，也就(jiu)可(ke)(ke)(ke)以(yi)減(jian)少磁盤IO次數(shu)，進而提升(sheng)性能。在實踐中發現，合理調節該(gai)參數(shu)，性能會(hui)有1%~5%的(de)(de)提升(sheng)。

spark.reducer.maxSizeInFlight

• 默認值：48m
• 參(can)數說明：該(gai)參(can)數用于設置shuffle read task的buffer緩(huan)沖大小，而這個(ge)buffer緩(huan)沖決定了每次能夠(gou)拉取多少(shao)數據。
• 調優建議：如(ru)(ru)果作業可用的(de)(de)內存資源較為充(chong)足的(de)(de)話，可以適(shi)當增(zeng)加(jia)這(zhe)個參數(shu)(shu)的(de)(de)大小（比(bi)如(ru)(ru)96m），從而減(jian)少(shao)拉取(qu)數(shu)(shu)據的(de)(de)次數(shu)(shu)，也就(jiu)可以減(jian)少(shao)網(wang)絡傳輸的(de)(de)次數(shu)(shu)，進(jin)而提升性能。在實踐(jian)中發現，合理調節(jie)該(gai)參數(shu)(shu)，性能會有1%~5%的(de)(de)提升。

spark.shuffle.io.maxRetries

• 默認值：3
• 參(can)數(shu)說明：shuffle read task從(cong)shuffle write task所在(zai)節點拉取(qu)屬于自己的數(shu)據時，如(ru)果(guo)因為網絡(luo)異常(chang)導(dao)致拉取(qu)失(shi)敗，是會自動(dong)進行(xing)(xing)重(zhong)試(shi)的。該(gai)參(can)數(shu)就(jiu)代表(biao)了(le)可(ke)以重(zhong)試(shi)的最(zui)大次數(shu)。如(ru)果(guo)在(zai)指定次數(shu)之(zhi)內拉取(qu)還是沒有成(cheng)功，就(jiu)可(ke)能會導(dao)致作業(ye)執(zhi)行(xing)(xing)失(shi)敗。
• 調(diao)優建議：對于(yu)那些包含了(le)特別耗時的(de)shuffle操作的(de)作業，建議增加重試最大次數（比如60次），以(yi)避免由(you)于(yu)JVM的(de)full gc或者網絡不(bu)穩定等因素導致的(de)數據拉取失敗(bai)。在實踐中發現(xian)，對于(yu)針對超大數據量（數十億~上百億）的(de)shuffle過程，調(diao)節該參數可以(yi)大幅度提升穩定性(xing)。

spark.shuffle.io.retryWait

• 默認值(zhi)：5s
• 參數說明：具(ju)體解(jie)釋同上，該參數代(dai)表了每次重試拉取數據的等(deng)待間隔，默認是5s。
• 調優建議：建議加大間隔(ge)時(shi)長(chang)（比如60s），以(yi)增(zeng)加shuffle操(cao)作的穩(wen)定性。

spark.shuffle.memoryFraction

• 默(mo)認值：0.2
• 參(can)數說明：該參(can)數代表(biao)了Executor內存中，分配(pei)給shuffle read task進行聚合操作的內存比例(li)，默認(ren)是20%。
• 調(diao)(diao)(diao)優建議(yi)：在資源(yuan)參數調(diao)(diao)(diao)優中(zhong)講解過(guo)(guo)這(zhe)個參數。如果(guo)內(nei)存(cun)充足，而且很少使用持久化操作，建議(yi)調(diao)(diao)(diao)高這(zhe)個比例，給(gei)shuffle read的聚合操作更(geng)多內(nei)存(cun)，以避免(mian)由于內(nei)存(cun)不足導致聚合過(guo)(guo)程中(zhong)頻繁讀寫(xie)磁盤。在實踐(jian)中(zhong)發現，合理調(diao)(diao)(diao)節該參數可以將性能提升10%左右。

spark.shuffle.manager

• 默認(ren)值：sort
• 參數(shu)說明(ming)：該參數(shu)用(yong)(yong)于設置ShuffleManager的類(lei)型。Spark 1.5以(yi)后，有三個可選項：hash、sort和(he)tungsten-sort。HashShuffleManager是(shi)Spark 1.2以(yi)前的默認選項，但是(shi)Spark 1.2以(yi)及(ji)之后的版本默認都是(shi)SortShuffleManager了。tungsten-sort與sort類(lei)似，但是(shi)使(shi)用(yong)(yong)了tungsten計劃中的堆(dui)外內存(cun)(cun)管(guan)理機(ji)制，內存(cun)(cun)使(shi)用(yong)(yong)效率更高。
• 調(diao)優建議：由(you)于SortShuffleManager默認會對(dui)數(shu)據進(jin)行排序，因此(ci)如(ru)果你(ni)的(de)業務邏(luo)輯(ji)中需要該排序機制(zhi)的(de)話，則(ze)使(shi)用(yong)默認的(de)SortShuffleManager就可以；而如(ru)果你(ni)的(de)業務邏(luo)輯(ji)不需要對(dui)數(shu)據進(jin)行排序，那么建議參(can)考后面的(de)幾個參(can)數(shu)調(diao)優，通過bypass機制(zhi)或優化的(de)HashShuffleManager來(lai)避免(mian)排序操(cao)作，同時提供較好的(de)磁(ci)盤讀寫性能。這里要注意的(de)是，tungsten-sort要慎(shen)用(yong)，因為之(zhi)前發現了一些相應的(de)bug。

spark.shuffle.sort.bypassMergeThreshold

• 默(mo)認值：200
• 參數說明(ming)：當ShuffleManager為SortShuffleManager時(shi)，如果shuffle read task的數量小于(yu)這個(ge)閾值(zhi)（默認是200），則shuffle write過(guo)程中不(bu)會(hui)進(jin)行排序操作，而(er)是直(zhi)接按照未(wei)經優化的HashShuffleManager的方式(shi)去寫數據，但(dan)是最后會(hui)將每個(ge)task產(chan)生的所有臨時(shi)磁盤文(wen)件(jian)都合(he)并成一個(ge)文(wen)件(jian)，并會(hui)創建單獨的索引文(wen)件(jian)。
• 調優(you)建(jian)議：當你使用SortShuffleManager時，如果的確(que)不需要排序(xu)操作，那么建(jian)議將這個(ge)參數調大(da)一些，大(da)于shuffle read task的數量。那么此(ci)(ci)時就會(hui)自動啟用bypass機(ji)制，map-side就不會(hui)進行排序(xu)了(le)，減少(shao)了(le)排序(xu)的性能(neng)開銷。但是這種方式下(xia)，依然會(hui)產生(sheng)大(da)量的磁盤(pan)文件(jian)，因此(ci)(ci)shuffle write性能(neng)有待提高(gao)。

spark.shuffle.consolidateFiles

• 默認(ren)值(zhi)：false
• 參(can)(can)數說(shuo)明：如果(guo)使用HashShuffleManager，該參(can)(can)數有效(xiao)。如果(guo)設(she)置為true，那么就會(hui)開(kai)(kai)啟consolidate機制(zhi)，會(hui)大幅度(du)合并shuffle write的輸出文件，對于(yu)shuffle read task數量特別多(duo)的情況下，這(zhe)種方法可以(yi)極(ji)大地減(jian)少磁(ci)盤IO開(kai)(kai)銷，提升性(xing)能。
• 調優建議(yi)：如果的確(que)不需要SortShuffleManager的排序(xu)機制，那么除了(le)使用bypass機制，還可(ke)以(yi)嘗試(shi)將spark.shffle.manager參數手動指定為hash，使用HashShuffleManager，同(tong)時開(kai)啟consolidate機制。在(zai)實踐中嘗試(shi)過，發現其性能比開(kai)啟了(le)bypass機制的SortShuffleManager要高出10%~30%。

寫在最后的話

本文(wen)分(fen)別(bie)講解(jie)了開發(fa)(fa)過程中的(de)(de)(de)(de)優(you)(you)化原則、運行前的(de)(de)(de)(de)資源(yuan)參數設置調(diao)優(you)(you)、運行中的(de)(de)(de)(de)數據傾斜的(de)(de)(de)(de)解(jie)決方案、為了精益求精的(de)(de)(de)(de)shuffle調(diao)優(you)(you)。希望(wang)大家能(neng)夠在(zai)閱讀本文(wen)之后，記(ji)住這些性能(neng)調(diao)優(you)(you)的(de)(de)(de)(de)原則以及方案，在(zai)Spark作(zuo)業開發(fa)(fa)、測試以及運行的(de)(de)(de)(de)過程中多嘗試，只有(you)這樣，我們才能(neng)開發(fa)(fa)出更優(you)(you)的(de)(de)(de)(de)Spark作(zuo)業，不斷提(ti)升其性能(neng)。