1、前言

在大數(shù)據(jù)計(jì)算領(lǐng)域，Spark已經(jīng)成為了越來(lái)越流行、越來(lái)越受歡迎的計(jì)算平臺(tái)之一。Spark的功能涵蓋了大數(shù)據(jù)領(lǐng)域的離線批處理、SQL類(lèi)處理、流式/實(shí)時(shí)計(jì)算、機(jī)器學(xué)習(xí)、圖計(jì)算等各種不同類(lèi)型的計(jì)算操作，應(yīng)用范圍與前景非常廣泛。在美團(tuán)•大眾點(diǎn)評(píng)，已經(jīng)有很多同學(xué)在各種項(xiàng)目中嘗試使用Spark。大多數(shù)同學(xué)(包括筆者在內(nèi))，最初開(kāi)始嘗試使用Spark的原因很簡(jiǎn)單，主要就是為了讓大數(shù)據(jù)計(jì)算作業(yè)的執(zhí)行速度更快、性能更高。

[[236996]]

然而，通過(guò)Spark開(kāi)發(fā)出高性能的大數(shù)據(jù)計(jì)算作業(yè)，并不是那么簡(jiǎn)單的。如果沒(méi)有對(duì)Spark作業(yè)進(jìn)行合理的調(diào)優(yōu)，Spark作業(yè)的執(zhí)行速度可能會(huì)很慢，這樣就完全體現(xiàn)不出Spark作為一種快速大數(shù)據(jù)計(jì)算引擎的優(yōu)勢(shì)來(lái)。因此，想要用好Spark，就必須對(duì)其進(jìn)行合理的性能優(yōu)化。

Spark的性能調(diào)優(yōu)實(shí)際上是由很多部分組成的，不是調(diào)節(jié)幾個(gè)參數(shù)就可以立竿見(jiàn)影提升作業(yè)性能的。我們需要根據(jù)不同的業(yè)務(wù)場(chǎng)景以及數(shù)據(jù)情況，對(duì)Spark作業(yè)進(jìn)行綜合性的分析，然后進(jìn)行多個(gè)方面的調(diào)節(jié)和優(yōu)化，才能獲得***性能。

筆者根據(jù)之前的Spark作業(yè)開(kāi)發(fā)經(jīng)驗(yàn)以及實(shí)踐積累，總結(jié)出了一套Spark作業(yè)的性能優(yōu)化方案。整套方案主要分為開(kāi)發(fā)調(diào)優(yōu)、資源調(diào)優(yōu)、數(shù)據(jù)傾斜調(diào)優(yōu)、shuffle調(diào)優(yōu)幾個(gè)部分。開(kāi)發(fā)調(diào)優(yōu)和資源調(diào)優(yōu)是所有Spark作業(yè)都需要注意和遵循的一些基本原則，是高性能Spark作業(yè)的基礎(chǔ);數(shù)據(jù)傾斜調(diào)優(yōu)，主要講解了一套完整的用來(lái)解決Spark作業(yè)數(shù)據(jù)傾斜的解決方案;shuffle調(diào)優(yōu)，面向的是對(duì)Spark的原理有較深層次掌握和研究的同學(xué)，主要講解了如何對(duì)Spark作業(yè)的shuffle運(yùn)行過(guò)程以及細(xì)節(jié)進(jìn)行調(diào)優(yōu)。

本文作為Spark性能優(yōu)化指南的基礎(chǔ)篇，主要講解開(kāi)發(fā)調(diào)優(yōu)以及資源調(diào)優(yōu)。

2、開(kāi)發(fā)調(diào)優(yōu)

Spark性能優(yōu)化的***步，就是要在開(kāi)發(fā)Spark作業(yè)的過(guò)程中注意和應(yīng)用一些性能優(yōu)化的基本原則。開(kāi)發(fā)調(diào)優(yōu)，就是要讓大家了解以下一些Spark基本開(kāi)發(fā)原則，包括：RDD lineage設(shè)計(jì)、算子的合理使用、特殊操作的優(yōu)化等。在開(kāi)發(fā)過(guò)程中，時(shí)時(shí)刻刻都應(yīng)該注意以上原則，并將這些原則根據(jù)具體的業(yè)務(wù)以及實(shí)際的應(yīng)用場(chǎng)景，靈活地運(yùn)用到自己的Spark作業(yè)中。Spark性能優(yōu)化的***步，就是要在開(kāi)發(fā)Spark作業(yè)的過(guò)程中注意和應(yīng)用一些性能優(yōu)化的基本原則。開(kāi)發(fā)調(diào)優(yōu)，就是要讓大家了解以下一些Spark基本開(kāi)發(fā)原則，包括：RDD lineage設(shè)計(jì)、算子的合理使用、特殊操作的優(yōu)化等。在開(kāi)發(fā)過(guò)程中，時(shí)時(shí)刻刻都應(yīng)該注意以上原則，并將這些原則根據(jù)具體的業(yè)務(wù)以及實(shí)際的應(yīng)用場(chǎng)景，靈活地運(yùn)用到自己的Spark作業(yè)中。

