From b5dd87fd845b1270747497b352c8c0784764a72b Mon Sep 17 00:00:00 2001 From: KennethCheng Date: Thu, 17 Sep 2026 20:46:18 +0800 Subject: [PATCH] =?UTF-8?q?feat:ChatGPT=E7=AC=AC=204=20=E8=AF=BE=EF=BC=9Ag?= =?UTF-8?q?roupByKey=E3=80=81reduceByKey=20=E4=B8=8E=20Shuffle?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../demo/Demo1GroupByKeyAndReduceByKey.scala | 42 ++++++++++ .../demo/Demo2GroupByKeyVsReduceByKey.scala | 80 +++++++++++++++++++ .../exercises4/exercises/Exercises1.scala | 38 +++++++++ .../exercises4/exercises/Exercises2.scala | 31 +++++++ .../exercises4/exercises/Exercises3.scala | 28 +++++++ .../exercises4/exercises/Exercises4.scala | 44 ++++++++++ .../exercises4/exercises/Exercises5.scala | 42 ++++++++++ 7 files changed, 305 insertions(+) create mode 100644 src/main/scala/org/exercises/exercises4/demo/Demo1GroupByKeyAndReduceByKey.scala create mode 100644 src/main/scala/org/exercises/exercises4/demo/Demo2GroupByKeyVsReduceByKey.scala create mode 100644 src/main/scala/org/exercises/exercises4/exercises/Exercises1.scala create mode 100644 src/main/scala/org/exercises/exercises4/exercises/Exercises2.scala create mode 100644 src/main/scala/org/exercises/exercises4/exercises/Exercises3.scala create mode 100644 src/main/scala/org/exercises/exercises4/exercises/Exercises4.scala create mode 100644 src/main/scala/org/exercises/exercises4/exercises/Exercises5.scala diff --git a/src/main/scala/org/exercises/exercises4/demo/Demo1GroupByKeyAndReduceByKey.scala b/src/main/scala/org/exercises/exercises4/demo/Demo1GroupByKeyAndReduceByKey.scala new file mode 100644 index 0000000..56adb02 --- /dev/null +++ b/src/main/scala/org/exercises/exercises4/demo/Demo1GroupByKeyAndReduceByKey.scala @@ -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) + + + } +} diff --git a/src/main/scala/org/exercises/exercises4/demo/Demo2GroupByKeyVsReduceByKey.scala b/src/main/scala/org/exercises/exercises4/demo/Demo2GroupByKeyVsReduceByKey.scala new file mode 100644 index 0000000..78290a3 --- /dev/null +++ b/src/main/scala/org/exercises/exercises4/demo/Demo2GroupByKeyVsReduceByKey.scala @@ -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() + } +} \ No newline at end of file diff --git a/src/main/scala/org/exercises/exercises4/exercises/Exercises1.scala b/src/main/scala/org/exercises/exercises4/exercises/Exercises1.scala new file mode 100644 index 0000000..057748f --- /dev/null +++ b/src/main/scala/org/exercises/exercises4/exercises/Exercises1.scala @@ -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) + } +} diff --git a/src/main/scala/org/exercises/exercises4/exercises/Exercises2.scala b/src/main/scala/org/exercises/exercises4/exercises/Exercises2.scala new file mode 100644 index 0000000..8e95395 --- /dev/null +++ b/src/main/scala/org/exercises/exercises4/exercises/Exercises2.scala @@ -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) + } +} diff --git a/src/main/scala/org/exercises/exercises4/exercises/Exercises3.scala b/src/main/scala/org/exercises/exercises4/exercises/Exercises3.scala new file mode 100644 index 0000000..ccf5b93 --- /dev/null +++ b/src/main/scala/org/exercises/exercises4/exercises/Exercises3.scala @@ -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) + } +} diff --git a/src/main/scala/org/exercises/exercises4/exercises/Exercises4.scala b/src/main/scala/org/exercises/exercises4/exercises/Exercises4.scala new file mode 100644 index 0000000..e8bff25 --- /dev/null +++ b/src/main/scala/org/exercises/exercises4/exercises/Exercises4.scala @@ -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) + } + +} diff --git a/src/main/scala/org/exercises/exercises4/exercises/Exercises5.scala b/src/main/scala/org/exercises/exercises4/exercises/Exercises5.scala new file mode 100644 index 0000000..0bf8393 --- /dev/null +++ b/src/main/scala/org/exercises/exercises4/exercises/Exercises5.scala @@ -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) + } +}