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

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

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

某步驟計算特別耗時，重新計算的代價較高，所以進行持久化

計算鏈條特別長的情況下，重新計算的代價也較高

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

shuffle操作之后，因為shuffle要進行網(wǎng)絡(luò)傳輸，網(wǎng)絡(luò)傳輸風險大，數(shù)據(jù)極易丟失，所以shuffle之前進行persist避免數(shù)據(jù)丟失

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

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

• persist()

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

默認使用MEMORY_ONLY這個緩存級別

persist(level:StorageLevel):RDD[T]

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

• cache()

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

cache()方法使用了默認的存儲級別—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控制操作的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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