From 3971b83a58569f1d1bc3f3d1459bf60b20394199 Mon Sep 17 00:00:00 2001 From: KennethCheng Date: Thu, 17 Sep 2026 18:44:26 +0800 Subject: [PATCH] =?UTF-8?q?feat:=E7=BA=BF=E4=B8=8B=E8=AF=BE=EF=BC=9A?= =?UTF-8?q?=E6=8C=81=E4=B9=85=E5=8C=96=E5=92=8C=E7=BC=93=E5=AD=98=E3=80=81?= =?UTF-8?q?Checkpoint=E3=80=81=E6=89=93=E5=8C=85=E7=A8=8B=E5=BA=8F?= =?UTF-8?q?=E6=8F=90=E4=BA=A4=E5=88=B0yarm?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/main/scala/org/example/App2.scala | 44 ++++++++++++++++ .../scala/org/example/core/CacheTest.scala | 41 +++++++++++++++ .../org/example/core/CheckpointDemo.scala | 30 +++++++++++ .../scala/org/example/core/DemoCache.scala | 50 +++++++++++++++++++ .../org/example/core/DemoCacheNoCache.scala | 45 +++++++++++++++++ .../org/example/core/ReadFromCheckpoint.scala | 28 +++++++++++ 6 files changed, 238 insertions(+) create mode 100644 src/main/scala/org/example/App2.scala create mode 100644 src/main/scala/org/example/core/CacheTest.scala create mode 100644 src/main/scala/org/example/core/CheckpointDemo.scala create mode 100644 src/main/scala/org/example/core/DemoCache.scala create mode 100644 src/main/scala/org/example/core/DemoCacheNoCache.scala create mode 100644 src/main/scala/org/example/core/ReadFromCheckpoint.scala diff --git a/src/main/scala/org/example/App2.scala b/src/main/scala/org/example/App2.scala new file mode 100644 index 0000000..219fbe5 --- /dev/null +++ b/src/main/scala/org/example/App2.scala @@ -0,0 +1,44 @@ +package org.example + +/* +* /home/hadoop/app/spark-2.4.3/bin/spark-submit \ + --class org.example.core.App2 \ + --master yarn \ + --deploy-mode client \ + --executor-memory 512M \ + --num-executors 1 \ + --executor-cores 2 \ + spark_project-1.0-SNAPSHOT.jar \ + /word.txt \ + /output/app2_result + * */ +import org.apache.spark.{SparkConf, SparkContext} + +object App2 { + def main(args: Array[String]): Unit = { + + val conf = new SparkConf() + // 提交到集群的话，不需要设置 local + // .setMaster("local") + .setAppName("my spark app") + + val sc = new SparkContext(conf) + + // 先当成集合来看 + val myTextFile = sc.textFile(args(0)) + + println("myTextFile.count() = " + myTextFile.count()) + + val result = + myTextFile + .flatMap(line => line.split(" ")) + .map(word => (word, 1)) + .reduceByKey((a, b) => a + b) + + result.collect().foreach(println) + + result.saveAsTextFile(args(1)) + + sc.stop() + } +} diff --git a/src/main/scala/org/example/core/CacheTest.scala b/src/main/scala/org/example/core/CacheTest.scala new file mode 100644 index 0000000..b184dbd --- /dev/null +++ b/src/main/scala/org/example/core/CacheTest.scala @@ -0,0 +1,41 @@ +package org.example.core + +import org.apache.spark.storage.StorageLevel +import org.apache.spark.{SparkConf, SparkContext} + +object CacheTest { + def main(args: Array[String]): Unit = { + + val sparkConf = new SparkConf() + .setMaster("local[2]") + .setAppName("scala spark app") + val sc = new SparkContext(sparkConf) + + + val rdd = sc.parallelize(1 to 10000000, 4) + rdd.cache() + rdd.persist(StorageLevel.DISK_ONLY) + rdd.collect() + + val start1 = System.currentTimeMillis() + val res1 = rdd.count() + val end1 = System.currentTimeMillis() + println(s"${end1 - start1}") + + val start2 = System.currentTimeMillis() + val res2 = rdd.count() + val end2 = System.currentTimeMillis() + println(s"${end2 - start2}") + + val start3 = System.currentTimeMillis() + val res3 = rdd.count() + val end3 = System.currentTimeMillis() + println(s"${end3 - start3}") + +// Thread.sleep(1000000) + + sc.stop() + + } + +} diff --git a/src/main/scala/org/example/core/CheckpointDemo.scala b/src/main/scala/org/example/core/CheckpointDemo.scala new file mode 100644 index 0000000..2b97778 --- /dev/null +++ b/src/main/scala/org/example/core/CheckpointDemo.scala @@ -0,0 +1,30 @@ +package org.example.core + +import org.apache.spark.{SparkConf, SparkContext} + +object CheckpointDemo { + def main(args: Array[String]): Unit = { + + val sparkConf = new SparkConf() + .setMaster("local") + .setAppName("my spark app") + + val sc = new SparkContext(sparkConf) + + sc.setCheckpointDir("src/main/resources/checkpoint") + + // 先当成集合来看 + val myTextFile = sc.textFile("src/main/resources/word.txt") + + // 设置检查点 + myTextFile.checkpoint() + + println("myTextFile.count() = " + myTextFile.count()) + + println("myTextFile.isCheckpointed() = " + myTextFile.isCheckpointed) + + println("myTextFile.getCheckpointFile() = " + myTextFile.getCheckpointFile) + + sc.stop() + } +} diff --git a/src/main/scala/org/example/core/DemoCache.scala b/src/main/scala/org/example/core/DemoCache.scala new file mode 100644 index 0000000..8191c7d --- /dev/null +++ b/src/main/scala/org/example/core/DemoCache.scala @@ -0,0 +1,50 @@ +package org.example.core + +import org.apache.spark.storage.StorageLevel +import org.apache.spark.{SparkConf, SparkContext} + +object DemoCache { + def main(args: Array[String]): Unit = { + + val sparkConf = new SparkConf() + .setMaster("local[2]") + .setAppName("scala spark app") + val sc = new SparkContext(sparkConf) + + + val rdd = sc.parallelize(1 to 10000000, 4) + .map { + x => + var sum = 0 + for (i <- 1 to 1000) { + sum += x + } + } + val start1 = System.currentTimeMillis() + val res = rdd.count() + val end1 = System.currentTimeMillis() + println(s"${end1 - start1}") + println("===========") + rdd.cache() // 缓存 + rdd.persist(StorageLevel.MEMORY_ONLY) // 缓存 + rdd.count() + + val start2 = System.currentTimeMillis() + val res2 = rdd.count() + val end2 = System.currentTimeMillis() + println(s"${end2 - start2}") + + // rdd.cache() + // println(rdd.sum()) + // println(rdd.sum()) + // println(rdd.sum()) + // + // + // Thread.sleep(1000000) + + + sc.stop() + + } + +} diff --git a/src/main/scala/org/example/core/DemoCacheNoCache.scala b/src/main/scala/org/example/core/DemoCacheNoCache.scala new file mode 100644 index 0000000..82e0df9 --- /dev/null +++ b/src/main/scala/org/example/core/DemoCacheNoCache.scala @@ -0,0 +1,45 @@ +package org.example.core + +import org.apache.spark.{SparkConf, SparkContext} + +object DemoCacheNoCache { + def main(args: Array[String]): Unit = { + + val sparkConf = new SparkConf() + .setMaster("local[2]") + .setAppName("scala spark app") + val sc = new SparkContext(sparkConf) + + + val rdd = sc.parallelize(1 to 10000000, 4).map { + x => + var sum = 0 + for (i <- 1 to 1000) { + sum += x + } + } + val start1 = System.currentTimeMillis() + val res = rdd.count() + val end1 = System.currentTimeMillis() + println(s"${end1 - start1}") + println("===========") + + val start2 = System.currentTimeMillis() + val res2 = rdd.count() + val end2 = System.currentTimeMillis() + println(s"${end2 - start2}") + + // rdd.cache() + // println(rdd.sum()) + // println(rdd.sum()) + // println(rdd.sum()) + // + // + // Thread.sleep(1000000) + + + sc.stop() + + } + +} diff --git a/src/main/scala/org/example/core/ReadFromCheckpoint.scala b/src/main/scala/org/example/core/ReadFromCheckpoint.scala new file mode 100644 index 0000000..32c64f0 --- /dev/null +++ b/src/main/scala/org/example/core/ReadFromCheckpoint.scala @@ -0,0 +1,28 @@ +package org.example.core + +import org.apache.spark.{SparkConf, SparkContext} + +object ReadFromCheckpoint { + def main(args: Array[String]): Unit = { + + val sparkConf = new SparkConf() + .setMaster("local") + .setAppName("my spark app") + + val sc = new SparkContext(sparkConf) + + sc.setCheckpointDir("src/main/resources/checkpoint") + + val myTextFile = sc.textFile("src/main/resources/word.txt") + + myTextFile.checkpoint() + + println("myTextFile.count() = " + myTextFile.count()) + + println("myTextFile.isCheckpointed() = " + myTextFile.isCheckpointed) + + println("myTextFile.getCheckpointFile() = " + myTextFile.getCheckpointFile) + + sc.stop() + } +} \ No newline at end of file