Spark RDD关联操作小结

朴灿烈づ我的快乐病毒、 2022-01-06 02:57 288阅读 0赞

## 前言 ##

Spark的rdd之间的关系需要通过一些特定的操作来实现，

操作比较多也，特别是一堆JOIN也挺容易让人产生混乱的。

因此做了下小结梳理一下。

## 准备数据 ##

var rdd1 = sc.makeRDD(Array(("A","a1"),("C","c1"),("D","d1"),("F","f1"),("F","f2")),2)
    var rdd2 = sc.makeRDD(Array(("A","a2"),("C","c2"),("C","c3"),("E","e1")),2)

这两个RDD 有以下几个特征：

*  “A” : rdd1中有rdd2中也有且他们都只有一个
 *  “C”: rdd1中有rdd2中有两个
 *  “D”: rdd1中有rdd2中没有
 *  “E”: rdd1中没有rdd2中有一个
 *  “F”: rdd1中有两个rdd2中没有

## 实验操作 ##

### 1. JOIN ###

类似SQL的inner join操作，返回结果是前面和后面配对成功的，过滤掉关联不上的。

执行结果

scala> rdd1.join(rdd2).collect()
    res5: Array[(String, (String, String))] = Array((A,(a1,a2)), (C,(c1,c2)), (C,(c1,c3)))

**可以看到，结果以左边的Key为准。且是一对多的关系。**

### 2. leftOuterJoin ###

leftOuterJoin类似于SQL中的左外关联left outer join，返回结果以前面的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可。

执行结果

scala> rdd1.leftOuterJoin(rdd2).collect()
    res6: Array[(String, (String, Option[String]))] = Array((F,(f1,None)), (F,(f2,None)), (D,(d1,None)), (A,(a1,Some(a2))), (C,(c1,Some(c2))), (C,(c1,Some(c3))))

**可以看到，其实leftOuterJoin和Join非常类似，只不过Join会直接过滤掉不存在的，而leftOuterJoin会保留值为None。**

### 3. rightOuterJoin ###

同上，只不过这次是以右边为准。

执行结果

scala> rdd1.rightOuterJoin(rdd2).collect()
    res7: Array[(String, (Option[String], String))] = Array((A,(Some(a1),a2)), (C,(Some(c1),c2)), (C,(Some(c1),c3)), (E,(None,e1)))

### 4. subtractByKey ###

返回左边RDD有的Key而右边没有对应的Key。值为左边RDD原有的值。

执行结果

scala> rdd1.subtractByKey(rdd2).collect()
    res9: Array[(String, String)] = Array((D,d1), (F,f1), (F,f2))

**可以看到该操作与值无关。仅仅是过滤一些指定Key。**

### 5. cogroup ###

cogroup相当于SQL中的全外关联full outer join，返回左右RDD中的记录，关联不上的为空。

执行结果

scala> rdd1.cogroup(rdd2).collect()
    res11: Array[(String, (Iterable[String], Iterable[String]))] = Array((F,(CompactBuffer(f1, f2),CompactBuffer())), 
    (D,(CompactBuffer(d1),CompactBuffer())), (A,(CompactBuffer(a1),CompactBuffer(a2))), (C,(CompactBuffer(c1),CompactBuffer(c2, c3))), 
    (E,(CompactBuffer(),CompactBuffer(e1))))