Spark可以將RDD持久化到內(nèi)存或者磁盤(pán)，持久化到內(nèi)存可以極大的提高迭代計(jì)算以及計(jì)算模型之間的數(shù)據(jù)共享，一般情況下，執(zhí)行節(jié)點(diǎn)60%內(nèi)存用于緩存數(shù)據(jù)，剩下40%用于運(yùn)行任務(wù)。Spark使用persist、cache進(jìn)行操作持久化，其中cache是persist的特例。

• cache():RDD[T]
• persist():RDD[T]
• persist(level:StorageLevel):RDD[T]

1.?什么情況下需要對(duì)數(shù)據(jù)進(jìn)行持久化

某步驟計(jì)算特別耗時(shí)，重新計(jì)算的代價(jià)較高，所以進(jìn)行持久化

計(jì)算鏈條特別長(zhǎng)的情況下，重新計(jì)算的代價(jià)也較高

checkpoint所在的RDD也一定要持久化數(shù)據(jù)，checkpoint是lazy的，框架本身會(huì)對(duì)checkpoint的RDD觸發(fā)新的job，不進(jìn)行persist的話，進(jìn)行checkpoint的時(shí)候數(shù)據(jù)就會(huì)重新計(jì)算一遍，所以checkpoint之前一定要進(jìn)行 persist，因?yàn)樵赾heckpoint前有了persist的前提下，計(jì)算過(guò)一遍之后，再進(jìn)行計(jì)算的時(shí)候計(jì)算速度非常快

shuffle操作之后，因?yàn)閟huffle要進(jìn)行網(wǎng)絡(luò)傳輸，網(wǎng)絡(luò)傳輸風(fēng)險(xiǎn)大，數(shù)據(jù)極易丟失，所以shuffle之前進(jìn)行persist避免數(shù)據(jù)丟失

shuffle操作之前，框架默認(rèn)幫助我們把數(shù)據(jù)持久化到本地磁盤(pán)，該步驟由框架自動(dòng)完成

2. 函數(shù)接口的使用

• persist()

//persist():RDD[] def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

默認(rèn)使用MEMORY_ONLY這個(gè)緩存級(jí)別

persist(level:StorageLevel):RDD[T]

根據(jù)不用的緩存級(jí)別對(duì)RDD做不同的緩存操作

• cache()

def cache(): this.type = persist()

cache()方法使用了默認(rèn)的存儲(chǔ)級(jí)別—StorageLevel.MEMORY_ONLY將RDD緩存在內(nèi)存中

• unpersist()

清除緩存操作

def unpersist(blocking: Boolean = true): this.type = {logInfo("Removing RDD " + id + " from persistence list")sc.unpersistRDD(id, blocking)storageLevel = StorageLevel.NONEthis }

?

總結(jié)

以上是生活随笔為你收集整理的[scala-spark]11. RDD控制操作的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。