原則一：避免創(chuàng)建重復(fù)的RDD

通常來(lái)說(shuō)，我們?cè)陂_(kāi)發(fā)一個(gè)Spark作業(yè)時(shí)，首先是基于某個(gè)數(shù)據(jù)源(比如Hive表或HDFS文件)創(chuàng)建一個(gè)初始的RDD;接著對(duì)這個(gè)RDD執(zhí)行某個(gè)算子操作，然后得到下一個(gè)RDD;以此類(lèi)推，循環(huán)往復(fù)，直到計(jì)算出最終我們需要的結(jié)果。在這個(gè)過(guò)程中，多個(gè)RDD會(huì)通過(guò)不同的算子操作(比如map、reduce等)串起來(lái)，這個(gè)“RDD串”，就是RDD lineage，也就是“RDD的血緣關(guān)系鏈”。

我們?cè)陂_(kāi)發(fā)過(guò)程中要注意：對(duì)于同一份數(shù)據(jù)，只應(yīng)該創(chuàng)建一個(gè)RDD，不能創(chuàng)建多個(gè)RDD來(lái)代表同一份數(shù)據(jù)。

一些Spark初學(xué)者在剛開(kāi)始開(kāi)發(fā)Spark作業(yè)時(shí)，或者是有經(jīng)驗(yàn)的工程師在開(kāi)發(fā)RDD lineage極其冗長(zhǎng)的Spark作業(yè)時(shí)，可能會(huì)忘了自己之前對(duì)于某一份數(shù)據(jù)已經(jīng)創(chuàng)建過(guò)一個(gè)RDD了，從而導(dǎo)致對(duì)于同一份數(shù)據(jù)，創(chuàng)建了多個(gè)RDD。這就意味著，我們的Spark作業(yè)會(huì)進(jìn)行多次重復(fù)計(jì)算來(lái)創(chuàng)建多個(gè)代表相同數(shù)據(jù)的RDD，進(jìn)而增加了作業(yè)的性能開(kāi)銷(xiāo)。

一個(gè)簡(jiǎn)單的例子 

//也就是說(shuō)，需要對(duì)一份數(shù)據(jù)執(zhí)行兩次算子操作。 
//錯(cuò)誤的做法：對(duì)于同一份數(shù)據(jù)執(zhí)行多次算子操作時(shí)，創(chuàng)建多個(gè)RDD。 
//這里執(zhí)行了兩次textFile方法，針對(duì)同一個(gè)HDFS文件，創(chuàng)建了兩個(gè)RDD出來(lái)， 
//然后分別對(duì)每個(gè)RDD都執(zhí)行了一個(gè)算子操作。 
//這種情況下，Spark需要從HDFS上兩次加載hello.txt文件的內(nèi)容，并創(chuàng)建兩個(gè)單獨(dú)的RDD； 
//第二次加載HDFS文件以及創(chuàng)建RDD的性能開(kāi)銷(xiāo)，很明顯是白白浪費(fèi)掉的。 
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt") 
rdd1.map(...) 
val rdd2 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt") 
rdd2.reduce(...) 
//正確的用法：對(duì)于一份數(shù)據(jù)執(zhí)行多次算子操作時(shí)，只使用一個(gè)RDD。 
//這種寫(xiě)法很明顯比上一種寫(xiě)法要好多了，因?yàn)槲覀儗?duì)于同一份數(shù)據(jù)只創(chuàng)建了一個(gè)RDD， 
//然后對(duì)這一個(gè)RDD執(zhí)行了多次算子操作。 
//但是要注意到這里為止優(yōu)化還沒(méi)有結(jié)束，由于rdd1被執(zhí)行了兩次算子操作，第二次執(zhí)行reduce操作的時(shí)候， 
//還會(huì)再次從源頭處重新計(jì)算一次rdd1的數(shù)據(jù)，因此還是會(huì)有重復(fù)計(jì)算的性能開(kāi)銷(xiāo)。 
//要徹底解決這個(gè)問(wèn)題，必須結(jié)合“原則三：對(duì)多次使用的RDD進(jìn)行持久化”， 
//才能保證一個(gè)RDD被多次使用時(shí)只被計(jì)算一次。 
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt") 
rdd1.map(...) 

原則二：盡可能復(fù)用同一個(gè)RDD

