RDD算子調(diào)優(yōu)

不廢話，直接進(jìn)入正題滩援！

1. RDD復(fù)用

在對RDD進(jìn)行算子時栅隐，要避免相同的算子和計算邏輯之下對RDD進(jìn)行重復(fù)的計算，如下圖所示：

RDD的重復(fù)計算

對上圖中的RDD計算架構(gòu)進(jìn)行修改玩徊，得到如下圖所示的優(yōu)化結(jié)果：

RDD架構(gòu)優(yōu)化

2. 盡早filter

獲取到初始RDD后租悄，應(yīng)該考慮盡早地過濾掉不需要的數(shù)據(jù)，進(jìn)而減少對內(nèi)存的占用恩袱，從而提升Spark作業(yè)的運(yùn)行效率泣棋。

本文首發(fā)于公眾號：五分鐘學(xué)大數(shù)據(jù)，歡迎圍觀

3. 讀取大量小文件-用wholeTextFiles

當(dāng)我們將一個文本文件讀取為 RDD 時畔塔，輸入的每一行都會成為RDD的一個元素潭辈。

也可以將多個完整的文本文件一次性讀取為一個pairRDD，其中鍵是文件名澈吨，值是文件內(nèi)容把敢。

val input:RDD[String] = sc.textFile("dir/*.log") 

如果傳遞目錄，則將目錄下的所有文件讀取作為RDD谅辣。文件路徑支持通配符修赞。

但是這樣對于大量的小文件讀取效率并不高，應(yīng)該使用 wholeTextFiles
返回值為RDD[(String, String)]桑阶，其中Key是文件的名稱柏副，Value是文件的內(nèi)容。

def wholeTextFiles(path: String, minPartitions: Int = defaultMinPartitions): RDD[(String, String)])

wholeTextFiles讀取小文件:

val filesRDD: RDD[(String, String)] =
sc.wholeTextFiles("D:\\data\\files", minPartitions = 3)
val linesRDD: RDD[String] = filesRDD.flatMap(_._2.split("\\r\\n"))
val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(" "))
wordsRDD.map((_, 1)).reduceByKey(_ + _).collect().foreach(println)

4. mapPartition和foreachPartition

• mapPartitions

map(_....) 表示每一個元素

mapPartitions(_....) 表示每個分區(qū)的數(shù)據(jù)組成的迭代器

普通的map算子對RDD中的每一個元素進(jìn)行操作蚣录，而mapPartitions算子對RDD中每一個分區(qū)進(jìn)行操作割择。

如果是普通的map算子，假設(shè)一個partition有1萬條數(shù)據(jù)包归，那么map算子中的function要執(zhí)行1萬次锨推，也就是對每個元素進(jìn)行操作铅歼。

map 算子

如果是mapPartition算子，由于一個task處理一個RDD的partition换可，那么一個task只會執(zhí)行一次function椎椰，function一次接收所有的partition數(shù)據(jù)，效率比較高沾鳄。

mapPartition 算子

比如慨飘，當(dāng)要把RDD中的所有數(shù)據(jù)通過JDBC寫入數(shù)據(jù)，如果使用map算子译荞，那么需要對RDD中的每一個元素都創(chuàng)建一個數(shù)據(jù)庫連接瓤的，這樣對資源的消耗很大，如果使用mapPartitions算子吞歼，那么針對一個分區(qū)的數(shù)據(jù)圈膏，只需要建立一個數(shù)據(jù)庫連接。

mapPartitions算子也存在一些缺點(diǎn)：對于普通的map操作篙骡，一次處理一條數(shù)據(jù)稽坤，如果在處理了2000條數(shù)據(jù)后內(nèi)存不足，那么可以將已經(jīng)處理完的2000條數(shù)據(jù)從內(nèi)存中垃圾回收掉糯俗；但是如果使用mapPartitions算子尿褪，但數(shù)據(jù)量非常大時，function一次處理一個分區(qū)的數(shù)據(jù)得湘，如果一旦內(nèi)存不足杖玲，此時無法回收內(nèi)存，就可能會OOM淘正，即內(nèi)存溢出摆马。

因此，mapPartitions算子適用于數(shù)據(jù)量不是特別大的時候跪帝，此時使用mapPartitions算子對性能的提升效果還是不錯的今膊。（當(dāng)數(shù)據(jù)量很大的時候，一旦使用mapPartitions算子伞剑，就會直接OOM）

在項目中斑唬，應(yīng)該首先估算一下RDD的數(shù)據(jù)量、每個partition的數(shù)據(jù)量黎泣，以及分配給每個Executor的內(nèi)存資源恕刘，如果資源允許，可以考慮使用mapPartitions算子代替map抒倚。

