添加關(guān)注 作者 ?祝威廉 2016.04.19 23:53*
寫了147997字，被1139人關(guān)注，獲得了540個喜歡
Spark會把數(shù)據(jù)都載入到內(nèi)存么？
字?jǐn)?shù)1239 閱讀3040 評論21 喜歡12
這篇文章算是個科普貼。如果已經(jīng)熟悉Spark的就略過吧。
前言
很多初學(xué)者其實對Spark的編程模式還是RDD這個概念理解不到位，就會產(chǎn)生一些誤解。
比如，很多時候我們常常以為一個文件是會被完整讀入到內(nèi)存，然后做各種變換，這很可能是受兩個概念的誤導(dǎo)：
RDD的定義，RDD是一個分布式的不可變數(shù)據(jù)集合
Spark 是一個內(nèi)存處理引擎
如果你沒有主動對RDDCache/Persist,它不過是一個概念上存在的虛擬數(shù)據(jù)集，你實際上是看不到這個RDD的數(shù)據(jù)的全集的(他不會真的都放到內(nèi)存里)。
RDD的本質(zhì)是什么
一個RDD 本質(zhì)上是一個函數(shù)，而RDD的變換不過是函數(shù)的嵌套。RDD我認(rèn)為有兩類：
輸入RDD,典型如KafkaRDD,JdbcRDD
轉(zhuǎn)換RDD，如MapPartitionsRDD
我們以下面的代碼為例做分析：
sc.textFile("abc.log").map().saveAsTextFile("")
textFile 會構(gòu)建出一個NewHadoopRDD,
map函數(shù)運行后會構(gòu)建出一個MapPartitionsRDD
saveAsTextFile觸發(fā)了實際流程代碼的執(zhí)行
所以RDD不過是對一個函數(shù)的封裝，當(dāng)一個函數(shù)對數(shù)據(jù)處理完成后，我們就得到一個RDD的數(shù)據(jù)集(是一個虛擬的，后續(xù)會解釋)。
NewHadoopRDD是數(shù)據(jù)來源，每個parition負(fù)責(zé)獲取數(shù)據(jù)，獲得過程是通過iterator.next 獲得一條一條記錄的。假設(shè)某個時刻拿到了一條數(shù)據(jù)A,這個A會立刻被map里的函數(shù)處理得到B（完成了轉(zhuǎn)換）,然后開始寫入到HDFS上。其他數(shù)據(jù)重復(fù)如此。所以整個過程：
理論上某個MapPartitionsRDD里實際在內(nèi)存里的數(shù)據(jù)等于其Partition的數(shù)目，是個非常小的數(shù)值。
NewHadoopRDD則會略多些，因為屬于數(shù)據(jù)源，讀取文件，假設(shè)讀取文件的buffer是1M，那么最多也就是partitionNum*1M 數(shù)據(jù)在內(nèi)存里
saveAsTextFile也是一樣的，往HDFS寫文件，需要buffer，最多數(shù)據(jù)量為 buffer* partitionNum
所以整個過程其實是流式的過程，一條數(shù)據(jù)被各個RDD所包裹的函數(shù)處理。
剛才我反復(fù)提到了嵌套函數(shù)，怎么知道它是嵌套的呢？
如果你寫了這樣一個代碼：
sc.textFile("abc.log").map().map().........map().saveAsTextFile("")
有成千上萬個map,很可能就堆棧溢出了。為啥?實際上是函數(shù)嵌套太深了。
按上面的邏輯，內(nèi)存使用其實是非常小的，10G內(nèi)存跑100T數(shù)據(jù)也不是難事。但是為什么Spark常常因為內(nèi)存問題掛掉呢？ 我們接著往下看。
Shuffle的本質(zhì)是什么？
這就是為什么要分Stage了。每個Stage其實就是我上面說的那樣，一套數(shù)據(jù)被N個嵌套的函數(shù)處理(也就是你的transform動作)。遇到了Shuffle,就被切開來，所謂的Shuffle，本質(zhì)上是把數(shù)據(jù)按規(guī)則臨時都落到磁盤上，相當(dāng)于完成了一個saveAsTextFile的動作，不過是存本地磁盤。然后被切開的下一個Stage則以本地磁盤的這些數(shù)據(jù)作為數(shù)據(jù)源，重新走上面描述的流程。
我們再做一次描述：
所謂Shuffle不過是把處理流程切分，給切分的上一段(我們稱為Stage M)加個存儲到磁盤的Action動作，把切分的下一段(Stage M+1)數(shù)據(jù)源變成Stage M存儲的磁盤文件。每個Stage都可以走我上面的描述，讓每條數(shù)據(jù)都可以被N個嵌套的函數(shù)處理，最后通過用戶指定的動作進(jìn)行存儲。
為什么Shuffle 容易導(dǎo)致Spark掛掉
前面我們提到，Shuffle不過是偷偷的幫你加上了個類似saveAsLocalDiskFile的動作。然而，寫磁盤是一個高昂的動作。所以我們盡可能的把數(shù)據(jù)先放到內(nèi)存，再批量寫到文件里，還有讀磁盤文件也是給費內(nèi)存的動作。把數(shù)據(jù)放內(nèi)存，就遇到個問題，比如10000條數(shù)據(jù)，到底會占用多少內(nèi)存？這個其實很難預(yù)估的。所以一不小心，就容易導(dǎo)致內(nèi)存溢出了。這其實也是一個很無奈的事情。
我們做Cache/Persist意味著什么？
其實就是給某個Stage加上了一個saveAsMemoryBlockFile的動作，然后下次再要數(shù)據(jù)的時候，就不用算了。這些存在內(nèi)存的數(shù)據(jù)就表示了某個RDD處理后的結(jié)果。這個才是說為啥Spark是內(nèi)存計算引擎的地方。在MR里，你是要放到HDFS里的，但Spark允許你把中間結(jié)果放內(nèi)存里。
總結(jié)
我們從一個較新的角度解釋了RDD 和Shuffle 都是一個什么樣的東西。
