Spark RDD 操作

RDDs 支持 2 種類(lèi)型的操作：轉(zhuǎn)換(transformations) 從已經(jīng)存在的數(shù)據(jù)集中創(chuàng)建一個(gè)新的數(shù)據(jù)集；動(dòng)作(actions) 在數(shù)據(jù)集上進(jìn)行計(jì)算之后返回一個(gè)值到驅(qū)動(dòng)程序。例如，map 是一個(gè)轉(zhuǎn)換操作，它將每一個(gè)數(shù)據(jù)集元素傳遞給一個(gè)函數(shù)并且返回一個(gè)新的 RDD。另一方面，reduce 是一個(gè)動(dòng)作，它使用相同的函數(shù)來(lái)聚合 RDD 的所有元素，并且將最終的結(jié)果返回到驅(qū)動(dòng)程序(不過(guò)也有一個(gè)并行 reduceByKey 能返回一個(gè)分布式數(shù)據(jù)集)。

在 Spark 中，所有的轉(zhuǎn)換(transformations)都是惰性(lazy)的，它們不會(huì)馬上計(jì)算它們的結(jié)果。相反的，它們僅僅記錄轉(zhuǎn)換操作是應(yīng)用到哪些基礎(chǔ)數(shù)據(jù)集(例如一個(gè)文件)上的。轉(zhuǎn)換僅僅在這個(gè)時(shí)候計(jì)算：當(dāng)動(dòng)作(action) 需要一個(gè)結(jié)果返回給驅(qū)動(dòng)程序的時(shí)候。這個(gè)設(shè)計(jì)能夠讓 Spark 運(yùn)行得更加高效。例如，我們可以實(shí)現(xiàn)：通過(guò) map 創(chuàng)建一個(gè)新數(shù)據(jù)集在 reduce 中使用，并且僅僅返回 reduce 的結(jié)果給 driver，而不是整個(gè)大的映射過(guò)的數(shù)據(jù)集。

默認(rèn)情況下，每一個(gè)轉(zhuǎn)換過(guò)的 RDD 會(huì)在每次執(zhí)行動(dòng)作(action)的時(shí)候重新計(jì)算一次。然而，你也可以使用 persist (或 cache)方法持久化(persist)一個(gè) RDD 到內(nèi)存中。在這個(gè)情況下，Spark 會(huì)在集群上保存相關(guān)的元素，在你下次查詢(xún)的時(shí)候會(huì)變得更快。在這里也同樣支持持久化 RDD 到磁盤(pán)，或在多個(gè)節(jié)點(diǎn)間復(fù)制。

基礎(chǔ)

為了說(shuō)明 RDD 基本知識(shí)，考慮下面的簡(jiǎn)單程序：

val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)

第一行是定義來(lái)自于外部文件的 RDD。這個(gè)數(shù)據(jù)集并沒(méi)有加載到內(nèi)存或做其他的操作：lines 僅僅是一個(gè)指向文件的指針。第二行是定義 lineLengths，它是 map 轉(zhuǎn)換(transformation)的結(jié)果。同樣，lineLengths 由于懶惰模式也沒(méi)有立即計(jì)算。最后，我們執(zhí)行 reduce，它是一個(gè)動(dòng)作(action)。在這個(gè)地方，Spark 把計(jì)算分成多個(gè)任務(wù)(task)，并且讓它們運(yùn)行在多個(gè)機(jī)器上。每臺(tái)機(jī)器都運(yùn)行自己的 map 部分和本地 reduce 部分。然后僅僅將結(jié)果返回給驅(qū)動(dòng)程序。

如果我們想要再次使用 lineLengths，我們可以添加：

lineLengths.persist()

在 reduce 之前，它會(huì)導(dǎo)致 lineLengths 在第一次計(jì)算完成之后保存到內(nèi)存中。