• foreachPartition

rrd.foreache(_....) 表示每一個元素

rrd.forPartitions(_....) 表示每個分區(qū)的數(shù)據(jù)組成的迭代器

在生產(chǎn)環(huán)境中褐着，通常使用foreachPartition算子來完成數(shù)據(jù)庫的寫入，通過foreachPartition算子的特性托呕，可以優(yōu)化寫數(shù)據(jù)庫的性能含蓉。

如果使用foreach算子完成數(shù)據(jù)庫的操作频敛，由于foreach算子是遍歷RDD的每條數(shù)據(jù)，因此馅扣，每條數(shù)據(jù)都會建立一個數(shù)據(jù)庫連接斟赚，這是對資源的極大浪費(fèi)，因此差油，對于寫數(shù)據(jù)庫操作拗军，我們應(yīng)當(dāng)使用foreachPartition算子。

與mapPartitions算子非常相似蓄喇，foreachPartition是將RDD的每個分區(qū)作為遍歷對象发侵，一次處理一個分區(qū)的數(shù)據(jù)，也就是說妆偏，如果涉及數(shù)據(jù)庫的相關(guān)操作刃鳄，一個分區(qū)的數(shù)據(jù)只需要創(chuàng)建一次數(shù)據(jù)庫連接，如下圖所示：

foreachPartition 算子

使用了foreachPartition 算子后钱骂，可以獲得以下的性能提升：

1. 對于我們寫的function函數(shù)铲汪，一次處理一整個分區(qū)的數(shù)據(jù)；
2. 對于一個分區(qū)內(nèi)的數(shù)據(jù)罐柳，創(chuàng)建唯一的數(shù)據(jù)庫連接；
3. 只需要向數(shù)據(jù)庫發(fā)送一次SQL語句和多組參數(shù)狰住；

在生產(chǎn)環(huán)境中张吉，全部都會使用foreachPartition算子完成數(shù)據(jù)庫操作。foreachPartition算子存在一個問題催植，與mapPartitions算子類似肮蛹，如果一個分區(qū)的數(shù)據(jù)量特別大，可能會造成OOM创南，即內(nèi)存溢出伦忠。

5. filter+coalesce/repartition(減少分區(qū))

在Spark任務(wù)中我們經(jīng)常會使用filter算子完成RDD中數(shù)據(jù)的過濾，在任務(wù)初始階段稿辙，從各個分區(qū)中加載到的數(shù)據(jù)量是相近的昆码，但是一旦進(jìn)過filter過濾后，每個分區(qū)的數(shù)據(jù)量有可能會存在較大差異邻储，如下圖所示：

分區(qū)數(shù)據(jù)過濾結(jié)果

根據(jù)上圖我們可以發(fā)現(xiàn)兩個問題：

1. 每個partition的數(shù)據(jù)量變小了赋咽，如果還按照之前與partition相等的task個數(shù)去處理當(dāng)前數(shù)據(jù)，有點(diǎn)浪費(fèi)task的計算資源吨娜；

2. 每個partition的數(shù)據(jù)量不一樣脓匿，會導(dǎo)致后面的每個task處理每個partition數(shù)據(jù)的時候，每個task要處理的數(shù)據(jù)量不同宦赠，這很有可能導(dǎo)致數(shù)據(jù)傾斜問題陪毡。

如上圖所示米母，第二個分區(qū)的數(shù)據(jù)過濾后只剩100條，而第三個分區(qū)的數(shù)據(jù)過濾后剩下800條毡琉，在相同的處理邏輯下铁瞒，第二個分區(qū)對應(yīng)的task處理的數(shù)據(jù)量與第三個分區(qū)對應(yīng)的task處理的數(shù)據(jù)量差距達(dá)到了8倍，這也會導(dǎo)致運(yùn)行速度可能存在數(shù)倍的差距绊起，這也就是數(shù)據(jù)傾斜問題精拟。

針對上述的兩個問題，我們分別進(jìn)行分析：

1. 針對第一個問題虱歪，既然分區(qū)的數(shù)據(jù)量變小了蜂绎，我們希望可以對分區(qū)數(shù)據(jù)進(jìn)行重新分配，比如將原來4個分區(qū)的數(shù)據(jù)轉(zhuǎn)化到2個分區(qū)中笋鄙，這樣只需要用后面的兩個task進(jìn)行處理即可师枣，避免了資源的浪費(fèi)。

2. 針對第二個問題萧落，解決方法和第一個問題的解決方法非常相似践美，對分區(qū)數(shù)據(jù)重新分配，讓每個partition中的數(shù)據(jù)量差不多找岖，這就避免了數(shù)據(jù)傾斜問題陨倡。

