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

大数据Spark（六十七）：Transformation转换算子distinct和mapValues

news 2025/10/8 5:18:46

文章目录

Transformation转换算子distinct和mapValues

一、distinct

二、mapValues

Transformation转换算子distinct和mapValues

一、distinct

用于对RDD数据进行去重，返回一个新的RDD，返回RDD包含原始RDD中所有唯一元素。实际底层distinct的实现为map+reduceByKey+map实现。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("DistinctTest");
JavaSparkContext sc = new JavaSparkContext(conf);JavaRDD<String> rdd = sc.parallelize(Arrays.asList("a", "b", "c", "a", "b", "c"));//如果手动实现去重：mapToPair+reduceByKey+map
JavaPairRDD<String, Integer> rdd1 = rdd.mapToPair(new PairFunction<String, String, Integer>() {@Overridepublic Tuple2<String, Integer> call(String s) throws Exception {return new Tuple2<>(s, 1);}
});
JavaPairRDD<String, Integer> rdd2 = rdd1.reduceByKey(new Function2<Integer, Integer, Integer>() {@Overridepublic Integer call(Integer v1, Integer v2) throws Exception {return v1 + v2;}
});rdd2.map(new Function<Tuple2<String,Integer>, String>() {@Overridepublic String call(Tuple2<String, Integer> v1) throws Exception {return v1._1;}
}).foreach(s->System.out.println(s));//使用distinct算子去重，对源RDD去重后返回一个新的RDD
JavaRDD<String> distinct = rdd.distinct();
distinct.foreach(s->System.out.println(s));
sc.stop();

Scala代码：

val conf = new SparkConf().setMaster("local").setAppName("DistinctTest")
val sc = new SparkContext(conf)val rdd: RDD[String] = sc.parallelize(Array("a","b","c","a","b","c"))
//手动去重，map+reduceByKey+map
rdd.map((_,1)).reduceByKey(_+_).map(t=>t._1).foreach(println)//使用distinct去重
rdd.distinct().foreach(println)sc.stop()

二、mapValues

该函数对K,V格式RDD中的Value按照传入的函数做转换，返回是K，V格式的RDD，Value为新转换后的Value。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("MapValuesTest");
JavaSparkContext sc = new JavaSparkContext(conf);JavaPairRDD<String, Integer> pairRDD = sc.parallelizePairs(Arrays.asList(new Tuple2<String, Integer>("a", 1),new Tuple2<String, Integer>("b", 2),new Tuple2<String, Integer>("c", 3)
));//mapValues算子，只对value进行操作，key保持不变
JavaPairRDD<String, Integer> result = pairRDD.mapValues(new Function<Integer, Integer>() {@Overridepublic Integer call(Integer integer) throws Exception {return integer * 2;}
});result.foreach(tp->System.out.println(tp));sc.stop();

Scala代码：

val conf = new SparkConf().setMaster("local").setAppName("MapValuesTest")
val sc = new SparkContext(conf)val rdd: RDD[(String, Int)] = sc.parallelize(List(("a", 1), ("b", 2), ("c", 3)))
//mapValues:对rdd中value进行操作，key不变
rdd.mapValues(_ * 10).foreach(println)sc.stop()

此外，还有个flatMapValues算子，其与mapValues类似，只是mapValues一对一返回数据，而flatMapValues一对多返回数据。