From 2fe83309ce3021d23adcabea98ba97b5ae435012 Mon Sep 17 00:00:00 2001 From: kennethcheng Date: Sun, 20 Sep 2026 22:55:10 +0800 Subject: [PATCH] =?UTF-8?q?feat:ChatGPT=20=E7=AC=AC=207=20=E8=AF=BE?= =?UTF-8?q?=EF=BC=9A=E7=BC=93=E5=AD=98=20Cache=20/=20Persist=20+=20mapPart?= =?UTF-8?q?itions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../demo/Demo1CacheAndPersist.scala | 67 ++++++++++++++ .../exercises7/demo/Demo2MapPartitions.scala | 39 ++++++++ .../exercises7/exercises/Exercises1.scala | 89 +++++++++++++++++++ .../exercises7/exercises/Exercises2.scala | 51 +++++++++++ .../exercises7/exercises/Exercises3.scala | 40 +++++++++ .../exercises7/exercises/Exercises4.scala | 45 ++++++++++ .../exercises7/exercises/Exercises5.scala | 50 +++++++++++ 7 files changed, 381 insertions(+) create mode 100644 src/main/scala/org/exercises/exercises7/demo/Demo1CacheAndPersist.scala create mode 100644 src/main/scala/org/exercises/exercises7/demo/Demo2MapPartitions.scala create mode 100644 src/main/scala/org/exercises/exercises7/exercises/Exercises1.scala create mode 100644 src/main/scala/org/exercises/exercises7/exercises/Exercises2.scala create mode 100644 src/main/scala/org/exercises/exercises7/exercises/Exercises3.scala create mode 100644 src/main/scala/org/exercises/exercises7/exercises/Exercises4.scala create mode 100644 src/main/scala/org/exercises/exercises7/exercises/Exercises5.scala diff --git a/src/main/scala/org/exercises/exercises7/demo/Demo1CacheAndPersist.scala b/src/main/scala/org/exercises/exercises7/demo/Demo1CacheAndPersist.scala new file mode 100644 index 0000000..0037bcd --- /dev/null +++ b/src/main/scala/org/exercises/exercises7/demo/Demo1CacheAndPersist.scala @@ -0,0 +1,67 @@ +package org.exercises.exercises7.demo + +import org.apache.spark.storage.StorageLevel +import org.apache.spark.{SparkConf, SparkContext} + + +/* +* 第 7 课：缓存 Cache / Persist + mapPartitions*/ +object Demo1CacheAndPersist { + def main(args: Array[String]): Unit = { + val conf = new SparkConf().setAppName("Demo1CacheAndPersist").setMaster("local[2]") + val sc = new SparkContext(conf) + + val rdd = sc.parallelize(1 to 10000, 4) + + val rdd2 = rdd + .map { x => + Thread.sleep(1) + x * 10 + } + .filter(_ > 20) + // .cache() + .persist(StorageLevel.DISK_ONLY) + + val t1 = System.currentTimeMillis() + println("第一次：" + rdd2.count()) + println("耗时：" + (System.currentTimeMillis() - t1) + " ms") + + val t2 = System.currentTimeMillis() + println("第二次：" + rdd2.count()) + println("耗时：" + (System.currentTimeMillis() - t2) + " ms") + + sc.stop() + + // 没有cache + // 第一次：9998 + // 耗时：7542 ms + // 第二次：9998 + // 耗时：7501 ms + + // 有cache + // 第一次：9998 + // 耗时：7791 ms + // 第二次：9998 + // 耗时：14 ms + + // persist(StorageLevel.MEMORY_ONLY) + // 第一次：9998 + // 耗时：7766 ms + // 第二次：9998 + // 耗时：16 ms + + // persist(StorageLevel.MEMORY_AND_DISK) + // 第一次：9998 + // 耗时：7813 ms + // 第二次：9998 + // 耗时：14 ms + + // persist(StorageLevel.DISK_ONLY) + // 第一次：9998 + // 耗时：7836 ms + // 第二次：9998 + // 耗时：23 ms + + + } +} diff --git a/src/main/scala/org/exercises/exercises7/demo/Demo2MapPartitions.scala b/src/main/scala/org/exercises/exercises7/demo/Demo2MapPartitions.scala new file mode 100644 index 0000000..f00e933 --- /dev/null +++ b/src/main/scala/org/exercises/exercises7/demo/Demo2MapPartitions.scala @@ -0,0 +1,39 @@ +package org.exercises.exercises7.demo + +/* +* mapPartitions 一次处理一个 Partition +* 如果一个 Partition 有 10000 条数据 +* 元素1 → 创建资源 → 处理 +* 元素2 → 创建资源 → 处理 +* 元素3 → 创建资源 → 处理 +* +* 如果这个资源很昂贵，就很浪费。 +* 用mapPartitions +* Partition +* ↓ +* 创建一次资源 +* ↓ +* 处理这个 Partition 的 10000 条数据 +* ↓ +* 关闭资源*/ + +import org.apache.spark.{SparkConf, SparkContext} + +object Demo2MapPartitions { + def main(args: Array[String]): Unit = { + val conf = new SparkConf().setAppName("Demo2MapPartitions").setMaster("local[2]") + val sc = new SparkContext(conf) + + val rdd = sc.parallelize(1 to 10, 2) + rdd.glom().zipWithIndex().foreach { + case (c, i) => println(s"Partition $i: ${c.toList}") + } + val result = rdd.mapPartitions(iter => { + iter.map(_ * 10) + }) + result.glom().zipWithIndex().foreach { + case (c, i) => println(s"Partition $i: ${c.toList}") + } + sc.stop() + } +} diff --git a/src/main/scala/org/exercises/exercises7/exercises/Exercises1.scala b/src/main/scala/org/exercises/exercises7/exercises/Exercises1.scala new file mode 100644 index 0000000..201e406 --- /dev/null +++ b/src/main/scala/org/exercises/exercises7/exercises/Exercises1.scala @@ -0,0 +1,89 @@ +package org.exercises.exercises7.exercises + +/* +* Exercises1：观察 Cache +创建： +val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 3 +) + +val result = rdd + .map(x => { + println(s"正在计算: $x") + x * 10 + }) + .filter(_ > 20) +执行： +println(result.count()) +println(result.collect().toList) +要求 +先不要加 cache()。 +观察： +map 中的 println +大概执行了几轮。 +然后改成： +val result = rdd + .map(x => { + println(s"正在计算: $x") + x * 10 + }) + .filter(_ > 20) + .cache() +再次： +println(result.count()) +println(result.collect().toList) +比较两次执行情况。 +注意：Executor 日志的显示位置和顺序可能与你预期不同，所以这个练习重点是理解是否发生重新计算，不要死磕 println 的输出顺序。*/ + +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, 6), + 3 + ) + + val result = rdd + .map(x => { + println(s"正在计算: $x") + x * 10 + }) + .filter(_ > 20) + .cache() + + println(result.count()) + println(result.collect().toList) + + sc.stop() + // 不加cache + // 正在计算: 1 + // 正在计算: 2 + // 正在计算: 3 + // 正在计算: 4 + // 正在计算: 5 + // 正在计算: 6 + // 4 + // 正在计算: 1 + // 正在计算: 2 + // 正在计算: 3 + // 正在计算: 4 + // 正在计算: 5 + // 正在计算: 6 + // List(30, 40, 50, 60) + + // 加了cache + // 正在计算: 1 + // 正在计算: 2 + // 正在计算: 3 + // 正在计算: 4 + // 正在计算: 5 + // 正在计算: 6 + // 4 + // List(30, 40, 50, 60) + } +} diff --git a/src/main/scala/org/exercises/exercises7/exercises/Exercises2.scala b/src/main/scala/org/exercises/exercises7/exercises/Exercises2.scala new file mode 100644 index 0000000..eab8025 --- /dev/null +++ b/src/main/scala/org/exercises/exercises7/exercises/Exercises2.scala @@ -0,0 +1,51 @@ +package org.exercises.exercises7.exercises + +/* +* Exercises2：比较 Cache 前后 +创建： +val rdd = sc.parallelize( + 1 to 100, + 4 +) +执行： +val result = rdd + .map(_ * 10) + .filter(_ > 500) +然后连续执行： +result.count() +result.collect() +result.take(5) +要求： +1. 第一次不使用 cache() +2. 第二次加上 cache() +3. 打开 Spark UI +4. 对比 Jobs / Stages +重点思考： +为什么加 Cache 后，后面的 Action 可以复用之前的计算结果？*/ + +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(_ * 10) + .filter(_ > 500) + .cache() + + result.count() + result.collect() + result.take(5) + + Thread.sleep(100000) + sc.stop() + + // MapPartitionsRDD [2] [Cached]filter at Exercises2.scala:3，Stage 能看出缓存了 + } +} diff --git a/src/main/scala/org/exercises/exercises7/exercises/Exercises3.scala b/src/main/scala/org/exercises/exercises7/exercises/Exercises3.scala new file mode 100644 index 0000000..147b8d2 --- /dev/null +++ b/src/main/scala/org/exercises/exercises7/exercises/Exercises3.scala @@ -0,0 +1,40 @@ +package org.exercises.exercises7.exercises + +/* +* Exercises3：练习 mapPartitions +创建： +val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 2 +) +使用： +mapPartitions +要求： +每个元素 × 10。 + +但不能使用 map。 +提示： +val result = rdd.mapPartitions(iter => { + iter.map(...) +}) +然后： +println(result.collect().toList)*/ + +import org.apache.spark.{SparkConf, SparkContext} + +object Exercises3 { + def main(args: Array[String]): Unit = { + val conf = new SparkConf().setAppName("Exercises3").setMaster("local[2]") + val sc = new SparkContext(conf) + + val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 2 + ) + + val res = rdd.mapPartitions(_.map(_ * 10)) + println(res.collect().toList) + sc.stop() + //Spark 用 mapPartitions 控制执行粒度，Scala Iterator.map 负责遍历分区里的元素。 + } +} diff --git a/src/main/scala/org/exercises/exercises7/exercises/Exercises4.scala b/src/main/scala/org/exercises/exercises7/exercises/Exercises4.scala new file mode 100644 index 0000000..e74c734 --- /dev/null +++ b/src/main/scala/org/exercises/exercises7/exercises/Exercises4.scala @@ -0,0 +1,45 @@ +package org.exercises.exercises7.exercises + +/* +* Exercises4：观察 Partition 级处理 +创建： +val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 3 +) +使用： +mapPartitions +打印： +开始处理一个 Partition +然后返回原来的数据。 +例如思路： +val result = rdd.mapPartitions(iter => { + println("开始处理一个 Partition") + iter +}) +执行： +result.collect() +思考： +为什么 println("开始处理一个 Partition") 不是打印 6 次，而通常与 Partition 数量相关？*/ + +import org.apache.spark.{SparkConf, SparkContext} + +object Exercises4 { + def main(args: Array[String]): Unit = { + val conf = new SparkConf().setAppName("Exercises4").setMaster("local") + val sc = new SparkContext(conf) + + val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 3 + ) + + val result = rdd.mapPartitions(iter => { + println("开始处理一个 Partition") + iter + }).collect() + + sc.stop() + // 因为 mapPartitions 是在每个 Partition 上进行操作，而不是在每个元素上进行操作。 + } +} diff --git a/src/main/scala/org/exercises/exercises7/exercises/Exercises5.scala b/src/main/scala/org/exercises/exercises7/exercises/Exercises5.scala new file mode 100644 index 0000000..c314362 --- /dev/null +++ b/src/main/scala/org/exercises/exercises7/exercises/Exercises5.scala @@ -0,0 +1,50 @@ +package org.exercises.exercises7.exercises + +/* +* Exercises5：foreach vs foreachPartition +创建： +val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 3 +) +分别执行： +第一种 +rdd.foreach(x => { + println(s"foreach: $x") +}) +第二种 +rdd.foreachPartition(iter => { + println("开始处理 Partition") + iter.foreach(x => println(s"foreachPartition: $x")) +}) +然后思考： +foreach +→ 为什么打印 6 个元素？ + +foreachPartition +→ 为什么“开始处理 Partition”通常打印 3 次？*/ + +import org.apache.spark.{SparkConf, SparkContext} + +object Exercises5 { + def main(args: Array[String]): Unit = { + val conf = new SparkConf().setAppName("Exercises5").setMaster("local") + val sc = new SparkContext(conf) + val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 3 + ) + rdd.foreach(x => { + println(s"foreach: $x") + }) + + rdd.foreachPartition(iter => { + println("开始处理 Partition") + iter.foreach(x => println(s"foreachPartition: $x")) + }) + + sc.stop() + + // 因为 foreach是针对元素，foreachPartition是针对分区。 + } +}