feat:ChatGPT 第 7 课：缓存 Cache / Persist + mapPartitions

2026-09-20 22:55:10 +08:00
parent 9606c68b94
commit 2fe83309ce
7 changed files with 381 additions and 0 deletions
@@ -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
+
+
+  }
+}
@@ -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()
+  }
+}
@@ -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)
+  }
+}
@@ -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 能看出缓存了
+  }
+}
@@ -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 负责遍历分区里的元素。
+  }
+}
@@ -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 上进行操作，而不是在每个元素上进行操作。
+  }
+}
@@ -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是针对分区。
+  }
+}