除了要避免在開(kāi)發(fā)過(guò)程中對(duì)一份完全相同的數(shù)據(jù)創(chuàng)建多個(gè)RDD之外，在對(duì)不同的數(shù)據(jù)執(zhí)行算子操作時(shí)還要盡可能地復(fù)用一個(gè)RDD。比如說(shuō)，有一個(gè)RDD的數(shù)據(jù)格式是key-value類(lèi)型的，另一個(gè)是單value類(lèi)型的，這兩個(gè)RDD的value數(shù)據(jù)是完全一樣的。那么此時(shí)我們可以只使用key-value類(lèi)型的那個(gè)RDD，因?yàn)槠渲幸呀?jīng)包含了另一個(gè)的數(shù)據(jù)。對(duì)于類(lèi)似這種多個(gè)RDD的數(shù)據(jù)有重疊或者包含的情況，我們應(yīng)該盡量復(fù)用一個(gè)RDD，這樣可以盡可能地減少RDD的數(shù)量，從而盡可能減少算子執(zhí)行的次數(shù)。

一個(gè)簡(jiǎn)單的例子 

// 錯(cuò)誤的做法。 
  
// 有一個(gè)<Long, String>格式的RDD，即rdd1。 
// 接著由于業(yè)務(wù)需要，對(duì)rdd1執(zhí)行了一個(gè)map操作，創(chuàng)建了一個(gè)rdd2，而rdd2中的數(shù)據(jù)僅僅是rdd1中的value值而已，也就是說(shuō)，rdd2是rdd1的子集。 
JavaPairRDD<Long, String> rdd1 = ... 
JavaRDD<String> rdd2 = rdd1.map(...) 
  
// 分別對(duì)rdd1和rdd2執(zhí)行了不同的算子操作。 
rdd1.reduceByKey(...) 
rdd2.map(...) 
  
// 正確的做法。 
  
// 上面這個(gè)case中，其實(shí)rdd1和rdd2的區(qū)別無(wú)非就是數(shù)據(jù)格式不同而已，rdd2的數(shù)據(jù)完全就是rdd1的子集而已，卻創(chuàng)建了兩個(gè)rdd，并對(duì)兩個(gè)rdd都執(zhí)行了一次算子操作。 
// 此時(shí)會(huì)因?yàn)閷?duì)rdd1執(zhí)行map算子來(lái)創(chuàng)建rdd2，而多執(zhí)行一次算子操作，進(jìn)而增加性能開(kāi)銷(xiāo)。 
// 其實(shí)在這種情況下完全可以復(fù)用同一個(gè)RDD。 
// 我們可以使用rdd1，既做reduceByKey操作，也做map操作。 
// 在進(jìn)行第二個(gè)map操作時(shí)，只使用每個(gè)數(shù)據(jù)的tuple._2，也就是rdd1中的value值，即可。 
JavaPairRDD<Long, String> rdd1 = ... 
rdd1.reduceByKey(...) 
rdd1.map(tuple._2...) 
  
// 第二種方式相較于***種方式而言，很明顯減少了一次rdd2的計(jì)算開(kāi)銷(xiāo)。 
// 但是到這里為止，優(yōu)化還沒(méi)有結(jié)束，對(duì)rdd1我們還是執(zhí)行了兩次算子操作，rdd1實(shí)際上還是會(huì)被計(jì)算兩次。 
// 因此還需要配合“原則三：對(duì)多次使用的RDD進(jìn)行持久化”進(jìn)行使用，才能保證一個(gè)RDD被多次使用時(shí)只被計(jì)算一次。 

原則三：對(duì)多次使用的RDD進(jìn)行持久化

當(dāng)你在Spark代碼中多次對(duì)一個(gè)RDD做了算子操作后，恭喜，你已經(jīng)實(shí)現(xiàn)Spark作業(yè)***步的優(yōu)化了，也就是盡可能復(fù)用RDD。此時(shí)就該在這個(gè)基礎(chǔ)之上，進(jìn)行第二步優(yōu)化了，也就是要保證對(duì)一個(gè)RDD執(zhí)行多次算子操作時(shí)，這個(gè)RDD本身僅僅被計(jì)算一次。

Spark中對(duì)于一個(gè)RDD執(zhí)行多次算子的默認(rèn)原理是這樣的：每次你對(duì)一個(gè)RDD執(zhí)行一個(gè)算子操作時(shí)，都會(huì)重新從源頭處計(jì)算一遍，計(jì)算出那個(gè)RDD來(lái)，然后再對(duì)這個(gè)RDD執(zhí)行你的算子操作。這種方式的性能是很差的。

因此對(duì)于這種情況，我們的建議是：對(duì)多次使用的RDD進(jìn)行持久化。此時(shí)Spark就會(huì)根據(jù)你的持久化策略，將RDD中的數(shù)據(jù)保存到內(nèi)存或者磁盤(pán)中。以后每次對(duì)這個(gè)RDD進(jìn)行算子操作時(shí)，都會(huì)直接從內(nèi)存或磁盤(pán)中提取持久化的RDD數(shù)據(jù)，然后執(zhí)行算子，而不會(huì)從源頭處重新計(jì)算一遍這個(gè)RDD，再執(zhí)行算子操作。

