From 973c17b498b2eaf8a7ec26d0e09b9943de6fb691 Mon Sep 17 00:00:00 2001 From: KennethCheng Date: Wed, 16 Sep 2026 17:16:51 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BA=BF=E4=B8=8B=E8=AF=BE=EF=BC=9ARDD?= =?UTF-8?q?=E7=9A=84=E5=88=9B=E5=BB=BA=E3=80=81RDD=E7=9A=84=E6=93=8D?= =?UTF-8?q?=E4=BD=9C=E3=80=81RDD=E9=87=8D=E5=88=86=E5=8C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../java/org/example/core/GroupByKeyDemo.java | 55 +++++++++++++ .../java/org/example/core/MyPartitioner.java | 23 ++++++ .../java/org/example/core/RddCreateDemo.java | 40 ++++++++++ .../example/core/RepartitionVsCoaleasce.java | 77 +++++++++++++++++++ .../example/spark => org/example}/App.scala | 6 +- .../org/example/core/GroupByKeyDemo.scala | 25 ++++++ .../org/example/core/MyPartitioner.scala | 19 +++++ .../org/example/core/RddCreateDemo.scala | 41 ++++++++++ .../example/core/RepartitionVsCoalesce.scala | 67 ++++++++++++++++ .../example/spark => org/example}/test.scala | 7 +- 10 files changed, 354 insertions(+), 6 deletions(-) create mode 100644 src/main/java/org/example/core/GroupByKeyDemo.java create mode 100644 src/main/java/org/example/core/MyPartitioner.java create mode 100644 src/main/java/org/example/core/RddCreateDemo.java create mode 100644 src/main/java/org/example/core/RepartitionVsCoaleasce.java rename src/main/scala/{com/example/spark => org/example}/App.scala (83%) create mode 100644 src/main/scala/org/example/core/GroupByKeyDemo.scala create mode 100644 src/main/scala/org/example/core/MyPartitioner.scala create mode 100644 src/main/scala/org/example/core/RddCreateDemo.scala create mode 100644 src/main/scala/org/example/core/RepartitionVsCoalesce.scala rename src/main/scala/{com/example/spark => org/example}/test.scala (68%) diff --git a/src/main/java/org/example/core/GroupByKeyDemo.java b/src/main/java/org/example/core/GroupByKeyDemo.java new file mode 100644 index 0000000..7b3351e --- /dev/null +++ b/src/main/java/org/example/core/GroupByKeyDemo.java @@ -0,0 +1,55 @@ +package org.example.core; + +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaPairRDD; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.api.java.function.Function; +import scala.Tuple2; + +import java.util.Arrays; + +/** + * Hello world! + */ +public class GroupByKeyDemo { + public static void main(String[] args) throws InterruptedException { + SparkConf sparkConf = new SparkConf() + .setMaster("local") + .setAppName("my spark app"); + JavaSparkContext sc = new JavaSparkContext(sparkConf); + // 先当成集合来看 + JavaRDD myTextFile = sc.textFile("src/main/resources/word.txt"); + + // (hello,[1,1,1,1,1]) + JavaPairRDD> groupByRet = myTextFile.flatMap(line -> Arrays.asList(line.split(" ")).iterator()) + .mapToPair(word -> new Tuple2(word, 1)) + .groupByKey(); + JavaPairRDD result = groupByRet.mapValues(new Function, Integer>() { + @Override + public Integer call(Iterable v1) throws Exception { + int sum = 0; + for (Integer i : v1) { + sum = sum + i; + } + return sum; + } + }); + + Thread.sleep(10000000L); + + + + + } +} + + + + + + + + + + diff --git a/src/main/java/org/example/core/MyPartitioner.java b/src/main/java/org/example/core/MyPartitioner.java new file mode 100644 index 0000000..f48d580 --- /dev/null +++ b/src/main/java/org/example/core/MyPartitioner.java @@ -0,0 +1,23 @@ +package org.example.core; + +import org.apache.spark.Partitioner; + +public class MyPartitioner extends Partitioner { +// private int partitionNum = 0; + + // 我想分成奇数和偶数，所以分区数是2 + @Override + public int numPartitions() { + return 2; + } + + @Override + public int getPartition(Object key) { + if (key instanceof Integer) { + Integer number = (Integer) key; + return number % 2; + } else { + throw new IllegalArgumentException("key must be an integer"); + } + } +} diff --git a/src/main/java/org/example/core/RddCreateDemo.java b/src/main/java/org/example/core/RddCreateDemo.java new file mode 100644 index 0000000..f549f09 --- /dev/null +++ b/src/main/java/org/example/core/RddCreateDemo.java @@ -0,0 +1,40 @@ +package org.example.core; + +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.JavaSparkContext; + +import java.util.ArrayList; +import java.util.List; + +public class RddCreateDemo { + public static void main(String[] args) throws InterruptedException { + + SparkConf sparkConf = new SparkConf() + .setMaster("local[1]") // local模式采用 多线程 模拟 多进程 （executor 个数 、partition） + .setAppName("my spark app"); + JavaSparkContext sc = new JavaSparkContext(sparkConf); + + // 1. 从集合中创建 + List data = new ArrayList<>(); + data.add("a"); + data.add("b"); + data.add("c"); + data.add("d"); + + JavaRDD rdd = sc.parallelize(data); + // 调整并行度 + JavaRDD rdd2 = rdd.repartition(2); +// 在action动作之前，是不是也print不出新的rdd对象 + JavaRDD map = rdd2.map(word -> "test:" + word); + rdd2.saveAsTextFile("src/main/resources/rddtest2"+System.currentTimeMillis()); + map.collect(); +// // 2. 从外部获取（可以是文件，也可以是jdbc【sparkSession】） +// JavaRDD myTextFile = sc.textFile("src/main/resources/word.txt"); +// JavaRDD myHdfsFile = sc.textFile("hdfs://192.168.56.101:8020/word.txt"); +// myTextFile.saveAsTextFile("src/main/resources/word"); + + + Thread.sleep(10000000L); + } +} diff --git a/src/main/java/org/example/core/RepartitionVsCoaleasce.java b/src/main/java/org/example/core/RepartitionVsCoaleasce.java new file mode 100644 index 0000000..c62726d --- /dev/null +++ b/src/main/java/org/example/core/RepartitionVsCoaleasce.java @@ -0,0 +1,77 @@ +package org.example.core; + +import org.apache.spark.SparkConf; +import org.apache.spark.api.java.JavaPairRDD; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.api.java.JavaSparkContext; +import scala.Tuple2; + +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; + +/* +repartition和coalesce都可以用来减少分区，区别在于要不要打散数据。另外repartition可以用来放大分区 +repartition的底层是通过 带 shuffle 的coalesce 来实现的。 + def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope { + coalesce(numPartitions, shuffle = true) + } +shuffle的话 + 缺点：带来一定的网络开销，可能会比较低效 + 优点：可以平均的打散数据，避免出现数据倾斜 + +coalesce通过合并相邻的物理分区来减少分区数量 + 优点：减少不必要的网络开销，高效 + 缺点：不可以平均的打散数据，可能出现数据倾斜 + + */ +public class RepartitionVsCoaleasce { + public static void main(String[] args) throws InterruptedException { + + + SparkConf sparkConf = new SparkConf() + .setMaster("local[2]") + .setAppName("my spark app"); + JavaSparkContext sc = new JavaSparkContext(sparkConf); + + List data = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); + + JavaRDD rdd = sc.parallelize(data, 4); + rdd.cache(); + + System.out.println("初始化分区的数据分布情况："); + printRdd(rdd); + + // 1. repartition + JavaRDD repartitionRdd = rdd.repartition(2); + System.out.println("repartition分区后的数据分布情况："); + printRdd(repartitionRdd); + + // 2. coalesce + JavaRDD coalesceRdd = rdd.coalesce(2,false); + System.out.println("coalesce分区后的数据分布情况："); + printRdd(coalesceRdd); + + // 3. 自定义分区器 + JavaPairRDD pairRDD = rdd.mapToPair(num -> new Tuple2<>(num, null)); + JavaRDD customerPartitionRdd = pairRDD.partitionBy(new MyPartitioner()).map(tuple2 -> tuple2._1); + System.out.println("customerPartitionRdd分区后的数据分布情况："); + printRdd(customerPartitionRdd); + + Thread.sleep(10000000L); + + sc.stop(); + + + } + + private static void printRdd(JavaRDD rdd) { + Iterator>> partitions = rdd.glom().collect().stream().map(Arrays::asList).iterator(); + int i = 0; + while (partitions.hasNext()) { + i++; + List> partiton = partitions.next(); + System.out.println("分区 【"+i+"】 ：" + partiton.get(0)); + } + } +} diff --git a/src/main/scala/com/example/spark/App.scala b/src/main/scala/org/example/App.scala similarity index 83% rename from src/main/scala/com/example/spark/App.scala rename to src/main/scala/org/example/App.scala index 95b35bf..6bfacc0 100644 --- a/src/main/scala/com/example/spark/App.scala +++ b/src/main/scala/org/example/App.scala @@ -1,4 +1,4 @@ -package com.example.spark +package org.example import org.apache.spark.{SparkConf, SparkContext} @@ -8,8 +8,8 @@ object App { val sparkConf = new SparkConf() .setMaster("local[2]") .setAppName("scala spark app") - .set("spark.driver.host", "127.0.0.1") - .set("spark.driver.bindAddress", "127.0.0.1") +// .set("spark.driver.host", "127.0.0.1") +// .set("spark.driver.bindAddress", "127.0.0.1") val sc = new SparkContext(sparkConf) diff --git a/src/main/scala/org/example/core/GroupByKeyDemo.scala b/src/main/scala/org/example/core/GroupByKeyDemo.scala new file mode 100644 index 0000000..999289c --- /dev/null +++ b/src/main/scala/org/example/core/GroupByKeyDemo.scala @@ -0,0 +1,25 @@ +package org.example.core + +import org.apache.spark.{SparkConf, SparkContext} + +object GroupByKeyDemo { + def main(args: Array[String]): Unit = { + + val conf = new SparkConf() + .setMaster("local") + .setAppName("my spark app") + + val sc = new SparkContext(conf) + + val myTextFile = sc.textFile("src/main/resources/word.txt") + + val result = myTextFile + .flatMap(_.split(" ")) + .map(word => (word, 1)) + .groupByKey() + .mapValues(_.sum) + + result.collect().foreach(println) + sc.stop() + } +} \ No newline at end of file diff --git a/src/main/scala/org/example/core/MyPartitioner.scala b/src/main/scala/org/example/core/MyPartitioner.scala new file mode 100644 index 0000000..18b98ef --- /dev/null +++ b/src/main/scala/org/example/core/MyPartitioner.scala @@ -0,0 +1,19 @@ +package org.example.core + +import org.apache.spark.Partitioner + +class MyPartitioner extends Partitioner { + + // 我想分成奇数和偶数，所以分区数是 2 + override def numPartitions = 2 + + override def getPartition(key: Any): Int = { + key match { + case number: Int => + number % 2 + + case _ => + throw new IllegalArgumentException("key must be an integer") + } + } +} \ No newline at end of file diff --git a/src/main/scala/org/example/core/RddCreateDemo.scala b/src/main/scala/org/example/core/RddCreateDemo.scala new file mode 100644 index 0000000..6377b26 --- /dev/null +++ b/src/main/scala/org/example/core/RddCreateDemo.scala @@ -0,0 +1,41 @@ +package org.example.core + +import org.apache.spark.{SparkConf, SparkContext} + +import java.io.File + +object RddCreateDemo { + def main(args: Array[String]): Unit = { + + val sc = new SparkContext(new SparkConf().setAppName("rdd").setMaster("local[2]")) + + // 1. 从集合中创建 RDD + val data = List("a", "b", "c", "d") + + val rdd2 = sc.parallelize(data, 4) + + // 调整并行度 + // val rdd2 = rdd.repartition(1) + + // 转换算子：map + val map = rdd2.map(word => "test:" + word) + + // Action：保存 RDD + if (!new File("src/main/resources/rddtest").exists()) { + rdd2.saveAsTextFile("src/main/resources/rddtest") + } + + // Action：收集结果 + map.collect() + + // 2. 从外部获取 + val myTextFile = sc.textFile("src/main/resources/word.txt", 2) + // val myHdfsFile = sc.textFile("hdfs://192.168.56.101:8020/word.txt") + if (!new File("src/main/resources/word").exists()) { + myTextFile.saveAsTextFile("src/main/resources/word") + } + + // Thread.sleep(10000000L) + sc.stop() + } +} \ No newline at end of file diff --git a/src/main/scala/org/example/core/RepartitionVsCoalesce.scala b/src/main/scala/org/example/core/RepartitionVsCoalesce.scala new file mode 100644 index 0000000..0076b36 --- /dev/null +++ b/src/main/scala/org/example/core/RepartitionVsCoalesce.scala @@ -0,0 +1,67 @@ +package org.example.core + +import org.apache.spark.{SparkConf, SparkContext} + +object RepartitionVsCoalesce { + + def main(args: Array[String]): Unit = { + + val conf = new SparkConf() + .setMaster("local[2]") + .setAppName("my spark app") + + val sc = new SparkContext(conf) + + val data = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) + + // 创建 4 个分区 + val rdd = sc.parallelize(data, 4) + + rdd.cache() + + println("初始化分区的数据分布情况：") + printRdd(rdd) + + // 1. repartition + // 增加或减少分区都会 shuffle + val repartitionRdd = rdd.repartition(2) + + println("repartition 分区后的数据分布情况：") + printRdd(repartitionRdd) + + // 2. coalesce + // 减少分区，第二个参数 false 表示不进行 shuffle + val coalesceRdd = rdd.coalesce(2, shuffle = false) + + println("coalesce 分区后的数据分布情况：") + printRdd(coalesceRdd) + + // 3. 自定义分区器 + // 先把普通 RDD 转成 Pair RDD + val pairRDD = rdd.map((_, null)) + + // 使用自定义分区器 + val customerPartitionRdd = + pairRDD + .partitionBy(new MyPartitioner) + .map(_._1) + + println("customerPartitionRdd 分区后的数据分布情况：") + printRdd(customerPartitionRdd) + + sc.stop() + } + + + def printRdd(rdd: org.apache.spark.rdd.RDD[Int]): Unit = { + + val partitions = rdd.glom().collect() + + partitions.zipWithIndex.foreach { + case (partition, index) => + println( + s"分区【${index + 1}】：${partition.mkString("[", ", ", "]")}" + ) + } + } +} diff --git a/src/main/scala/com/example/spark/test.scala b/src/main/scala/org/example/test.scala similarity index 68% rename from src/main/scala/com/example/spark/test.scala rename to src/main/scala/org/example/test.scala index 01bc5e4..8b26561 100644 --- a/src/main/scala/com/example/spark/test.scala +++ b/src/main/scala/org/example/test.scala @@ -1,4 +1,4 @@ -package com.example.spark +package org.example import org.apache.spark.{SparkConf, SparkContext} @@ -7,8 +7,9 @@ object test { val sc = new SparkContext(new SparkConf() .setMaster("local") .setAppName("test") - .set("spark.driver.host", "127.0.0.1") - .set("spark.driver.bindAddress", "127.0.0.1")) +// .set("spark.driver.host", "127.0.0.1") +// .set("spark.driver.bindAddress", "127.0.0.1") + ) sc.textFile("src/main/resources/word.txt").foreach(println) } }