一、問題引出

/**
案例：對同一份數據文件分別做 WordCount 聚合操作和 Word 分組操作
期望：針對數據文件只進行一次分詞、轉換操作得到 RDD 對象，然后再對該對象分別進行聚合和分組，實現數據重用
*/
object TestRDDPersist {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("persist")val sc = new SparkContext(conf)val rdd = sc.makeRDD(List("hello world", "hello spark"))val flatRdd = rdd.flatMap(_.split(" "))val mapRdd = flatRdd.map(word => {println("@@@@@@@@@@")(word, 1)})// 聚合操作val reduceRdd = mapRdd.reduceByKey(_ + _)reduceRdd.collect().foreach(println)println("**********")// 分組操作val groupRdd = mapRdd.groupByKey()groupRdd.collect().foreach(println)}
}/**
結果：flatRdd.map過程在聚合時和分組時分別都執行了，說明針對數據文件的分詞、轉換操作被重復執行了，只有對象被重用，而數據沒有被重用
解析：1.RDD是不會存儲數據的，當某個 RDD 轉換成新的 RDD 后，該 RDD 中的數據就沒有了2.如果需要再次用到該 RDD 的數據，則需要從數據源開始重新執行到該 RDD 來獲取數據
解決：針對某個需要被重復使用的 RDD 對象在其進行下一步操作時先將數據進行緩存持久化或checkpoint，后續的其它操作從緩存持久化或checkpoint中獲取數據
*/

二、RDD Cache

/**
緩存或持久化方法：1.rdd.cache()：底層調用 persist() 方法，默認是將數據保存到 JVM 堆內存中2.rdd.persist(StorageLevel)：可以指定數據的保存級別
說明：1.持久化方法被調用時不會立即進行緩存，而是在觸發action算子時，數據才會被緩存在計算節點的內存中2.緩存除了用于數據重用，還可以提高容錯性
*/
object TestRDDPersist {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("persist")val sc = new SparkContext(conf)val rdd = sc.makeRDD(List("hello world", "hello spark"))val flatRdd = rdd.flatMap(_.split(" "))val mapRdd = flatRdd.map(word => {println("@@@@@@@@@@")(word, 1)})//mapRdd.cache()mapRdd.persist()// 聚合操作val reduceRdd = mapRdd.reduceByKey(_ + _)reduceRdd.collect().foreach(println)println("**********")// 分組操作val groupRdd = mapRdd.groupByKey()groupRdd.collect().foreach(println)/*結果：聚合和分組前的操作過程只執行了一遍，實現了數據重用*/}
}// 存儲級別
object StorageLevel {val NONE = new StorageLevel(false, false, false, false)val DISK_ONLY = new StorageLevel(true, false, false, false)val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2) // 副本val MEMORY_ONLY = new StorageLevel(false, true, false, true) // 內存不足丟棄數據val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)val MEMORY_AND_DISK = new StorageLevel(true, true, false, true) // 內存不足溢寫磁盤val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
}

三、RDD CheckPoint

/**
方法：rdd.checkpoint()，將 RDD 中間結果寫入磁盤
說明：1.對 RDD 進行 checkpoint 操作并不會馬上被執行，必須執行 Action 操作才能觸發2.checkpoint保存由于在job執行完不會被刪除，所以必須指定保存路徑，一般保存在分布式文件系統
*/
object TestRDDPersist {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("persist")val sc = new SparkContext(conf)// 指定checkpoint保存路徑sc.setCheckpointDir("checkpoint")val rdd = sc.makeRDD(List("hello world", "hello spark"))val flatRdd = rdd.flatMap(_.split(" "))val mapRdd = flatRdd.map(word => {println("@@@@@@@@@@")(word, 1)})mapRdd.checkpoint()// 聚合操作val reduceRdd = mapRdd.reduceByKey(_ + _)reduceRdd.collect().foreach(println)println("**********")// 分組操作val groupRdd = mapRdd.groupByKey()groupRdd.collect().foreach(println)/*結果：聚合和分組前的操作過程只執行了一遍，實現了數據重用*/}
}

四、緩存和檢查點區別

• cache 和 persist 會在原有的血緣關系中添加新的依賴，一旦數據出錯可以重頭讀取數據；checkpoint 檢查點會切斷原有的血緣關系，重新建立新的血緣關系，相當于改變數據源
• cache 是將數據臨時存儲在 JVM 堆內存中，性能較高，但安全性低，persist 可以指定存儲級別，將數據臨時存儲在磁盤文件中，涉及到 IO，性能較低，作業執行完畢后臨時文件會被刪除；checkpoint 是將數據長久地存儲分布式文件系統中，安全性較高，但涉及 IO 且會獨立開啟一個作業從數據源開始獲取數據，所以性能較低，一般在 checkpoint 前先進行 cache，當 checkpoint 時 job 只需從緩存中讀取數據即可，可以提高性能