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