From 65ad70ebd7566dac745ec3d471f47dceeef79453 Mon Sep 17 00:00:00 2001 From: kennethcheng Date: Sun, 20 Sep 2026 23:52:14 +0800 Subject: [PATCH] =?UTF-8?q?feat:ChatGPT=20=E7=AC=AC=208=20=E8=AF=BE?= =?UTF-8?q?=EF=BC=9Acache=E3=80=81persist=E3=80=81checkpoint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../exercises8/demo/Demo1Checkpoint.scala | 26 ++++++++ .../exercises8/exercises/Exercises1.scala | 38 +++++++++++ .../exercises8/exercises/Exercises2.scala | 52 +++++++++++++++ .../exercises8/exercises/Exercises3.scala | 61 +++++++++++++++++ .../exercises8/exercises/Exercises4.scala | 63 ++++++++++++++++++ .../exercises8/exercises/Exercises5.scala | 66 +++++++++++++++++++ 6 files changed, 306 insertions(+) create mode 100644 src/main/scala/org/exercises/exercises8/demo/Demo1Checkpoint.scala create mode 100644 src/main/scala/org/exercises/exercises8/exercises/Exercises1.scala create mode 100644 src/main/scala/org/exercises/exercises8/exercises/Exercises2.scala create mode 100644 src/main/scala/org/exercises/exercises8/exercises/Exercises3.scala create mode 100644 src/main/scala/org/exercises/exercises8/exercises/Exercises4.scala create mode 100644 src/main/scala/org/exercises/exercises8/exercises/Exercises5.scala diff --git a/src/main/scala/org/exercises/exercises8/demo/Demo1Checkpoint.scala b/src/main/scala/org/exercises/exercises8/demo/Demo1Checkpoint.scala new file mode 100644 index 0000000..fee3551 --- /dev/null +++ b/src/main/scala/org/exercises/exercises8/demo/Demo1Checkpoint.scala @@ -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() + + } +} diff --git a/src/main/scala/org/exercises/exercises8/exercises/Exercises1.scala b/src/main/scala/org/exercises/exercises8/exercises/Exercises1.scala new file mode 100644 index 0000000..e4af9c1 --- /dev/null +++ b/src/main/scala/org/exercises/exercises8/exercises/Exercises1.scala @@ -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 + } +} diff --git a/src/main/scala/org/exercises/exercises8/exercises/Exercises2.scala b/src/main/scala/org/exercises/exercises8/exercises/Exercises2.scala new file mode 100644 index 0000000..7d1c22a --- /dev/null +++ b/src/main/scala/org/exercises/exercises8/exercises/Exercises2.scala @@ -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 + } +} diff --git a/src/main/scala/org/exercises/exercises8/exercises/Exercises3.scala b/src/main/scala/org/exercises/exercises8/exercises/Exercises3.scala new file mode 100644 index 0000000..07267ac --- /dev/null +++ b/src/main/scala/org/exercises/exercises8/exercises/Exercises3.scala @@ -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触发 + } +} diff --git a/src/main/scala/org/exercises/exercises8/exercises/Exercises4.scala b/src/main/scala/org/exercises/exercises8/exercises/Exercises4.scala new file mode 100644 index 0000000..d806815 --- /dev/null +++ b/src/main/scala/org/exercises/exercises8/exercises/Exercises4.scala @@ -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。 + } +} diff --git a/src/main/scala/org/exercises/exercises8/exercises/Exercises5.scala b/src/main/scala/org/exercises/exercises8/exercises/Exercises5.scala new file mode 100644 index 0000000..6b0fb47 --- /dev/null +++ b/src/main/scala/org/exercises/exercises8/exercises/Exercises5.scala @@ -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： + // “我的血缘太长了，给我留一个可靠的落盘恢复点。” + } +}