diff --git a/src/main/scala/org/exercises/exercises5/demo/Demo1PartitionBasic.scala b/src/main/scala/org/exercises/exercises5/demo/Demo1PartitionBasic.scala new file mode 100644 index 0000000..38bfddf --- /dev/null +++ b/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(", ")}") + } + } +} diff --git a/src/main/scala/org/exercises/exercises5/demo/Demo2Shuffle.scala b/src/main/scala/org/exercises/exercises5/demo/Demo2Shuffle.scala new file mode 100644 index 0000000..638983b --- /dev/null +++ b/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() + } +} \ No newline at end of file diff --git a/src/main/scala/org/exercises/exercises5/demo/Demo3PartitionBy.scala b/src/main/scala/org/exercises/exercises5/demo/Demo3PartitionBy.scala new file mode 100644 index 0000000..097d984 --- /dev/null +++ b/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) + } +} diff --git a/src/main/scala/org/exercises/exercises5/demo/Demo4MyPartitioner.scala b/src/main/scala/org/exercises/exercises5/demo/Demo4MyPartitioner.scala new file mode 100644 index 0000000..210aabd --- /dev/null +++ b/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 + } +} \ No newline at end of file diff --git a/src/main/scala/org/exercises/exercises5/exercises/Exercises1.scala b/src/main/scala/org/exercises/exercises5/exercises/Exercises1.scala new file mode 100644 index 0000000..defa67a --- /dev/null +++ b/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() + } +} diff --git a/src/main/scala/org/exercises/exercises5/exercises/Exercises2.scala b/src/main/scala/org/exercises/exercises5/exercises/Exercises2.scala new file mode 100644 index 0000000..60b26fc --- /dev/null +++ b/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() + } +} diff --git a/src/main/scala/org/exercises/exercises5/exercises/Exercises3.scala b/src/main/scala/org/exercises/exercises5/exercises/Exercises3.scala new file mode 100644 index 0000000..4d060cc --- /dev/null +++ b/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() + } +} diff --git a/src/main/scala/org/exercises/exercises5/exercises/Exercises4.scala b/src/main/scala/org/exercises/exercises5/exercises/Exercises4.scala new file mode 100644 index 0000000..a741b3b --- /dev/null +++ b/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 + } +} \ No newline at end of file diff --git a/src/main/scala/org/exercises/exercises5/exercises/Exercises5.scala b/src/main/scala/org/exercises/exercises5/exercises/Exercises5.scala new file mode 100644 index 0000000..37c3f79 --- /dev/null +++ b/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() + } +}