對(duì)多次使用的RDD進(jìn)行持久化的代碼示例 

// 如果要對(duì)一個(gè)RDD進(jìn)行持久化，只要對(duì)這個(gè)RDD調(diào)用cache()和persist()即可。 
  
// 正確的做法。 
// cache()方法表示：使用非序列化的方式將RDD中的數(shù)據(jù)全部嘗試持久化到內(nèi)存中。 
// 此時(shí)再對(duì)rdd1執(zhí)行兩次算子操作時(shí)，只有在***次執(zhí)行map算子時(shí)，才會(huì)將這個(gè)rdd1從源頭處計(jì)算一次。 
// 第二次執(zhí)行reduce算子時(shí)，就會(huì)直接從內(nèi)存中提取數(shù)據(jù)進(jìn)行計(jì)算，不會(huì)重復(fù)計(jì)算一個(gè)rdd。 
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt").cache() 
rdd1.map(...) 
rdd1.reduce(...) 
  
// persist()方法表示：手動(dòng)選擇持久化級(jí)別，并使用指定的方式進(jìn)行持久化。 
// 比如說(shuō)，StorageLevel.MEMORY_AND_DISK_SER表示，內(nèi)存充足時(shí)優(yōu)先持久化到內(nèi)存中，內(nèi)存不充足時(shí)持久化到磁盤(pán)文件中。 
// 而且其中的_SER后綴表示，使用序列化的方式來(lái)保存RDD數(shù)據(jù)，此時(shí)RDD中的每個(gè)partition都會(huì)序列化成一個(gè)大的字節(jié)數(shù)組，然后再持久化到內(nèi)存或磁盤(pán)中。 
// 序列化的方式可以減少持久化的數(shù)據(jù)對(duì)內(nèi)存/磁盤(pán)的占用量，進(jìn)而避免內(nèi)存被持久化數(shù)據(jù)占用過(guò)多，從而發(fā)生頻繁GC。 
val rdd1 = sc.textFile("hdfs://192.168.0.1:9000/hello.txt").persist(StorageLevel.MEMORY_AND_DISK_SER) 
rdd1.map(...) 
rdd1.reduce(...) 

對(duì)于persist()方法而言，我們可以根據(jù)不同的業(yè)務(wù)場(chǎng)景選擇不同的持久化級(jí)別。

Spark的持久化級(jí)別

如何選擇一種最合適的持久化策略

• 默認(rèn)情況下，性能***的當(dāng)然是MEMORY_ONLY，但前提是你的內(nèi)存必須足夠足夠大，可以綽綽有余地存放下整個(gè)RDD的所有數(shù)據(jù)。因?yàn)椴贿M(jìn)行序列化與反序列化操作，就避免了這部分的性能開(kāi)銷(xiāo);對(duì)這個(gè)RDD的后續(xù)算子操作，都是基于純內(nèi)存中的數(shù)據(jù)的操作，不需要從磁盤(pán)文件中讀取數(shù)據(jù)，性能也很高;而且不需要復(fù)制一份數(shù)據(jù)副本，并遠(yuǎn)程傳送到其他節(jié)點(diǎn)上。但是這里必須要注意的是，在實(shí)際的生產(chǎn)環(huán)境中，恐怕能夠直接用這種策略的場(chǎng)景還是有限的，如果RDD中數(shù)據(jù)比較多時(shí)(比如幾十億)，直接用這種持久化級(jí)別，會(huì)導(dǎo)致JVM的OOM內(nèi)存溢出異常。
• 如果使用MEMORY_ONLY級(jí)別時(shí)發(fā)生了內(nèi)存溢出，那么建議嘗試使用MEMORY_ONLY_SER級(jí)別。該級(jí)別會(huì)將RDD數(shù)據(jù)序列化后再保存在內(nèi)存中，此時(shí)每個(gè)partition僅僅是一個(gè)字節(jié)數(shù)組而已，大大減少了對(duì)象數(shù)量，并降低了內(nèi)存占用。這種級(jí)別比MEMORY_ONLY多出來(lái)的性能開(kāi)銷(xiāo)，主要就是序列化與反序列化的開(kāi)銷(xiāo)。但是后續(xù)算子可以基于純內(nèi)存進(jìn)行操作，因此性能總體還是比較高的。此外，可能發(fā)生的問(wèn)題同上，如果RDD中的數(shù)據(jù)量過(guò)多的話，還是可能會(huì)導(dǎo)致OOM內(nèi)存溢出的異常。
• 如果純內(nèi)存的級(jí)別都無(wú)法使用，那么建議使用MEMORY_AND_DISK_SER策略，而不是MEMORY_AND_DISK策略。因?yàn)榧热坏搅诉@一步，就說(shuō)明RDD的數(shù)據(jù)量很大，內(nèi)存無(wú)法完全放下。序列化后的數(shù)據(jù)比較少，可以節(jié)省內(nèi)存和磁盤(pán)的空間開(kāi)銷(xiāo)。同時(shí)該策略會(huì)優(yōu)先盡量嘗試將數(shù)據(jù)緩存在內(nèi)存中，內(nèi)存緩存不下才會(huì)寫(xiě)入磁盤(pán)。
• 通常不建議使用DISK_ONLY和后綴為_(kāi)2的級(jí)別：因?yàn)橥耆诖疟P(pán)文件進(jìn)行數(shù)據(jù)的讀寫(xiě)，會(huì)導(dǎo)致性能急劇降低，有時(shí)還不如重新計(jì)算一次所有RDD。后綴為_(kāi)2的級(jí)別，必須將所有數(shù)據(jù)都復(fù)制一份副本，并發(fā)送到其他節(jié)點(diǎn)上，數(shù)據(jù)復(fù)制以及網(wǎng)絡(luò)傳輸會(huì)導(dǎo)致較大的性能開(kāi)銷(xiāo)，除非是要求作業(yè)的高可用性，否則不建議使用。

