diff --git a/pom.xml b/pom.xml index 0e96d8b..af17aee 100644 --- a/pom.xml +++ b/pom.xml @@ -62,14 +62,7 @@ ${spark.version} - - - org.apache.hive - hive-jdbc - 2.1.0 - - - + org.apache.hadoop hadoop-client diff --git a/src/main/scala/org/exercises/exercises6/demo/Demo.scala b/src/main/scala/org/exercises/exercises6/demo/Demo.scala new file mode 100644 index 0000000..152d540 --- /dev/null +++ b/src/main/scala/org/exercises/exercises6/demo/Demo.scala @@ -0,0 +1,28 @@ +package org.exercises.exercises6.demo + +/* +* Spark 第 6 课:窄依赖、宽依赖、Stage、Task*/ + +import org.apache.spark.{SparkConf, SparkContext} + +object Demo { + def main(args: Array[String]): Unit = { + val conf = new SparkConf().setAppName("Demo").setMaster("local") + val sc = new SparkContext(conf) + val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 2 + ) + // 窄依赖: + // map + // filter + // flatMap + // mapValues + // mapPartitions + val result = rdd + .map(_ * 10) + .filter(_ > 20) + + println(result.collect().toList) + } +} diff --git a/src/main/scala/org/exercises/exercises6/exercises/Exercises1.scala b/src/main/scala/org/exercises/exercises6/exercises/Exercises1.scala new file mode 100644 index 0000000..9a93974 --- /dev/null +++ b/src/main/scala/org/exercises/exercises6/exercises/Exercises1.scala @@ -0,0 +1,50 @@ +package org.exercises.exercises6.exercises + +/* +* Exercises1:判断窄依赖 +创建一个 RDD: +val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 3 +) +执行: +val result = rdd + .map(_ * 10) + .filter(_ > 20) +要求打印: +原始 RDD 分区数 +result 分区数 +result 每个 Partition 的数据 +使用: +getNumPartitions() +glom()*/ + +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(_ * 10) + .filter(_ > 20) + + println("=========原始分区数========") + println(rdd.getNumPartitions) + println("=========原始每个partition数据========") + rdd.glom().zipWithIndex().foreach { + case (data, index) => println(s"partition $index: ${data.toList}") + } + println("=========result分区数========") + println(result.getNumPartitions) + println("=========result每个partition数据========") + result.glom().zipWithIndex().foreach { + case (data, index) => println(s"partition $index: ${data.toList}") + } + // map 和 filter 没有改变 Partition 数 + } +} diff --git a/src/main/scala/org/exercises/exercises6/exercises/Exercises2.scala b/src/main/scala/org/exercises/exercises6/exercises/Exercises2.scala new file mode 100644 index 0000000..efc8b94 --- /dev/null +++ b/src/main/scala/org/exercises/exercises6/exercises/Exercises2.scala @@ -0,0 +1,56 @@ +package org.exercises.exercises6.exercises + +/* +* Exercises2:观察 reduceByKey +创建: +val rdd = sc.parallelize( + List( + ("A", 1), + ("B", 2), + ("A", 3), + ("B", 4), + ("A", 5) + ), + 3 +) +执行: +val result = rdd.reduceByKey(_ + _) +打印: +原始 Partition 数 +result Partition 数 +result 的 Partition 数据 +然后自己回答: +reduceByKey 为什么需要 Shuffle?*/ + +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( + List( + ("A", 1), + ("B", 2), + ("A", 3), + ("B", 4), + ("A", 5) + ), + 3 + ) + val result = rdd.reduceByKey(_ + _) + println("========原始Partition数========") + println(rdd.getNumPartitions) + println("========原始Partition数据========") + rdd.glom().zipWithIndex().foreach { + case (partition, index) => println(s"Partition $index: ${partition.toList}") + } + println("========result的Partition数========") + println(result.getNumPartitions) + println("========result的Partition数据========") + result.glom().zipWithIndex().foreach { + case (partition, index) => println(s"Partition $index: ${partition.toList}") + } + // 因为 reduceByKey 需要将相同 key 的数据聚合到一起,而这些数据可能分布在不同的 Partition 中,因此需要进行 Shuffle 操作,将相同 key 的数据移动到同一个 Partition 中。 + } +} diff --git a/src/main/scala/org/exercises/exercises6/exercises/Exercises3.scala b/src/main/scala/org/exercises/exercises6/exercises/Exercises3.scala new file mode 100644 index 0000000..531990a --- /dev/null +++ b/src/main/scala/org/exercises/exercises6/exercises/Exercises3.scala @@ -0,0 +1,38 @@ +package org.exercises.exercises6.exercises + +/* +* Exercises3:一个 Job 有几个 Stage? +写: +val result = rdd + .map(_ * 10) + .filter(_ > 20) + .collect() +运行程序,然后打开: +http://hadoop:4040 +观察: +Jobs +Stages +重点看: +这个 Job 有几个 Stage? +这个 Stage 有几个 Task?*/ + +import org.apache.spark.SparkConf +import org.apache.spark.SparkContext + +object Exercises3 { + def main(args: Array[String]): Unit = { + val conf = new SparkConf().setAppName("Exercises3").setMaster("local[*]") + val sc = new SparkContext(conf) + val rdd = sc.parallelize( + List(1, 2, 3, 4, 5, 6), + 3 + ) + val result = rdd + .map(_ * 10) + .filter(_ > 20) + .collect() + + Thread.sleep(1000000) + // 这个job有1个stage,这个stage有3个task + } +} diff --git a/src/main/scala/org/exercises/exercises6/exercises/Exercises4.scala b/src/main/scala/org/exercises/exercises6/exercises/Exercises4.scala new file mode 100644 index 0000000..7950caa --- /dev/null +++ b/src/main/scala/org/exercises/exercises6/exercises/Exercises4.scala @@ -0,0 +1,41 @@ +package org.exercises.exercises6.exercises + +/* +* Exercises4:加入 Shuffle +写: +val result = rdd + .map(x => (x % 2, x)) + .reduceByKey(_ + _) + .collect() +打开 Spark UI。 +观察: +Job +Stage +Task +然后比较 Exercises3: +map + filter + collect +和 Exercises4: +map + reduceByKey + collect +看看 Stage 有没有变化。*/ + +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 + .map(x => (x % 2, x)) + .reduceByKey(_ + _) + .collect() + + result.foreach(println) + + Thread.sleep(100000) + // 多了一个Stage,因为reduceByKeyy要将相同key的数据聚合在一起 + } +} diff --git a/src/main/scala/org/exercises/exercises6/exercises/Exercises5.scala b/src/main/scala/org/exercises/exercises6/exercises/Exercises5.scala new file mode 100644 index 0000000..2a41f0a --- /dev/null +++ b/src/main/scala/org/exercises/exercises6/exercises/Exercises5.scala @@ -0,0 +1,23 @@ +package org.exercises.exercises6.exercises + +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, 7, 8 + ), + 4 + ) + val result = rdd + // .filter(_ % 2 == 0) + .map(x => (if (x % 2 == 0) "偶数" else "奇数", x)) + .reduceByKey(_ + _) + .map(x => (x._1, x._2 * 10)) + result.foreach(println) + // 筛选偶数后不就少了一列奇数了吗?所以我把filter注释掉了 + } +}