Spark算子[06]：union，distinct，cartesian，intersection，subtract

输入：

#scala
val rdd1 = sc.parallelize(List(“a”,”b”,”b”,”c”))
val rdd2 = sc.parallelize(List(“c”,”d”,”e”))
——————————————-
#java
JavaRDD rdd1 = sc.parallelize(Arrays.asList(“a”,”b”,”b”,”c”));
JavaRDD rdd2 = sc.parallelize(Arrays.asList(“c”,”d”,”e”));

---

1、union

返回一个新的数据集，该数据集包含源数据集和参数中的元素的联合。不去重

scala版本

scala> val unionRdd = rdd1.union(rdd2)
scala> unionRdd.collect

res2: Array[String] = Array(a, b, b, c, c, d, e)

java版本

JavaRDD<String> res = rdd1.union(rdd2);
res.foreach(x -> System.out.print(x+" "));

# a b b c c d e

---

2、distinct

distinct([numTasks])) 返回包含源数据集去重后元素的新数据集。

scala版本

scala> val distinctRdd = rdd1.distinct(2)
scala> distinctRdd.collect

res3: Array[String] = Array(b, a, c)

java版本

JavaRDD<String> res = rdd1.distinct(2);
res.foreach(x -> System.out.print(x+" "));

# b a c 

---

3、cartesian 笛卡尔

cartesian(otherDataset) 当调用类型T和U的数据集时，返回(T,U)对的数据集(所有对元素)。

scala版本

scala> val cartesainRdd = rdd1.cartesian(rdd2)
scala> cartesainRdd.collect
res5: Array[(String, String)] = Array(
(a,c), (a,d), (a,e), 
(b,c), (b,d), (b,e), 
(b,c), (b,d), (b,e), 
(c,c), (c,d), (c,e)
)

java版本

JavaPairRDD<String,String> res = rdd1.cartesian(rdd2);
res.foreach(x -> System.out.print(x+" "));

#(a,c) (a,d) (a,e) (b,c) (b,d) (b,e) (b,c) (b,d) (b,e) (c,c) (c,d) (c,e) 

---

4、intersection 交集

intersection(otherDataset) 返回一个新的RDD，它包含源数据集中元素和参数的交集。去重

scala版本

scala> val intersectionRdd = rdd1.intersection(rdd2)
scala> intersectionRdd.collect

res6: Array[String] = Array(c)

java版本

JavaRDD<String> res = rdd1.intersection(rdd2);
res.foreach(x -> System.out.print(x+" "));

# c

---

5、subtract 差集

rdd1.subtract (rdd2,2) 返回在rdd1中出现，但是不在rdd2中出现的元素。不去重

scala版本

scala> val subtractRdd = rdd1.subtract(rdd2)
scala> subtractRdd.collect

res7: Array[String] = Array(b, b, a)

java版本

JavaRDD<String> res = rdd1.subtract(rdd2,2);
res.foreach(x -> System.out.print(x+" "));

# b b a