原則四：盡量避免使用shuffle類(lèi)算子

如果有可能的話，要盡量避免使用shuffle類(lèi)算子。因?yàn)镾park作業(yè)運(yùn)行過(guò)程中，最消耗性能的地方就是shuffle過(guò)程。shuffle過(guò)程，簡(jiǎn)單來(lái)說(shuō)，就是將分布在集群中多個(gè)節(jié)點(diǎn)上的同一個(gè)key，拉取到同一個(gè)節(jié)點(diǎn)上，進(jìn)行聚合或join等操作。比如reduceByKey、join等算子，都會(huì)觸發(fā)shuffle操作。

shuffle過(guò)程中，各個(gè)節(jié)點(diǎn)上的相同key都會(huì)先寫(xiě)入本地磁盤(pán)文件中，然后其他節(jié)點(diǎn)需要通過(guò)網(wǎng)絡(luò)傳輸拉取各個(gè)節(jié)點(diǎn)上的磁盤(pán)文件中的相同key。而且相同key都拉取到同一個(gè)節(jié)點(diǎn)進(jìn)行聚合操作時(shí)，還有可能會(huì)因?yàn)橐粋€(gè)節(jié)點(diǎn)上處理的key過(guò)多，導(dǎo)致內(nèi)存不夠存放，進(jìn)而溢寫(xiě)到磁盤(pán)文件中。因此在shuffle過(guò)程中，可能會(huì)發(fā)生大量的磁盤(pán)文件讀寫(xiě)的IO操作，以及數(shù)據(jù)的網(wǎng)絡(luò)傳輸操作。磁盤(pán)IO和網(wǎng)絡(luò)數(shù)據(jù)傳輸也是shuffle性能較差的主要原因。

因此在我們的開(kāi)發(fā)過(guò)程中，能避免則盡可能避免使用reduceByKey、join、distinct、repartition等會(huì)進(jìn)行shuffle的算子，盡量使用map類(lèi)的非shuffle算子。這樣的話，沒(méi)有shuffle操作或者僅有較少shuffle操作的Spark作業(yè)，可以大大減少性能開(kāi)銷(xiāo)。

Broadcast與map進(jìn)行join代碼示例 

// 傳統(tǒng)的join操作會(huì)導(dǎo)致shuffle操作。 
// 因?yàn)閮蓚€(gè)RDD中，相同的key都需要通過(guò)網(wǎng)絡(luò)拉取到一個(gè)節(jié)點(diǎn)上，由一個(gè)task進(jìn)行join操作。 
val rdd3 = rdd1.join(rdd2) 
  
// Broadcast+map的join操作，不會(huì)導(dǎo)致shuffle操作。 
// 使用Broadcast將一個(gè)數(shù)據(jù)量較小的RDD作為廣播變量。 
val rdd2Data = rdd2.collect() 
val rdd2DataBroadcast = sc.broadcast(rdd2Data) 
  
// 在rdd1.map算子中，可以從rdd2DataBroadcast中，獲取rdd2的所有數(shù)據(jù)。 
// 然后進(jìn)行遍歷，如果發(fā)現(xiàn)rdd2中某條數(shù)據(jù)的key與rdd1的當(dāng)前數(shù)據(jù)的key是相同的，那么就判定可以進(jìn)行join。 
// 此時(shí)就可以根據(jù)自己需要的方式，將rdd1當(dāng)前數(shù)據(jù)與rdd2中可以連接的數(shù)據(jù)，拼接在一起（String或Tuple）。 
val rdd3 = rdd1.map(rdd2DataBroadcast...) 
  
