feat:ChatGPT第 5 课：Partition 分区

src/main/scala/org/exercises/exercises5/demo/Demo1PartitionBasic.scala

@@ -0,0 +1,62 @@
+package org.exercises.exercises5.demo
+
+/*
+* Spark 第 5 课：Partition 分区*/
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Demo1PartitionBasic {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf().setAppName("Demo1PartitionBasic").setMaster("local")
+    val sc = new SparkContext(conf)
+    val rdd = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8), 2)
+    println("=============获取分区===========")
+    println("=============获取分区===========")
+    println(rdd.getNumPartitions) // 2
+
+    println("=============查看分区的数据===========")
+    val res = rdd.glom()
+    println(res.collect().toList) // List([I@7d755813, [I@69c93ca4)
+
+    //    分区 0: 1, 2, 3, 4
+    //    分区 1: 5, 6, 7, 8
+    printPartition(res)
+
+    println("=============Transformation 一般会不会改变 Partition 数？===========")
+    println("=============Transformation 一般会不会改变 Partition 数？===========")
+    val rddMap = sc.parallelize(
+      List(1, 2, 3, 4, 5, 6),
+      2
+    )
+
+    val rddMap2 = rddMap.map(_ * 10)
+    //    Partition 0
+    //    1 2 3
+    //    ↓ map
+    //    10 20 30
+    //
+    //    Partition 1
+    //    4 5 6
+    //    ↓ map
+    //    40 50 60
+    //  map 是在原有 Partition 中进行转换
+    println(s"map前：${rddMap.getNumPartitions}")
+    println(s"map后：${rddMap2.getNumPartitions}")
+    println(rddMap2.toDebugString)
+
+    //    map
+    //    filter
+    //    flatMap
+    //    这类算子通常属于窄依赖，数据不需要大规模跨 Partition 重分配。
+
+    sc.stop()
+  }
+
+  private def printPartition(res: RDD[Array[Int]]): Unit = {
+    res.collect().zipWithIndex.foreach {
+      case (data, index) =>
+        println(s"分区 $index: ${data.mkString(", ")}")
+    }
+  }
+}

src/main/scala/org/exercises/exercises5/demo/Demo2Shuffle.scala

@@ -0,0 +1,79 @@
+package org.exercises.exercises5.demo
+
+/*
+* 什么操作会导致 Shuffle？*/
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object Demo2Shuffle {
+
+  def main(args: Array[String]): Unit = {
+
+    val conf = new SparkConf()
+      .setAppName("ShuffleDemo")
+      .setMaster("local[2]")
+
+    val sc = new SparkContext(conf)
+
+    val rdd = sc.parallelize(
+      List(
+        ("A", 1),
+        ("B", 2),
+        ("A", 3),
+        ("B", 4),
+        ("A", 5),
+        ("B", 6)
+      ),
+      2
+    )
+
+    // 查看原始 RDD 每个 Partition 里面的数据
+    println("========== 原始 RDD ==========")
+
+    rdd.glom().collect().zipWithIndex.foreach {
+      case (data, index) =>
+        println(s"Partition $index: ${data.toList}")
+    }
+
+
+    // ==========================
+    // reduceByKey
+    // ==========================
+
+    println("\n========== reduceByKey ==========")
+
+    val reduceRdd = rdd.reduceByKey(_ + _, 2)
+
+    println("reduceByKey 后的 Partition：")
+
+    reduceRdd.glom().collect().zipWithIndex.foreach {
+      case (data, index) =>
+        println(s"Partition $index: ${data.toList}")
+    }
+
+    println("\nreduceByKey 的依赖关系：")
+    println(reduceRdd.toDebugString)
+
+
+    // ==========================
+    // groupByKey
+    // ==========================
+
+    println("\n========== groupByKey ==========")
+
+    val groupRdd = rdd.groupByKey(2)
+
+    println("groupByKey 后的 Partition：")
+
+    groupRdd.glom().collect().zipWithIndex.foreach {
+      case (data, index) =>
+        println(s"Partition $index: ${data.toList}")
+    }
+
+    println("\ngroupByKey 的依赖关系：")
+    println(groupRdd.toDebugString)
+
+
+    sc.stop()
+  }
+}

