feat:线下课：持久化和缓存、Checkpoint、打包程序提交到yarm

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()
+  }
+}

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()
+
+  }
+
+}

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()
+  }
+}

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()
+
+  }
+
+}

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()
+
+  }
+
+}

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()
+  }
+}