前言
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))))