当前位置：首页 > news >正文

大数据Spark（六十九）：Transformation转换算子intersection和subtract使用案例

news 2025/10/18 5:49:39

文章目录

Transformation转换算子intersection和subtract使用案例

一、intersection使用案例

二、subtract使用案例

Transformation转换算子intersection和subtract使用案例

一、intersection使用案例

取两个RDD数据集的交集。

注意：返回新的RDD分区数与父RDD分区多的一致。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("IntersectionTest");
JavaSparkContext sc = new JavaSparkContext(conf);JavaRDD<String> rdd1 = sc.parallelize(Arrays.asList("a","b","c","d"), 3);
JavaRDD<String> rdd2 = sc.parallelize(Arrays.asList("c","d","e","f"), 4);//intersection算子:对两个RDD进行intersection操作，返回一个新的RDD，RDD的分区数与父RDD分区数多的保持一致。
JavaRDD<String> rdd3 = rdd1.intersection(rdd2);
System.out.println("rdd3 分区数：" + rdd3.getNumPartitions());rdd3.foreach(x-> System.out.println(x));sc.stop();

Scala代码：

val conf: SparkConf = new SparkConf().setMaster("local").setAppName("IntersectionTest")
val sc = new SparkContext(conf)
val rdd1: RDD[String] = sc.parallelize(List("a", "b", "c", "d"), 4)
val rdd2: RDD[String] = sc.parallelize(List("c", "d", "e", "f"), 3)//intersection算子:对两个RDD进行intersection操作，返回一个新的RDD，RDD的分区数与父RDD分区数多的保持一致。
val rdd3: RDD[String] = rdd1.intersection(rdd2)
println(s"rdd3 分区数:${rdd3.getNumPartitions}")rdd3.foreach(println)sc.stop()

二、subtract使用案例

取两个RDD数据集的差集，rdd1.subtract(rdd2):返回rdd1中有但rdd2中没有的元素。

注意：生成RDD的分区数与subtract前面的RDD的分区数一致。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("SubtractTest");
JavaSparkContext sc = new JavaSparkContext(conf);JavaRDD<String> rdd1 = sc.parallelize(Arrays.asList("a","b","c","d"), 3);
JavaRDD<String> rdd2 = sc.parallelize(Arrays.asList("c","d","e","f"), 4);//subtract算子:对两个RDD进行取差集操作，返回一个新的RDD，RDD的分区数与父RDD分区数多的保持一致。
JavaRDD<String> rdd3 = rdd1.subtract(rdd2);
System.out.println("rdd3 分区数：" + rdd3.getNumPartitions());
rdd3.foreach(x-> System.out.println(x));
sc.stop();

Scala代码：

val conf: SparkConf = new SparkConf().setMaster("local").setAppName("SubtractTest")
val sc = new SparkContext(conf)val rdd1: RDD[String] = sc.parallelize(List("a", "b", "c", "d"), 3)
val rdd2: RDD[String] = sc.parallelize(List("c", "d", "e", "f"), 4)//subtract算子:对两个RDD进行取差集操作，返回一个新的RDD，生成RDD的分区数与subtract前面的RDD的分区数一致。
val rdd3: RDD[String] = rdd1.subtract(rdd2)
println(s"rdd3 分区数:${rdd3.getNumPartitions}")rdd3.foreach(println)
sc.stop()