// 注意，以上操作，建議僅僅在rdd2的數(shù)據(jù)量比較少（比如幾百M(fèi)，或者一兩G）的情況下使用。 
// 因?yàn)槊總€(gè)Executor的內(nèi)存中，都會(huì)駐留一份rdd2的全量數(shù)據(jù)。 

原則五：使用map-side預(yù)聚合的shuffle操作

如果因?yàn)闃I(yè)務(wù)需要，一定要使用shuffle操作，無(wú)法用map類(lèi)的算子來(lái)替代，那么盡量使用可以map-side預(yù)聚合的算子。

所謂的map-side預(yù)聚合，說(shuō)的是在每個(gè)節(jié)點(diǎn)本地對(duì)相同的key進(jìn)行一次聚合操作，類(lèi)似于MapReduce中的本地combiner。map-side預(yù)聚合之后，每個(gè)節(jié)點(diǎn)本地就只會(huì)有一條相同的key，因?yàn)槎鄺l相同的key都被聚合起來(lái)了。其他節(jié)點(diǎn)在拉取所有節(jié)點(diǎn)上的相同key時(shí)，就會(huì)大大減少需要拉取的數(shù)據(jù)數(shù)量，從而也就減少了磁盤(pán)IO以及網(wǎng)絡(luò)傳輸開(kāi)銷(xiāo)。通常來(lái)說(shuō)，在可能的情況下，建議使用reduceByKey或者aggregateByKey算子來(lái)替代掉groupByKey算子。因?yàn)閞educeByKey和aggregateByKey算子都會(huì)使用用戶(hù)自定義的函數(shù)對(duì)每個(gè)節(jié)點(diǎn)本地的相同key進(jìn)行預(yù)聚合。而groupByKey算子是不會(huì)進(jìn)行預(yù)聚合的，全量的數(shù)據(jù)會(huì)在集群的各個(gè)節(jié)點(diǎn)之間分發(fā)和傳輸，性能相對(duì)來(lái)說(shuō)比較差。

比如如下兩幅圖，就是典型的例子，分別基于reduceByKey和groupByKey進(jìn)行單詞計(jì)數(shù)。其中***張圖是groupByKey的原理圖，可以看到，沒(méi)有進(jìn)行任何本地聚合時(shí)，所有數(shù)據(jù)都會(huì)在集群節(jié)點(diǎn)之間傳輸;第二張圖是reduceByKey的原理圖，可以看到，每個(gè)節(jié)點(diǎn)本地的相同key數(shù)據(jù)，都進(jìn)行了預(yù)聚合，然后才傳輸?shù)狡渌?jié)點(diǎn)上進(jìn)行全局聚合。

原則六：使用高性能的算子

除了shuffle相關(guān)的算子有優(yōu)化原則之外，其他的算子也都有著相應(yīng)的優(yōu)化原則。

使用reduceByKey/aggregateByKey替代groupByKey

詳情見(jiàn)“原則五：使用map-side預(yù)聚合的shuffle操作”。

使用mapPartitions替代普通map

mapPartitions類(lèi)的算子，一次函數(shù)調(diào)用會(huì)處理一個(gè)partition所有的數(shù)據(jù)，而不是一次函數(shù)調(diào)用處理一條，性能相對(duì)來(lái)說(shuō)會(huì)高一些。但是有的時(shí)候，使用mapPartitions會(huì)出現(xiàn)OOM(內(nèi)存溢出)的問(wèn)題。因?yàn)閱未魏瘮?shù)調(diào)用就要處理掉一個(gè)partition所有的數(shù)據(jù)，如果內(nèi)存不夠，垃圾回收時(shí)是無(wú)法回收掉太多對(duì)象的，很可能出現(xiàn)OOM異常。所以使用這類(lèi)操作時(shí)要慎重!

使用foreachPartitions替代foreach

原理類(lèi)似于“使用mapPartitions替代map”，也是一次函數(shù)調(diào)用處理一個(gè)partition的所有數(shù)據(jù)，而不是一次函數(shù)調(diào)用處理一條數(shù)據(jù)。在實(shí)踐中發(fā)現(xiàn)，foreachPartitions類(lèi)的算子，對(duì)性能的提升還是很有幫助的。比如在foreach函數(shù)中，將RDD中所有數(shù)據(jù)寫(xiě)MySQL，那么如果是普通的foreach算子，就會(huì)一條數(shù)據(jù)一條數(shù)據(jù)地寫(xiě)，每次函數(shù)調(diào)用可能就會(huì)創(chuàng)建一個(gè)數(shù)據(jù)庫(kù)連接，此時(shí)就勢(shì)必會(huì)頻繁地創(chuàng)建和銷(xiāo)毀數(shù)據(jù)庫(kù)連接，性能是非常低下;但是如果用foreachPartitions算子一次性處理一個(gè)partition的數(shù)據(jù)，那么對(duì)于每個(gè)partition，只要?jiǎng)?chuàng)建一個(gè)數(shù)據(jù)庫(kù)連接即可，然后執(zhí)行批量插入操作，此時(shí)性能是比較高的。實(shí)踐中發(fā)現(xiàn)，對(duì)于1萬(wàn)條左右的數(shù)據(jù)量寫(xiě)MySQL，性能可以提升30%以上。

