feat:ChatGPT 第 8 课：cache、persist、checkpoint

2026-09-20 23:52:14 +08:00
parent 2fe83309ce
commit 65ad70ebd7
6 changed files with 306 additions and 0 deletions
@@ -0,0 +1,26 @@
+package org.exercises.exercises8.demo
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+/*
+* 第 8 课：cache、persist、checkpoint*/
+object Demo1Checkpoint {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf().setAppName("Demo1Checkpoint").setMaster("local")
+    val sc = new SparkContext(conf)
+    val rdd = sc.parallelize(1 to 100)
+    sc.setCheckpointDir("src/main/resources/tmp/spark-checkpoint")
+    val result = rdd
+      .map(_ * 10)
+      .filter(_ > 500)
+    println(result.isCheckpointed)
+    println(result.getCheckpointFile)
+    result.checkpoint()
+    result.count()
+    println(result.isCheckpointed)
+    println(result.getCheckpointFile)
+
+    sc.stop()
+
+  }
+}
@@ -0,0 +1,38 @@
+package org.exercises.exercises8.exercises
+
+/*
+* Exercises1：观察 cache 和 persist
+要求
+分别运行下面两种写法：
+版本 A
+val result = rdd.map(_ * 10).cache()
+版本 B
+val result = rdd
+  .map(_ * 10)
+  .persist(StorageLevel.MEMORY_ONLY)
+打印：
+println(result.getStorageLevel)
+并回答：
+为什么两种写法得到的 StorageLevel 基本一致？*/
+
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Exercises1 {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf().setAppName("Exercises1").setMaster("local[*]")
+    val sc = new SparkContext(conf)
+    val rdd = sc.parallelize(List(1, 2, 3, 4, 5))
+    val resultCache = rdd.map(_ * 10).cache()
+    println("=====cache=====")
+    println(resultCache.getStorageLevel)
+    val resultPersist = rdd
+      .map(_ * 10)
+      .persist(StorageLevel.MEMORY_ONLY)
+    println("=====persist=====")
+    println(resultPersist.getStorageLevel)
+    sc.stop()
+
+    // 因为cache 等于 persist的MEMORY_ONLY。def cache(): this.type = persist()，而def persist(): this.type = persist(StorageLevel.MEMORY_ONLY) 默认是MEMORY_ONLY
+  }
+}
@@ -0,0 +1,52 @@
+package org.exercises.exercises8.exercises
+
+/*
+* Exercises2：MEMORY_AND_DISK
+创建：
+val rdd = sc.parallelize(1 to 100, 4)
+
+val result = rdd
+  .map(x => {
+    println(s"计算: $x")
+    x * 10
+  })
+  .persist(StorageLevel.MEMORY_AND_DISK)
+
+println(result.count())
+println(result.collect().take(5).toList)
+
+Thread.sleep(100000)
+观察：
+Storage
+以及 Job / Stage。
+理解：
+persist(MEMORY_AND_DISK)
+和：
+cache()
+相比，多了什么能力。*/
+
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Exercises2 {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf().setAppName("Exercises2").setMaster("local[*]")
+    val sc = new SparkContext(conf)
+    val rdd = sc.parallelize(1 to 100, 4)
+
+    val result = rdd
+      .map(x => {
+        println(s"计算: $x")
+        x * 10
+      })
+      .persist(StorageLevel.DISK_ONLY)
+
+    println(result.count())
+    println(result.collect().take(5).toList)
+
+    Thread.sleep(100000)
+
+    // 看不出来，跟cache一样stage都是显示cache：MapPartitionsRDD [1] [Cached]map at Exercises2.scala:13
+    // DISK_ONLY倒是能看出差别，Storage上Storage Level: Disk Serialized 1x Replicated，cache和MEMORY_AND_DISK都是Memory Deserialized 1x Replicated
+  }
+}
@@ -0,0 +1,61 @@
+package org.exercises.exercises8.exercises
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+/*
+* Exercises3：第一次使用 checkpoint
+这是本节最重要的一题。
+val rdd = sc.parallelize(1 to 10, 2)
+
+val result = rdd
+  .map(_ * 10)
+  .filter(_ > 30)
+
+sc.setCheckpointDir("/tmp/spark-checkpoint")
+
+result.checkpoint()
+
+println("Action 前：")
+println(result.isCheckpointed())
+println(result.getCheckpointFile)
+
+result.count()
+
+println("Action 后：")
+println(result.isCheckpointed())
+println(result.getCheckpointFile)
+
+Thread.sleep(100000)
+重点观察：
+result.isCheckpointed()
+为什么 Action 前后不同。*/
+object Exercises3 {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf().setMaster("local[2]").setAppName("Exercises3")
+    val sc = new SparkContext(conf)
+
+    val rdd = sc.parallelize(1 to 10, 2)
+
+    val result = rdd
+      .map(_ * 10)
+      .filter(_ > 30)
+
+    sc.setCheckpointDir("src/main/resources/tmp/spark-checkpoint")
+
+    result.checkpoint()
+
+    println("Action 前：")
+    println(result.isCheckpointed)
+    println(result.getCheckpointFile)
+
+    result.count()
+
+    println("Action 后：")
+    println(result.isCheckpointed)
+    println(result.getCheckpointFile)
+
+    Thread.sleep(100000)
+
+    // 因为checkpoint也是懒加载，需等待Action触发
+  }
+}
@@ -0,0 +1,63 @@
+package org.exercises.exercises8.exercises
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+/*
+* Exercises4：观察血缘变化
+这题开始真正理解 checkpoint。
+val rdd = sc.parallelize(1 to 10, 2)
+
+val result = rdd
+  .map(_ * 10)
+  .filter(_ > 30)
+  .map(_ + 1)
+  .filter(_ < 100)
+
+sc.setCheckpointDir("/tmp/spark-checkpoint")
+
+println("checkpoint 前：")
+println(result.toDebugString)
+
+result.checkpoint()
+
+result.count()
+
+println("checkpoint 后：")
+println(result.toDebugString)
+
+Thread.sleep(100000)
+你要重点观察：
+result.toDebugString
+然后回答：
+checkpoint 以后，为什么 RDD 的血缘关系发生了变化？
+
+这一题比单纯看 isCheckpointed() 更重要。*/
+object Exercises4 {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf().setMaster("local[2]").setAppName("Exercises4")
+    val sc = new SparkContext(conf)
+    val rdd = sc.parallelize(1 to 10, 2)
+
+    val result = rdd
+      .map(_ * 10)
+      .filter(_ > 30)
+      .map(_ + 1)
+      .filter(_ < 100)
+
+    sc.setCheckpointDir("src/main/resources/tmp/spark-checkpoint")
+
+    println("checkpoint 前：")
+    println(result.toDebugString)
+
+    result.checkpoint()
+
+    result.count()
+
+    println("checkpoint 后：")
+    println(result.toDebugString)
+
+    Thread.sleep(100000)
+
+    // checkpoint 以后，RDD 的血缘关系发生了变化，因为 checkpoint 会将 RDD 的血缘关系截断，只保留到 checkpoint 的 RDD。
+  }
+}
@@ -0,0 +1,66 @@
+package org.exercises.exercises8.exercises
+/*
+* Exercises5：综合题——cache + checkpoint
+写出：
+val rdd = sc.parallelize(1 to 100, 4)
+
+val result = rdd
+  .map(_ * 10)
+  .filter(_ > 500)
+  .map(_ + 1)
+
+result.cache()
+result.checkpoint()
+
+println(result.count())
+println(result.collect().take(5).toList)
+println(result.take(3).toList)
+
+Thread.sleep(100000)
+然后回答这 3 个问题：
+问题 1
+result.cache()
+主要解决什么？
+问题 2
+result.checkpoint()
+主要解决什么？
+问题 3
+为什么两个 Action 之间：
+cache
+有复用效果，但：
+checkpoint
+不是简单地等价于 cache？*/
+import org.apache.spark.SparkConf
+import org.apache.spark.SparkContext
+
+object Exercises5 {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf().setAppName("Exercises5").setMaster("local")
+    val sc = new SparkContext(conf)
+    sc.setCheckpointDir("src/main/resources/tmp/spark-checkpoint")
+    val rdd = sc.parallelize(1 to 100, 4)
+
+    val result = rdd
+      .map(_ * 10)
+      .filter(_ > 500)
+      .map(_ + 1)
+
+    result.cache()
+    result.checkpoint()
+
+    println(result.count())
+    println(result.collect().take(5).toList)
+    println(result.take(3).toList)
+
+    Thread.sleep(100000)
+    sc.stop()
+
+    // 1、cache解决RDD复用
+    // 2、checkpoint解决RDD血缘关系过长
+    // 3cache / persist：
+    //  “我以后还要用，先帮我存着，避免重复计算。”
+    //
+    //  checkpoint：
+    //  “我的血缘太长了，给我留一个可靠的落盘恢复点。”
+  }
+}