那么具體應(yīng)該如何實(shí)現(xiàn)上面的解決思路？我們需要coalesce算子许布。

repartition與coalesce都可以用來進(jìn)行重分區(qū)兴革，其中repartition只是coalesce接口中shuffle為true的簡易實(shí)現(xiàn)，coalesce默認(rèn)情況下不進(jìn)行shuffle蜜唾，但是可以通過參數(shù)進(jìn)行設(shè)置杂曲。

假設(shè)我們希望將原本的分區(qū)個數(shù)A通過重新分區(qū)變?yōu)锽，那么有以下幾種情況：

1. A > B（多數(shù)分區(qū)合并為少數(shù)分區(qū)）

   • A與B相差值不大

     此時使用coalesce即可袁余，無需shuffle過程擎勘。

   • A與B相差值很大

     此時可以使用coalesce并且不啟用shuffle過程，但是會導(dǎo)致合并過程性能低下颖榜，所以推薦設(shè)置coalesce的第二個參數(shù)為true棚饵，即啟動shuffle過程。

2. A < B（少數(shù)分區(qū)分解為多數(shù)分區(qū)）

此時使用repartition即可朱转，如果使用coalesce需要將shuffle設(shè)置為true蟹地，否則coalesce無效。

我們可以在filter操作之后藤为，使用coalesce算子針對每個partition的數(shù)據(jù)量各不相同的情況怪与，壓縮partition的數(shù)量，而且讓每個partition的數(shù)據(jù)量盡量均勻緊湊缅疟，以便于后面的task進(jìn)行計算操作分别，在某種程度上能夠在一定程度上提升性能遍愿。

注意：local模式是進(jìn)程內(nèi)模擬集群運(yùn)行，已經(jīng)對并行度和分區(qū)數(shù)量有了一定的內(nèi)部優(yōu)化耘斩，因此不用去設(shè)置并行度和分區(qū)數(shù)量沼填。

6. 并行度設(shè)置

Spark作業(yè)中的并行度指各個stage的task的數(shù)量。

如果并行度設(shè)置不合理而導(dǎo)致并行度過低括授，會導(dǎo)致資源的極大浪費(fèi)坞笙，例如，20個Executor荚虚，每個Executor分配3個CPU core薛夜，而Spark作業(yè)有40個task，這樣每個Executor分配到的task個數(shù)是2個版述，這就使得每個Executor有一個CPU core空閑梯澜，導(dǎo)致資源的浪費(fèi)。

理想的并行度設(shè)置渴析，應(yīng)該是讓并行度與資源相匹配晚伙，簡單來說就是在資源允許的前提下，并行度要設(shè)置的盡可能大俭茧，達(dá)到可以充分利用集群資源咆疗。合理的設(shè)置并行度，可以提升整個Spark作業(yè)的性能和運(yùn)行速度母债。

Spark官方推薦民傻，task數(shù)量應(yīng)該設(shè)置為Spark作業(yè)總CPU core數(shù)量的2~3倍。之所以沒有推薦task數(shù)量與CPU core總數(shù)相等场斑，是因?yàn)閠ask的執(zhí)行時間不同，有的task執(zhí)行速度快而有的task執(zhí)行速度慢牵署，如果task數(shù)量與CPU core總數(shù)相等漏隐，那么執(zhí)行快的task執(zhí)行完成后，會出現(xiàn)CPU core空閑的情況奴迅。如果task數(shù)量設(shè)置為CPU core總數(shù)的2~3倍青责，那么一個task執(zhí)行完畢后，CPU core會立刻執(zhí)行下一個task取具，降低了資源的浪費(fèi)脖隶，同時提升了Spark作業(yè)運(yùn)行的效率。

Spark作業(yè)并行度的設(shè)置如下：

val conf = new SparkConf().set("spark.default.parallelism", "500")

原則：讓 cpu 的 Core（cpu 核心數(shù)） 充分利用起來暇检，
如有100個 Core,那么并行度可以設(shè)置為200~300产阱。

7. repartition/coalesce調(diào)節(jié)并行度

我們知道 Spark 中有并行度的調(diào)節(jié)策略，但是块仆，并行度的設(shè)置對于Spark SQL是不生效的构蹬，用戶設(shè)置的并行度只對于Spark SQL以外的所有Spark的stage生效王暗。

Spark SQL的并行度不允許用戶自己指定，Spark SQL自己會默認(rèn)根據(jù)hive表對應(yīng)的HDFS文件的split個數(shù)自動設(shè)置Spark SQL所在的那個stage的并行度庄敛，用戶自己通 spark.default.parallelism 參數(shù)指定的并行度俗壹，只會在沒Spark SQL的stage中生效。

