所谓双Value,就是两个数据集RDD之间进行操作。
1 union(otherDataset):对源RDD和参数RDD合并后返回一个新的RDD,不会去重
val listRDD1: RDD[Int] = sc.makeRDD(1 to 5) val listRDD2: RDD[Int] = sc.makeRDD(5 to 10) val union: Array[Int] = listRDD1.union(listRDD2).collect() println(union.mkString(",")) //1,2,3,4,5,5,6,7,8,9,10
2 subtract (otherDataset):计算差的一种函数,去除两个RDD中相同的元素,不同的RDD将保留下来
val listRDD3: RDD[Int] = sc.makeRDD(1 to 8) val listRDD4: RDD[Int] = sc.makeRDD(4 to 10) val subtract: Array[Int] = listRDD3.subtract(listRDD4).collect() println(subtract.mkString(",")) //1,2,3
3 intersection(otherDataset):对源RDD和参数RDD求交集后返回一个新的RDD
val listRDD5: RDD[Int] = sc.makeRDD(1 to 4) val listRDD6: RDD[Int] = sc.makeRDD(2 to 6) val intersection: Array[Int] = listRDD5.intersection(listRDD6).collect() println(intersection.mkString(",")) //4,2,3
4 cartesian(otherDataset):笛卡尔积(尽量避免使用)
val listRDD7: RDD[Int] = sc.makeRDD(1 to 3) val listRDD8: RDD[Int] = sc.makeRDD(4 to 6) val cartesian: Array[(Int, Int)] = listRDD7.cartesian(listRDD8).collect() println(cartesian.mkString(",")) //(1,4),(1,5),(1,6),(2,4),(2,5),(2,6),(3,4),(3,5),(3,6)
5 zip(otherDataset):将两个RDD组合成Key/Value形式的RDD,这里默认两个RDD的partition数量以及元素数量都相同,否则会抛出异常。
val listRDD9: RDD[Int] = sc.makeRDD(1 to 4, 3) val listRDD10: RDD[Char] = sc.makeRDD(List('a', 'b', 'c', 'd'), 3) val zip: Array[(Int, Char)] = listRDD9.zip(listRDD10).collect() println(zip.mkString(",")) //(1,a),(2,b),(3,c),(4,d)