使用filter之后進(jìn)行coalesce操作*

通常對(duì)一個(gè)RDD執(zhí)行filter算子過(guò)濾掉RDD中較多數(shù)據(jù)后(比如30%以上的數(shù)據(jù))，建議使用coalesce算子，手動(dòng)減少RDD的partition數(shù)量，將RDD中的數(shù)據(jù)壓縮到更少的partition中去。因?yàn)閒ilter之后，RDD的每個(gè)partition中都會(huì)有很多數(shù)據(jù)被過(guò)濾掉，此時(shí)如果照常進(jìn)行后續(xù)的計(jì)算，其實(shí)每個(gè)task處理的partition中的數(shù)據(jù)量并不是很多，有一點(diǎn)資源浪費(fèi)，而且此時(shí)處理的task越多，可能速度反而越慢。因此用coalesce減少partition數(shù)量，將RDD中的數(shù)據(jù)壓縮到更少的partition之后，只要使用更少的task即可處理完所有的partition。在某些場(chǎng)景下，對(duì)于性能的提升會(huì)有一定的幫助。

使用repartitionAndSortWithinPartitions替代repartition與sort類(lèi)操作

repartitionAndSortWithinPartitions是Spark官網(wǎng)推薦的一個(gè)算子，官方建議，如果需要在repartition重分區(qū)之后，還要進(jìn)行排序，建議直接使用repartitionAndSortWithinPartitions算子。因?yàn)樵撍阕涌梢砸贿呥M(jìn)行重分區(qū)的shuffle操作，一邊進(jìn)行排序。shuffle與sort兩個(gè)操作同時(shí)進(jìn)行，比先shuffle再sort來(lái)說(shuō)，性能可能是要高的。

原則七：廣播大變量

有時(shí)在開(kāi)發(fā)過(guò)程中，會(huì)遇到需要在算子函數(shù)中使用外部變量的場(chǎng)景(尤其是大變量，比如100M以上的大集合)，那么此時(shí)就應(yīng)該使用Spark的廣播(Broadcast)功能來(lái)提升性能。

在算子函數(shù)中使用到外部變量時(shí)，默認(rèn)情況下，Spark會(huì)將該變量復(fù)制多個(gè)副本，通過(guò)網(wǎng)絡(luò)傳輸?shù)絫ask中，此時(shí)每個(gè)task都有一個(gè)變量副本。如果變量本身比較大的話(比如100M，甚至1G)，那么大量的變量副本在網(wǎng)絡(luò)中傳輸?shù)男阅荛_(kāi)銷(xiāo)，以及在各個(gè)節(jié)點(diǎn)的Executor中占用過(guò)多內(nèi)存導(dǎo)致的頻繁GC，都會(huì)極大地影響性能。

因此對(duì)于上述情況，如果使用的外部變量比較大，建議使用Spark的廣播功能，對(duì)該變量進(jìn)行廣播。廣播后的變量，會(huì)保證每個(gè)Executor的內(nèi)存中，只駐留一份變量副本，而Executor中的task執(zhí)行時(shí)共享該Executor中的那份變量副本。這樣的話，可以大大減少變量副本的數(shù)量，從而減少網(wǎng)絡(luò)傳輸?shù)男阅荛_(kāi)銷(xiāo)，并減少對(duì)Executor內(nèi)存的占用開(kāi)銷(xiāo)，降低GC的頻率。

廣播大變量的代碼示例 

// 以下代碼在算子函數(shù)中，使用了外部的變量。 
// 此時(shí)沒(méi)有做任何特殊操作，每個(gè)task都會(huì)有一份list1的副本。 
val list1 = ... 
rdd1.map(list1...) 
  
// 以下代碼將list1封裝成了Broadcast類(lèi)型的廣播變量。 
// 在算子函數(shù)中，使用廣播變量時(shí)，首先會(huì)判斷當(dāng)前task所在Executor內(nèi)存中，是否有變量副本。 
// 如果有則直接使用；如果沒(méi)有則從Driver或者其他Executor節(jié)點(diǎn)上遠(yuǎn)程拉取一份放到本地Executor內(nèi)存中。 
// 每個(gè)Executor內(nèi)存中，就只會(huì)駐留一份廣播變量副本。 
val list1 = ... 
val list1Broadcast = sc.broadcast(list1) 
rdd1.map(list1Broadcast...) 

