feat:ChatGPT第 3 课：PairRDD 与 reduceByKey

2026-09-16 23:17:35 +08:00
parent cf43dd1428
commit bd8c637b48
10 changed files with 315 additions and 0 deletions
@@ -0,0 +1,37 @@
+package org.exercises.exercises3.demo
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+/*
+* Spark 第 3 课：PairRDD 与 reduceByKey*/
+object Demo1Map {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Demo1").setMaster("local"))
+
+    // 先看普通rdd，类型是RDD[Int]
+    val normalRdd = sc.parallelize(List(1, 2, 3, 4))
+
+    // RDD[(String, Int)]这种以二元组作为元素的 RDD，通常称为：PairRDD
+    val PairRdd = sc.parallelize(
+      List(
+        ("张三", 80),
+        ("李四", 90),
+        ("王五", 85)
+      )
+    )
+
+    val student = ("张三", 80)
+    println(student._1) // 得到 张三
+    println(student._2) // 得到 80
+
+
+    // ("张三",80) → 80
+    // ("李四",90) → 90
+    // ("王五",85) → 85
+    // RDD(80,90,85)
+    val result = PairRdd.map(_._2)
+    println(result.collect().toList)
+
+    sc.stop()
+  }
+}
@@ -0,0 +1,20 @@
+package org.exercises.exercises3.demo
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Demo2MapValues {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Demo2MapValues").setMaster("local"))
+    val rdd = sc.parallelize(
+      List(
+        ("张三", 80),
+        ("李四", 90),
+        ("王五", 85)
+      )
+    )
+    // 如果我们只想修改 value，可以使用：mapValues
+    val result = rdd.mapValues(_ + 5)
+    // List((张三,85), (李四,95), (王五,90))
+    println(result.collect().toList)
+  }
+}
@@ -0,0 +1,25 @@
+package org.exercises.exercises3.demo
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Demo3ReduceByKey {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Demo3ReduceByKey").setMaster("local"))
+    val rdd = sc.parallelize(
+      List(
+        ("张三", 80),
+        ("李四", 90),
+        ("张三", 70),
+        ("李四", 60)
+      )
+    )
+    // 现在我们想：
+    // 按名字统计每个人的总成绩。
+    // 也就是：
+    // 张三：80 + 70 = 150
+    // 李四：90 + 60 = 150
+    val result = rdd.reduceByKey(_ + _)
+    // List((张三,150), (李四,150))
+    println(result.collect().toList)
+  }
+}
@@ -0,0 +1,23 @@
+package org.exercises.exercises3.demo
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Demo4GroupByKey {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Demo4GroupByKey").setMaster("local"))
+    val rdd = sc.parallelize(
+      List(
+        ("A", 1),
+        ("B", 2),
+        ("A", 3),
+        ("B", 4)
+      )
+    )
+    val result = rdd
+      // A → [1,3]
+      // B → [2,4]
+      .groupByKey() // List((B,CompactBuffer(2, 4)), (A,CompactBuffer(1, 3)))
+      .mapValues(_.sum) // List((B,6), (A,4))
+    println(result.collect().toList)
+  }
+}
@@ -0,0 +1,20 @@
+package org.exercises.exercises3.demo
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Demo5WordCount {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Demo5WordCount").setMaster("local"))
+    val rdd = sc.parallelize(
+      List(
+        "hello spark",
+        "hello scala",
+        "spark is good"
+      )
+    )
+    val words = rdd.flatMap(_.split(" "))
+    val pairRDD = words.map((_, 1))
+    val result = pairRDD.reduceByKey(_ + _)
+    println(result.collect().toList)
+  }
+}
@@ -0,0 +1,32 @@
+package org.exercises.exercises3.exercises
+
+/*
+* Exercises1：读取 key 和 value
+创建：
+List(
+  ("张三", 80),
+  ("李四", 90),
+  ("王五", 70)
+)
+转成 RDD。
+只提取成绩。
+要求最终：
+List(80, 90, 70)
+使用：
+map*/
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Exercises1 {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Exercises1").setMaster("local"))
+    val rdd = sc.parallelize(List(
+      ("张三", 80),
+      ("李四", 90),
+      ("王五", 70)
+    ))
+    val result = rdd.map(_._2)
+    println(result.collect().toList)
+    sc.stop()
+  }
+}
@@ -0,0 +1,35 @@
+package org.exercises.exercises3.exercises
+
+/*
+* Exercises2：mapValues
+还是：
+List(
+  ("张三", 80),
+  ("李四", 90),
+  ("王五", 70)
+)
+要求：
+所有人成绩加 10 分。
+
+最终：
+("张三",90)
+("李四",100)
+("王五",80)
+要求使用：
+mapValues*/
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Exercises2 {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Exercises2").setMaster("local"))
+    val rdd = sc.parallelize(List(
+      ("张三", 80),
+      ("李四", 90),
+      ("王五", 70)
+    ))
+    val result = rdd.mapValues(_ + 10).collect()
+    result.foreach(println)
+    sc.stop()
+  }
+}
@@ -0,0 +1,39 @@
+package org.exercises.exercises3.exercises
+
+/*
+* Exercises3：reduceByKey
+创建：
+List(
+  ("张三", 80),
+  ("李四", 90),
+  ("张三", 70),
+  ("李四", 60),
+  ("王五", 100)
+)
+要求：
+统计每个人的总成绩。
+
+最终应该是：
+("张三",150)
+("李四",150)
+("王五",100)
+使用：
+reduceByKey*/
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Exercises3 {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Exercises3").setMaster("local"))
+    val rdd = sc.parallelize(List(
+      ("张三", 80),
+      ("李四", 90),
+      ("张三", 70),
+      ("李四", 60),
+      ("王五", 100)
+    ))
+    val result = rdd.reduceByKey(_ + _)
+    result.foreach(println)
+    sc.stop()
+  }
+}
@@ -0,0 +1,39 @@
+package org.exercises.exercises3.exercises
+
+/*
+* Exercises4：统计商品销量
+创建：
+List(
+  ("苹果", 3),
+  ("香蕉", 5),
+  ("苹果", 2),
+  ("橘子", 4),
+  ("香蕉", 3),
+  ("苹果", 5)
+)
+使用：
+reduceByKey(_ + _)
+计算每种水果的总销量。
+结果应该是：
+苹果 → 10
+香蕉 → 8
+橘子 → 4*/
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Exercises4 {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Exercises4").setMaster("local"))
+    val rdd = sc.parallelize(List(
+      ("苹果", 3),
+      ("香蕉", 5),
+      ("苹果", 2),
+      ("橘子", 4),
+      ("香蕉", 3),
+      ("苹果", 5)
+    ))
+    val result = rdd.reduceByKey(_ + _)
+    result.foreach(println)
+    sc.stop()
+  }
+}
@@ -0,0 +1,45 @@
+package org.exercises.exercises3.exercises
+
+/*
+* Exercises5：正式练 WordCount
+数据：
+List(
+  "hello spark",
+  "hello scala",
+  "spark is powerful",
+  "hello spark"
+)
+完成 WordCount。
+要求依次使用：
+flatMap
+↓
+map
+↓
+reduceByKey
+↓
+collect
+最终结果应该包含：
+hello    3
+spark    3
+scala    1
+is       1
+powerful 1*/
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Exercises5 {
+  def main(args: Array[String]): Unit = {
+    val sc = new SparkContext(new SparkConf().setAppName("Exercises5").setMaster("local"))
+    val rdd = sc.parallelize(List(
+      "hello spark",
+      "hello scala",
+      "spark is powerful",
+      "hello spark"
+    ))
+    val result = rdd.flatMap(_.split(" "))
+      .map((_, 1))
+      .reduceByKey(_ + _)
+    result.foreach(println)
+    sc.stop()
+  }
+}