大數據開發中Spark-RDD的持久化和緩存該如何實現

大數據開發中Spark-RDD的持久化和緩存該如何實現，相信很多沒有經驗的人對此束手無策，為此本文總結了問題出現的原因和解決方法，通過這篇文章希望你能解決這個問題。

1.RDD緩存機制 cache, persist

Spark 速度非常快的一個原因是 RDD 支持緩存。成功緩存后，如果之后的操作使用到了該數據集，則直接從緩存中獲取。雖然緩存也有丟失的風險，但是由于 RDD 之間的依賴關系，如果某個分區的緩存數據丟失，只需要重新計算該分區即可。

涉及到的算子：persist、cache、unpersist；都是 Transformation

緩存是將計算結果寫入不同的介質，用戶定義可定義存儲級別（存儲級別定義了緩存存儲的介質，目前支持內存、堆 外內存、磁盤）；

通過緩存，Spark避免了RDD上的重復計算，能夠極大地提升計算速度； RDD持久化或緩存，是Spark最重要的特征之一。可以說，緩存是Spark構建迭代式算法和快速交互式查詢的關鍵因 素；

Spark速度非常快的原因之一，就是在內存中持久化（或緩存）一個數據集。當持久化一個RDD后，每一個節點都將 把計算的分片結果保存在內存中，并在對此數據集（或者衍生出的數據集）進行的其他動作（Action）中重用。這使 得后續的動作變得更加迅速；使用persist()方法對一個RDD標記為持久化。之所以說“標記為持久化”，是因為出現persist()語句的地方，并不會馬 上計算生成RDD并把它持久化，而是要等到遇到第一個行動操作觸發真正計算以后，才會把計算結果進行持久化；通過persist()或cache()方法可以標記一個要被持久化的RDD，持久化被觸發，RDD將會被保留在計算節點的內存中 并重用；

什么時候緩存數據，需要對空間和速度進行權衡。一般情況下，如果多個動作需要用到某個 RDD，而它的計算代價 又很高，那么就應該把這個 RDD 緩存起來；

緩存有可能丟失，或者存儲于內存的數據由于內存不足而被刪除。RDD的緩存的容錯機制保證了即使緩存丟失也能保 證計算的正確執行。通過基于RDD的一系列的轉換，丟失的數據會被重算。RDD的各個Partition是相對獨立的，因此 只需要計算丟失的部分即可，并不需要重算全部Partition。

啟動堆外內存需要配置兩個參數：

• spark.memory.offHeap.enabled ：是否開啟堆外內存，默認值為 false，需要設置為 true；

• spark.memory.offHeap.size : 堆外內存空間的大小，默認值為 0，需要設置為正值。

1.1 緩存級別

Spark 速度非常快的一個原因是 RDD 支持緩存。成功緩存后，如果之后的操作使用到了該數據集，則直接從緩存中獲取。雖然緩存也有丟失的風險，但是由于 RDD 之間的依賴關系，如果某個分區的緩存數據丟失，只需要重新計算該分區即可。

Spark 支持多種緩存級別 ：

啟動堆外內存需要配置兩個參數：

• spark.memory.offHeap.enabled ：是否開啟堆外內存，默認值為 false，需要設置為 true；

• spark.memory.offHeap.size : 堆外內存空間的大小，默認值為 0，需要設置為正值。

1.2 使用緩存

緩存數據的方法有兩個：persist 和 cache 。cache 內部調用的也是 persist，它是 persist 的特殊化形式，等價于 persist(StorageLevel.MEMORY_ONLY)。示例如下：

// 所有存儲級別均定義在 StorageLevel 對象中
fileRDD.persist(StorageLevel.MEMORY_AND_DISK)
fileRDD.cache()

被緩存的RDD在DAG圖中有一個綠色的圓點。

1.3 移除緩存

Spark 會自動監視每個節點上的緩存使用情況，并按照最近最少使用（LRU）的規則刪除舊數據分區。當然，你也可以使用 RDD.unpersist() 方法進行手動刪除。

2.RDD容錯機制Checkpoint

2.1 涉及到的算子：checkpoint；也是 Transformation

Spark中對于數據的保存除了持久化操作之外，還提供了檢查點的機制；檢查點本質是通過將RDD寫入高可靠的磁盤，主要目的是為了容錯。檢查點通過將數據寫入到HDFS文件系統實現了

RDD的檢查點功能。Lineage過長會造成容錯成本過高，這樣就不如在中間階段做檢查點容錯，如果之后有節點出現問題而丟失分區，從

做檢查點的RDD開始重做Lineage，就會減少開銷。

2.2 cache 和 checkpoint 區別

cache 和 checkpoint 是有顯著區別的，緩存把 RDD 計算出來然后放在內存中，但是 RDD 的依賴鏈不能丟掉， 當某個點某個 executor 宕了，上面 cache 的RDD就會丟掉， 需要通過依賴鏈重放計算。不同的是，checkpoint 是把

RDD 保存在 HDFS中，是多副本可靠存儲，此時依賴鏈可以丟掉，所以斬斷了依賴鏈。

2.3 checkpoint適合場景

以下場景適合使用檢查點機制：

1. DAG中的Lineage過長，如果重算，則開銷太大

2. 在寬依賴上做 Checkpoint 獲得的收益更大

與cache類似 checkpoint 也是 lazy 的。

val rdd1 = sc.parallelize(1 to 100000)
// 設置檢查點目錄

sc.setCheckpointDir("/tmp/checkpoint")

val rdd2 = rdd1.map(_*2)

rdd2.checkpoint

// checkpoint是lazy操作

rdd2.isCheckpointed

// checkpoint之前的rdd依賴關系

rdd2.dependencies(0).rdd

rdd2.dependencies(0).rdd.collect

// 執行一次action，觸發checkpoint的執行

rdd2.count

rdd2.isCheckpointed

// 再次查看RDD的依賴關系。可以看到checkpoint后，RDD的lineage被截斷，變成從checkpointRDD開始

rdd2.dependencies(0).rdd

rdd2.dependencies(0).rdd.collect

//查看RDD所依賴的checkpoint文件

rdd2.getCheckpointFile

看完上述內容，你們掌握大數據開發中Spark-RDD的持久化和緩存該如何實現的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！