原則八：使用Kryo優(yōu)化序列化性能

在Spark中，主要有三個(gè)地方涉及到了序列化：

• 在算子函數(shù)中使用到外部變量時(shí)，該變量會(huì)被序列化后進(jìn)行網(wǎng)絡(luò)傳輸(見(jiàn)“原則七：廣播大變量”中的講解)。
• 將自定義的類(lèi)型作為RDD的泛型類(lèi)型時(shí)(比如JavaRDD，Student是自定義類(lèi)型)，所有自定義類(lèi)型對(duì)象，都會(huì)進(jìn)行序列化。因此這種情況下，也要求自定義的類(lèi)必須實(shí)現(xiàn)Serializable接口。
• 使用可序列化的持久化策略時(shí)(比如MEMORY_ONLY_SER)，Spark會(huì)將RDD中的每個(gè)partition都序列化成一個(gè)大的字節(jié)數(shù)組。

對(duì)于這三種出現(xiàn)序列化的地方，我們都可以通過(guò)使用Kryo序列化類(lèi)庫(kù)，來(lái)優(yōu)化序列化和反序列化的性能。Spark默認(rèn)使用的是Java的序列化機(jī)制，也就是ObjectOutputStream/ObjectInputStream API來(lái)進(jìn)行序列化和反序列化。但是Spark同時(shí)支持使用Kryo序列化庫(kù)，Kryo序列化類(lèi)庫(kù)的性能比Java序列化類(lèi)庫(kù)的性能要高很多。官方介紹，Kryo序列化機(jī)制比Java序列化機(jī)制，性能高10倍左右。Spark之所以默認(rèn)沒(méi)有使用Kryo作為序列化類(lèi)庫(kù)，是因?yàn)镵ryo要求***要注冊(cè)所有需要進(jìn)行序列化的自定義類(lèi)型，因此對(duì)于開(kāi)發(fā)者來(lái)說(shuō)，這種方式比較麻煩。

以下是使用Kryo的代碼示例，我們只要設(shè)置序列化類(lèi)，再注冊(cè)要序列化的自定義類(lèi)型即可(比如算子函數(shù)中使用到的外部變量類(lèi)型、作為RDD泛型類(lèi)型的自定義類(lèi)型等)： 

// 創(chuàng)建SparkConf對(duì)象。 
val conf = new SparkConf().setMaster(...).setAppName(...) 
// 設(shè)置序列化器為KryoSerializer。 
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") 
// 注冊(cè)要序列化的自定義類(lèi)型。 
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2])) 

原則九：優(yōu)化數(shù)據(jù)結(jié)構(gòu)

Java中，有三種類(lèi)型比較耗費(fèi)內(nèi)存：

• 對(duì)象，每個(gè)Java對(duì)象都有對(duì)象頭、引用等額外的信息，因此比較占用內(nèi)存空間。
• 字符串，每個(gè)字符串內(nèi)部都有一個(gè)字符數(shù)組以及長(zhǎng)度等額外信息。
• 集合類(lèi)型，比如HashMap、LinkedList等，因?yàn)榧项?lèi)型內(nèi)部通常會(huì)使用一些內(nèi)部類(lèi)來(lái)封裝集合元素，比如Map.Entry。

因此Spark官方建議，在Spark編碼實(shí)現(xiàn)中，特別是對(duì)于算子函數(shù)中的代碼，盡量不要使用上述三種數(shù)據(jù)結(jié)構(gòu)，盡量使用字符串替代對(duì)象，使用原始類(lèi)型(比如Int、Long)替代字符串，使用數(shù)組替代集合類(lèi)型，這樣盡可能地減少內(nèi)存占用，從而降低GC頻率，提升性能。

但是在筆者的編碼實(shí)踐中發(fā)現(xiàn)，要做到該原則其實(shí)并不容易。因?yàn)槲覀兺瑫r(shí)要考慮到代碼的可維護(hù)性，如果一個(gè)代碼中，完全沒(méi)有任何對(duì)象抽象，全部是字符串拼接的方式，那么對(duì)于后續(xù)的代碼維護(hù)和修改，無(wú)疑是一場(chǎng)巨大的災(zāi)難。同理，如果所有操作都基于數(shù)組實(shí)現(xiàn)，而不使用HashMap、LinkedList等集合類(lèi)型，那么對(duì)于我們的編碼難度以及代碼可維護(hù)性，也是一個(gè)極大的挑戰(zhàn)。因此筆者建議，在可能以及合適的情況下，使用占用內(nèi)存較少的數(shù)據(jù)結(jié)構(gòu)，但是前提是要保證代碼的可維護(hù)性。