中國IDC圈4月22日?qǐng)?bào)道，這篇文章算是個(gè)科普貼。假如已經(jīng)熟悉Spark的就略過吧。

媒介

許多初學(xué)者其實(shí)對(duì)Spark的編程模式照舊RDD這個(gè)觀念領(lǐng)略不到位，就會(huì)發(fā)生一些誤解。

好比，許多時(shí)候我們經(jīng)常覺得一個(gè)文件是會(huì)被完整讀入到內(nèi)存，然后做各類調(diào)動(dòng)，這很大概是受兩個(gè)觀念的誤導(dǎo)：

RDD的界說，RDD是一個(gè)漫衍式的不行變數(shù)據(jù)薈萃

Spark 是一個(gè)內(nèi)存處理懲罰引擎

假如你沒有主動(dòng)對(duì)RDDCache/Persist,它不外是一個(gè)觀念上存在的虛擬數(shù)據(jù)集，你實(shí)際上是看不到這個(gè)RDD的數(shù)據(jù)的全集的(他不會(huì)真的都放到內(nèi)存里)。

RDD的本質(zhì)是什么

一個(gè)RDD 本質(zhì)上是一個(gè)函數(shù)，而RDD的調(diào)動(dòng)不外是函數(shù)的嵌套。RDD我認(rèn)為有兩類：

輸入RDD,典范如KafkaRDD,JdbcRDD

轉(zhuǎn)換RDD，如MapPartitionsRDD

我們以下面的代碼為例做闡明：

sc.textFile(“abc.log”).map().saveAsTextFile(“”)

textFile 會(huì)構(gòu)建出一個(gè)NewHadoopRDD,

map函數(shù)運(yùn)行后會(huì)構(gòu)建出一個(gè)MapPartitionsRDD

saveAsTextFile觸發(fā)了實(shí)際流程代碼的執(zhí)行

所以RDD不外是對(duì)一個(gè)函數(shù)的封裝，當(dāng)一個(gè)函數(shù)對(duì)數(shù)據(jù)處理懲罰完成后，我們就獲得一個(gè)RDD的數(shù)據(jù)集(是一個(gè)虛擬的，后續(xù)會(huì)表明)。

NewHadoopRDD是數(shù)據(jù)來歷，每個(gè)parition認(rèn)真獲取數(shù)據(jù)，得到進(jìn)程是通過iterator.next 得到一條一筆記錄的。假設(shè)某個(gè)時(shí)刻拿到了一條數(shù)據(jù)A,這個(gè)A會(huì)立即被map里的函數(shù)處理懲罰獲得B(完成了轉(zhuǎn)換),然后開始寫入到HDFS上。其他數(shù)據(jù)反復(fù)如此。所以整個(gè)進(jìn)程：

理論上某個(gè)MapPartitionsRDD里實(shí)際在內(nèi)存里的數(shù)據(jù)便是其Partition的數(shù)目，是個(gè)很是小的數(shù)值。

NewHadoopRDD則會(huì)略多些，因?yàn)閷儆跀?shù)據(jù)源，讀取文件，假設(shè)讀取文件的buffer是1M，那么最多也就是partitionNum*1M 數(shù)據(jù)在內(nèi)存里

saveAsTextFile也是一樣的，往HDFS寫文件，需要buffer，最大都據(jù)量為 buffer* partitionNum

所以整個(gè)進(jìn)程其實(shí)是流式的進(jìn)程，一條數(shù)據(jù)被各個(gè)RDD所包裹的函數(shù)處理懲罰。

適才我重復(fù)提到了嵌套函數(shù)，怎么知道它是嵌套的呢?

假如你寫了這樣一個(gè)代碼：

sc.textFile(“abc.log”).map().map()………map().saveAsTextFile(“”)

有成千上萬個(gè)map,很大概就倉庫溢出了。為啥?實(shí)際上是函數(shù)嵌套太深了。

按上面的邏輯，內(nèi)存利用其實(shí)長短常小的，10G內(nèi)存跑100T數(shù)據(jù)也不是難事。可是為什么Spark經(jīng)常因?yàn)閮?nèi)存問題掛掉呢? 我們接著往下看。

Shuffle的本質(zhì)是什么?

這就是為什么要分Stage了。每個(gè)Stage其實(shí)就是我上面說的那樣，一套數(shù)據(jù)被N個(gè)嵌套的函數(shù)處理懲罰(也就是你的transform行動(dòng))。碰著了Shuffle,就被切開來，所謂的Shuffle，本質(zhì)上是把數(shù)據(jù)按法則姑且都落到磁盤上，相當(dāng)于完成了一個(gè)saveAsTextFile的行動(dòng)，不外是存當(dāng)?shù)卮疟P。然后被切開的下一個(gè)Stage則以當(dāng)?shù)卮疟P的這些數(shù)據(jù)作為數(shù)據(jù)源，從頭走上面描寫的流程。

我們?cè)僮鲆淮蚊鑼懀?/strong>