feat:ChatGPT第 4 课：groupByKey、reduceByKey 与 Shuffle

2026-09-17 20:46:18 +08:00
parent ece98345df
commit b5dd87fd84
7 changed files with 305 additions and 0 deletions
@@ -0,0 +1,42 @@
+package org.exercises.exercises4.demo
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+/*
+* groupByKey、reduceByKey 与 Shuffle
+*
+* groupByKey
+→ 相同 Key 的 Value 放到一起
+
+reduceByKey
+→ 相同 Key 的 Value 进行聚合
+
+Shuffle
+→ 为了让相同 Key 的数据重新分配到合适的分区
+
+reduceByKey
+→ 可以先进行本地聚合，再 Shuffle，通常能减少数据传输*/
+object Demo1GroupByKeyAndReduceByKey {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf()
+      .setAppName("Demo1")
+      .setMaster("local")
+    val sc = new SparkContext(conf)
+    val rdd = sc.parallelize(List(
+      ("A", 1),
+      ("B", 2),
+      ("A", 3),
+      ("B", 4),
+      ("A", 5)
+    ))
+    val resGroupByKey = rdd.groupByKey() // 为了让 A 的 Value 全部到一起，Spark 要 Shuffle。
+//    A → [1,3,5]
+//    B → [2,4]
+    println(resGroupByKey.collect().toList)
+
+    val resReduceByKey = rdd.reduceByKey(_ + _) //它可以先在本地做部分聚合,在 Shuffle 前进行本地聚合，减少需要传输的数据量。
+    println(resReduceByKey.collect().toList)
+
+
+  }
+}
@@ -0,0 +1,80 @@
+package org.exercises.exercises4.demo
+
+/*
+* groupByKey：先把 Value 收集起来，再计算。
+reduceByKey：能先算的地方先算，减少 Shuffle 后再计算。*/
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Demo2GroupByKeyVsReduceByKey {
+  def main(args: Array[String]): Unit = {
+
+    val conf = new SparkConf()
+      .setMaster("local[3]")
+      .setAppName("DemoGroupByKeyVsReduceByKey")
+
+    val sc = new SparkContext(conf)
+
+    // =========================================================
+    // 1. 模拟 200 条数据
+    // =========================================================
+    //
+    // 100 个 A，每个 Value = 1
+    // 100 个 B，每个 Value = 2
+    //
+    // 最终：
+    // A = 1 + 1 + ... + 1 = 100
+    // B = 2 + 2 + ... + 2 = 200
+    //
+    val data =
+      (1 to 100).map(_ => ("A", 1)) ++
+        (1 to 100).map(_ => ("B", 2))
+
+    val rdd = sc.parallelize(data, 3)
+
+    println("========== 原始数据 ==========")
+    println("总数据量：" + rdd.count())
+
+    // =========================================================
+    // 2. 查看三个 Partition 中的数据
+    // =========================================================
+
+    println("\n========== 各 Partition 中的数据 ==========")
+
+    rdd.glom().collect().zipWithIndex.foreach {
+      case (partitionData, index) =>
+        println(s"Partition $index：${partitionData.toList}")
+    }
+
+    // =========================================================
+    // 3. groupByKey
+    // =========================================================
+
+    println("\n========== groupByKey ==========")
+
+    val groupByKeyResult = rdd
+      .groupByKey()
+      .mapValues(values => values.sum)
+
+    groupByKeyResult.collect().foreach {
+      case (key, value) =>
+        println(s"$key -> $value")
+    }
+
+    // =========================================================
+    // 4. reduceByKey
+    // =========================================================
+
+    println("\n========== reduceByKey ==========")
+
+    val reduceByKeyResult = rdd
+      .reduceByKey(_ + _)
+
+    reduceByKeyResult.collect().foreach {
+      case (key, value) =>
+        println(s"$key -> $value")
+    }
+
+    sc.stop()
+  }
+}
@@ -0,0 +1,38 @@
+package org.exercises.exercises4.exercises
+/*
+* Exercises1：groupByKey
+创建：
+List(
+  ("A", 1),
+  ("B", 2),
+  ("A", 3),
+  ("B", 4),
+  ("A", 5)
+)
+使用：
+groupByKey()
+并打印结果。
+你应该观察到类似：
+(A,CompactBuffer(1,3,5))
+(B,CompactBuffer(2,4))
+注意顺序不一定固定。
+* */
+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(
+      ("A", 1),
+      ("B", 2),
+      ("A", 3),
+      ("B", 4),
+      ("A", 5)
+    ))
+    val res = rdd.groupByKey()
+    println(res.collect().toList)
+  }
+}
@@ -0,0 +1,31 @@
+package org.exercises.exercises4.exercises
+/*
+* Exercises2：groupByKey + mapValues
+还是上面的数据。
+使用：
+groupByKey()
+然后计算每个 Key 的总和。
+要求使用：
+mapValues
+最终：
+(A,9)
+(B,6)*/
+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)
+    ))
+    val res = rdd.groupByKey().mapValues(_.sum)
+    println(res.collect().toList)
+  }
+}
@@ -0,0 +1,28 @@
+package org.exercises.exercises4.exercises
+/*
+* Exercises3：reduceByKey
+还是相同数据。
+直接：
+reduceByKey(_ + _)
+得到：
+(A,9)
+(B,6)*/
+import org.apache.spark.{SparkConf, 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(
+      ("A", 1),
+      ("B", 2),
+      ("A", 3),
+      ("B", 4),
+      ("A", 5)
+    ))
+    val res = rdd.reduceByKey(_ + _)
+    println(res.collect().toList)
+  }
+}
@@ -0,0 +1,44 @@
+package org.exercises.exercises4.exercises
+/*
+* Exercises4：比较两种写法
+分别计算下面数据：
+List(
+  ("A", 10),
+  ("B", 20),
+  ("A", 30),
+  ("B", 40),
+  ("A", 50)
+)
+方法一
+groupByKey
+→ mapValues(_.sum)
+方法二
+reduceByKey(_ + _)
+两个结果都应该：
+A → 90
+B → 60
+这题的重点不是结果，而是你自己观察：
+groupByKey
+中间会出现什么结构。*/
+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(
+      ("A", 10),
+      ("B", 20),
+      ("A", 30),
+      ("B", 40),
+      ("A", 50)
+    ))
+    val resGBK = rdd.groupByKey().mapValues(_.sum)
+    val resRBK = rdd.reduceByKey(_ + _)
+    println("groupByKey" + resGBK.collect().toList)
+    println("reduceByKey" + resRBK.collect().toList)
+  }
+
+}
@@ -0,0 +1,42 @@
+package org.exercises.exercises4.exercises
+/*
+* Exercises：综合题——WordCount
+重新写一次 WordCount：
+List(
+  "hello spark",
+  "hello scala",
+  "spark hello",
+  "scala spark"
+)
+要求：
+flatMap
+ ↓
+map((_, 1))
+ ↓
+reduceByKey(_ + _)
+ ↓
+collect
+最终：
+hello  → 3
+spark  → 3
+scala  → 2*/
+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(
+      "hello spark",
+      "hello scala",
+      "spark hello",
+      "scala spark"
+    ))
+    val res = rdd.flatMap(_.split(" "))
+      .map((_, 1))
+      .reduceByKey(_ + _)
+    println(res.collect().toList)
+  }
+}