src/main/scala/org/exercises/exercises5/demo/Demo3PartitionBy.scala

@@ -0,0 +1,26 @@
+package org.exercises.exercises5.demo
+
+/*
+* 现在我们开始主动控制 PairRDD 的分区方式。*/
+
+import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
+
+object Demo3PartitionBy {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf()
+      .setAppName("Demo3PartitionBy")
+      .setMaster("local")
+    val sc = new SparkContext(conf)
+    val rdd = sc.parallelize(
+      List(
+        ("A", 1),
+        ("B", 2),
+        ("C", 3),
+        ("D", 4)
+      ), 5
+    )
+    println("partitionByq前：" + rdd.getNumPartitions)
+    val result = rdd.partitionBy(new HashPartitioner(2))
+    println("partitionByq后：" + result.getNumPartitions)
+  }
+}

src/main/scala/org/exercises/exercises5/demo/Demo4MyPartitioner.scala

@@ -0,0 +1,39 @@
+package org.exercises.exercises5.demo
+
+/*
+* 奇数偶数分区*/
+
+import org.apache.spark.{Partitioner, SparkConf, SparkContext}
+
+
+object Demo4MyPartitioner {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf()
+      .setAppName("MyApp")
+      .setMaster("local")
+    val sc = new SparkContext(conf)
+    val pairRDD = sc.parallelize(
+      List(
+        (1, "A"),
+        (2, "B"),
+        (3, "C"),
+        (4, "D"),
+        (5, "E"),
+        (6, "F")
+      ), 5
+    )
+    val result = pairRDD.partitionBy(new MyPartitioner)
+    println("partitionBy前：" + pairRDD.getNumPartitions)
+    println("partitionBy后：" + result.getNumPartitions)
+    result.glom().collect().foreach(x => println(x.mkString(",")))
+  }
+}
+
+class MyPartitioner extends Partitioner {
+
+  override def numPartitions: Int = 2
+
+  override def getPartition(key: Any): Int = {
+    key.asInstanceOf[Int] % 2
+  }
+}

src/main/scala/org/exercises/exercises5/exercises/Exercises1.scala

@@ -0,0 +1,34 @@
+package org.exercises.exercises5.exercises
+
+/*
+* Exercises1：观察 Partition
+创建：
+List(1, 2, 3, 4, 5, 6, 7, 8)
+并且指定：
+4 个 Partition
+要求打印：
+RDD 的 Partition 数量
+每个 Partition 中的数据
+提示使用：
+getNumPartitions()
+glom()
+collect()*/
+
+import org.apache.spark.SparkConf
+import org.apache.spark.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, 7, 8), 4)
+    println(rdd.getNumPartitions)
+    rdd.glom().collect().zipWithIndex.foreach {
+      case (data, index) =>
+        println(s"Partition $index: ${data.toList}")
+    }
+    sc.stop()
+  }
+}

src/main/scala/org/exercises/exercises5/exercises/Exercises2.scala

@@ -0,0 +1,28 @@
+package org.exercises.exercises5.exercises
+/*
+* Exercises2：观察 map 是否改变 Partition
+创建：
+List(1,2,3,4,5,6,7,8)
+4 个 Partition。
+然后：
+val result = rdd.map(_ * 10)
+打印：
+原 RDD Partition 数
+result Partition 数
+要求验证：
+map 前后 Partition 数是否相同。*/
+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(1, 2, 3, 4, 5, 6, 7, 8), 4)
+    val result = rdd.map(_ * 10)
+    println(rdd.getNumPartitions)
+    println(result.getNumPartitions)
+    sc.stop()
+  }
+}

src/main/scala/org/exercises/exercises5/exercises/Exercises3.scala

@@ -0,0 +1,46 @@
+package org.exercises.exercises5.exercises
+
+import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
+
+/*
+* Exercises3：使用 partitionBy
+创建：
+List(
+  ("A", 1),
+  ("B", 2),
+  ("C", 3),
+  ("D", 4),
+  ("E", 5),
+  ("F", 6)
+)
+转换为 PairRDD。
+使用：
+new HashPartitioner(2)
+进行：
+partitionBy
+然后使用：
+glom()
+观察：
+数据最终进入了哪个 Partition。*/
+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),
+      ("C", 3),
+      ("D", 4),
+      ("E", 5),
+      ("F", 6)
+    ))
+    val res = rdd.partitionBy(new HashPartitioner(2))
+    res.glom().collect().zipWithIndex.foreach {
+      case (value, index) =>
+        println(s"Partition $index: ${value.mkString("Array(", ", ", ")")}")
+    }
+    sc.stop()
+  }
+}

