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

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

1． RDD復(fù)用

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

RDD的重復(fù)計(jì)算

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

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

2． 盡早filter

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

3． 讀取大量小文件－用wholeTextFiles

當(dāng)我們將一個(gè)文本文件讀取為 RDD 時(shí)，輸入的每一行都會(huì)成為RDD的一個(gè)元素。

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

val input：RDD［String］ ＝ sc．textFile（＂dir．log＂）

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

但是這樣對(duì)于大量的小文件讀取效率并不高，應(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（＿…．）  表示每一個(gè)元素

mapPartitions（＿…．）  表示每個(gè)分區(qū)的數(shù)據(jù)組成的迭代器

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

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

map 算子

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

mapPartition 算子

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

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

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

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

foreachPartition

rrd．foreache（＿…．） 表示每一個(gè)元素

rrd．forPartitions（＿…．）  表示每個(gè)分區(qū)的數(shù)據(jù)組成的迭代器

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

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

與mapPartitions算子非常相似，foreachPartition是將RDD的每個(gè)分區(qū)作為遍歷對(duì)象，一次處理一個(gè)分區(qū)的數(shù)據(jù)，也就是說，如果涉及數(shù)據(jù)庫(kù)的相關(guān)操作，一個(gè)分區(qū)的數(shù)據(jù)只需要?jiǎng)?chuàng)建一次數(shù)據(jù)庫(kù)連接，如下圖所示：

foreachPartition 算子

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

對(duì)于我們寫的function函數(shù)，一次處理一整個(gè)分區(qū)的數(shù)據(jù)；

對(duì)于一個(gè)分區(qū)內(nèi)的數(shù)據(jù)，創(chuàng)建唯一的數(shù)據(jù)庫(kù)連接；

只需要向數(shù)據(jù)庫(kù)發(fā)送一次SQL語句和多組參數(shù)；

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