feat:ChatGPT 第 6 课：窄依赖、宽依赖、Stage、Task

pom.xml

@@ -62,14 +62,7 @@
             <version>${spark.version}</version>
         </dependency>
 
-        <!-- Hive jdbc-->
+        <!-- Hadoop -->
-        <dependency>
-            <groupId>org.apache.hive</groupId>
-            <artifactId>hive-jdbc</artifactId>
-            <version>2.1.0</version>
-        </dependency>
-
-        <!-- Hadoop Client -->
         <dependency>
             <groupId>org.apache.hadoop</groupId>
             <artifactId>hadoop-client</artifactId>

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

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 数
+  }
+}

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 中。
+  }
+}

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

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的数据聚合在一起
+  }
+}

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注释掉了
+  }
+}