src/main/scala/org/exercises/exercises5/exercises/Exercises4.scala

@@ -0,0 +1,68 @@
+package org.exercises.exercises5.exercises
+
+/*
+* Exercises4：自定义奇偶分区
+定义：
+class MyPartitioner extends Partitioner
+要求：
+偶数 → Partition 0
+奇数 → Partition 1
+数据：
+List(
+  (1, "one"),
+  (2, "two"),
+  (3, "three"),
+  (4, "four"),
+  (5, "five"),
+  (6, "six")
+)
+使用：
+partitionBy(new MyPartitioner)
+然后通过：
+glom()
+观察结果。
+预期：
+Partition 0:
+(2,two)
+(4,four)
+(6,six)
+
+Partition 1:
+(1,one)
+(3,three)
+(5,five)*/
+
+import org.apache.spark.{Partitioner, SparkConf, SparkContext}
+
+object Exercises4 {
+  def main(args: Array[String]): Unit = {
+    val conf = new SparkConf()
+      .setMaster("local")
+      .setAppName("Exercises4")
+    val sc = new SparkContext(conf)
+    val rdd = sc.parallelize(List(
+      (1, "one"),
+      (2, "two"),
+      (3, "three"),
+      (4, "four"),
+      (5, "five"),
+      (6, "six")
+    ))
+    val res = rdd.partitionBy(new MyPartitioner())
+    res.glom().collect().zipWithIndex.foreach {
+      case (arr, index) => println(s"分区 $index: ${arr.mkString(", ")}")
+    }
+    sc.stop()
+  }
+}
+
+// 一共有几个分区？
+// 数据怎么分？
+class MyPartitioner extends Partitioner {
+  // 我的 RDD 一共有 2 个分区
+  override def numPartitions: Int = 2
+
+  override def getPartition(key: Any): Int = {
+    key.asInstanceOf[Int] % 2
+  }
+}

src/main/scala/org/exercises/exercises5/exercises/Exercises5.scala

@@ -0,0 +1,77 @@
+package org.exercises.exercises5.exercises
+
+/*
+* Exercises25：综合题——观察 Shuffle
+创建：
+List(
+  ("A", 1),
+  ("B", 2),
+  ("A", 3),
+  ("B", 4),
+  ("A", 5),
+  ("C", 6)
+)
+要求完成两个版本。
+版本一
+groupByKey
+然后观察：
+glom()
+版本二
+reduceByKey(_ + _)
+然后观察：
+glom()
+最后打印：
+Partition 数量
+Partition 中的数据
+最终结果
+你不需要通过代码证明 Shuffle 的内部每一个细节，重点观察：
+groupByKey
+→ 分组后的数据
+
+reduceByKey
+→ 聚合后的数据
+并思考：
+为什么两个算子的最终结果类似，但 Shuffle 阶段处理的数据量可能不同？*/
+
+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(
+      ("A", 1),
+      ("B", 2),
+      ("A", 3),
+      ("B", 4),
+      ("A", 5),
+      ("C", 6)
+    ), 4)
+    println("=============原始数据==============")
+
+    println(rdd.getNumPartitions)
+    rdd.glom().collect().zipWithIndex.foreach {
+      case (value, index) => println(s"Partition $index: ${value.toList}")
+    }
+
+    println("=============reduceByKey==============")
+    val resRBK = rdd.reduceByKey(_ + _)
+    println(resRBK.getNumPartitions)
+
+    resRBK.glom().collect().zipWithIndex.foreach {
+      case (value, index) => println(s"Partition $index: ${value.toList}")
+    }
+
+
+    println("=============groupByKey==============")
+    val resGBK = rdd.groupByKey()
+    println(resGBK.getNumPartitions)
+
+    resGBK.glom().collect().zipWithIndex.foreach {
+      case (value, index) => println(s"Partition $index: ${value.toList}")
+    }
+    sc.stop()
+  }
+}