由于Spark SQL所在stage的并行度無法手動設(shè)置藻烤，如果數(shù)據(jù)量較大绷雏，并且此stage中后續(xù)的transformation操作有著復(fù)雜的業(yè)務(wù)邏輯，而Spark SQL自動設(shè)置的task數(shù)量很少怖亭，這就意味著每個task要處理為數(shù)不少的數(shù)據(jù)量涎显，然后還要執(zhí)行非常復(fù)雜的處理邏輯，這就可能表現(xiàn)為第一個有Spark SQL的stage速度很慢依许，而后續(xù)的沒有Spark SQL的stage運(yùn)行速度非彻缀蹋快。

為了解決Spark SQL無法設(shè)置并行度和task數(shù)量的問題峭跳，我們可以使用repartition算子膘婶。

repartition 算子使用前后對比圖如下：

repartition 算子使用前后對比圖

Spark SQL這一步的并行度和task數(shù)量肯定是沒有辦法去改變了，但是蛀醉，對于Spark SQL查詢出來的RDD悬襟，立即使用repartition算子，去重新進(jìn)行分區(qū)拯刁，這樣可以重新分區(qū)為多個partition脊岳，從repartition之后的RDD操作，由于不再涉及Spark SQL垛玻，因此stage的并行度就會等于你手動設(shè)置的值割捅，這樣就避免了Spark SQL所在的stage只能用少量的task去處理大量數(shù)據(jù)并執(zhí)行復(fù)雜的算法邏輯。使用repartition算子的前后對比如上圖所示帚桩。

8. reduceByKey本地預(yù)聚合

reduceByKey相較于普通的shuffle操作一個顯著的特點(diǎn)就是會進(jìn)行map端的本地聚合亿驾，map端會先對本地的數(shù)據(jù)進(jìn)行combine操作，然后將數(shù)據(jù)寫入給下個stage的每個task創(chuàng)建的文件中账嚎，也就是在map端莫瞬，對每一個key對應(yīng)的value，執(zhí)行reduceByKey算子函數(shù)郭蕉。

reduceByKey算子的執(zhí)行過程如下圖所示：

reduceByKey 算子執(zhí)行過程

使用reduceByKey對性能的提升如下：

1. 本地聚合后疼邀，在map端的數(shù)據(jù)量變少，減少了磁盤IO召锈，也減少了對磁盤空間的占用旁振；
2. 本地聚合后，下一個stage拉取的數(shù)據(jù)量變少，減少了網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量规求；
3. 本地聚合后筐付，在reduce端進(jìn)行數(shù)據(jù)緩存的內(nèi)存占用減少；
4. 本地聚合后阻肿，在reduce端進(jìn)行聚合的數(shù)據(jù)量減少瓦戚。

基于reduceByKey的本地聚合特征，我們應(yīng)該考慮使用reduceByKey代替其他的shuffle算子丛塌，例如groupByKey较解。

groupByKey與reduceByKey的運(yùn)行原理如下圖1和圖2所示：

圖1：groupByKey原理

圖2：reduceByKey原理

根據(jù)上圖可知，groupByKey不會進(jìn)行map端的聚合赴邻，而是將所有map端的數(shù)據(jù)shuffle到reduce端印衔，然后在reduce端進(jìn)行數(shù)據(jù)的聚合操作。由于reduceByKey有map端聚合的特性姥敛，使得網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量減小奸焙，因此效率要明顯高于groupByKey。

9. 使用持久化+checkpoint

Spark持久化在大部分情況下是沒有問題的彤敛，但是有時數(shù)據(jù)可能會丟失与帆，如果數(shù)據(jù)一旦丟失，就需要對丟失的數(shù)據(jù)重新進(jìn)行計算墨榄，計算完后再緩存和使用玄糟，為了避免數(shù)據(jù)的丟失，可以選擇對這個RDD進(jìn)行checkpoint袄秩，也就是將數(shù)據(jù)持久化一份到容錯的文件系統(tǒng)上（比如HDFS）阵翎。

一個RDD緩存并checkpoint后，如果一旦發(fā)現(xiàn)緩存丟失之剧，就會優(yōu)先查看checkpoint數(shù)據(jù)存不存在郭卫，如果有，就會使用checkpoint數(shù)據(jù)背稼，而不用重新計算芋簿。也即是說犀填，checkpoint可以視為cache的保障機(jī)制猎拨，如果cache失敗颈畸，就使用checkpoint的數(shù)據(jù)拢肆。

使用checkpoint的優(yōu)點(diǎn)在于提高了Spark作業(yè)的可靠性援制，一旦緩存出現(xiàn)問題迄本，不需要重新計算數(shù)據(jù)爷辙，缺點(diǎn)在于聘萨，checkpoint時需要將數(shù)據(jù)寫入HDFS等文件系統(tǒng)竹椒，對性能的消耗較大。

持久化設(shè)置如下：

sc.setCheckpointDir(‘HDFS’)
rdd.cache/persist(memory_and_disk)
rdd.checkpoint

10. 使用廣播變量

默認(rèn)情況下米辐，task中的算子中如果使用了外部的變量胸完，每個task都會獲取一份變量的復(fù)本书释，這就造成了內(nèi)存的極大消耗。一方面赊窥，如果后續(xù)對RDD進(jìn)行持久化爆惧，可能就無法將RDD數(shù)據(jù)存入內(nèi)存，只能寫入磁盤锨能，磁盤IO將會嚴(yán)重消耗性能扯再；另一方面，task在創(chuàng)建對象的時候址遇，也許會發(fā)現(xiàn)堆內(nèi)存無法存放新創(chuàng)建的對象熄阻，這就會導(dǎo)致頻繁的GC，GC會導(dǎo)致工作線程停止倔约，進(jìn)而導(dǎo)致Spark暫停工作一段時間秃殉，嚴(yán)重影響Spark性能。

假設(shè)當(dāng)前任務(wù)配置了20個Executor浸剩，指定500個task钾军，有一個20M的變量被所有task共用，此時會在500個task中產(chǎn)生500個副本乒省，耗費(fèi)集群10G的內(nèi)存巧颈，如果使用了廣播變量， 那么每個Executor保存一個副本袖扛，一共消耗400M內(nèi)存砸泛，內(nèi)存消耗減少了5倍。

廣播變量在每個Executor保存一個副本蛆封，此Executor的所有task共用此廣播變量唇礁，這讓變量產(chǎn)生的副本數(shù)量大大減少。

在初始階段惨篱，廣播變量只在Driver中有一份副本盏筐。task在運(yùn)行的時候，想要使用廣播變量中的數(shù)據(jù)砸讳，此時首先會在自己本地的Executor對應(yīng)的BlockManager中嘗試獲取變量琢融，如果本地沒有，BlockManager就會從Driver或者其他節(jié)點(diǎn)的BlockManager上遠(yuǎn)程拉取變量的復(fù)本簿寂，并由本地的BlockManager進(jìn)行管理漾抬；之后此Executor的所有task都會直接從本地的BlockManager中獲取變量。

對于多個Task可能會共用的數(shù)據(jù)可以廣播到每個Executor上：

val 廣播變量名= sc.broadcast(會被各個Task用到的變量,即需要廣播的變量)

廣播變量名.value//獲取廣播變量

11. 使用Kryo序列化

默認(rèn)情況下常遂，Spark使用Java的序列化機(jī)制纳令。Java的序列化機(jī)制使用方便，不需要額外的配置，在算子中使用的變量實(shí)現(xiàn)Serializable接口即可平绩，但是圈匆，Java序列化機(jī)制的效率不高，序列化速度慢并且序列化后的數(shù)據(jù)所占用的空間依然較大捏雌。

Spark官方宣稱Kryo序列化機(jī)制比Java序列化機(jī)制性能提高10倍左右跃赚，Spark之所以沒有默認(rèn)使用Kryo作為序列化類庫，是因?yàn)?strong>它不支持所有對象的序列化腹忽，同時Kryo需要用戶在使用前注冊需要序列化的類型来累，不夠方便，但從Spark 2.0.0版本開始窘奏，簡單類型嘹锁、簡單類型數(shù)組、字符串類型的Shuffling RDDs 已經(jīng)默認(rèn)使用Kryo序列化方式了着裹。

Kryo序列化注冊方式的代碼如下：

public class MyKryoRegistrator implements KryoRegistrator{
  @Override
  public void registerClasses(Kryo kryo){
    kryo.register(StartupReportLogs.class);
  }
}

配置Kryo序列化方式的代碼如下：

//創(chuàng)建SparkConf對象
val conf = new SparkConf().setMaster(…).setAppName(…)
//使用Kryo序列化庫
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");  
//在Kryo序列化庫中注冊自定義的類集合
conf.set("spark.kryo.registrator", "bigdata.com.MyKryoRegistrator");

作者：五分鐘學(xué)大數(shù)據(jù)
原文鏈接：https://www.cnblogs.com/itlz/p/14487244.html