spark transform系列__join

Join

Join包含有join,fullOuterJoin,leftOuterJoin,rightOuterJoin四种实现,
下面先根据join的实现流程进行分析,再对join分析的结果的基础上,最后根据后面三种join的实现,说明下这几种JOIN的差别所在.
Join的函数中,需要一个Partitioner的实例,默认是HashPartitioner,join的实现中,是直接通过调用 的cogroup函数来进行的实现,因此,要分析join的流程时,我们需要先去了解下cogroup的流程实现;

Join函数,
在进行join操作时,需要先定义W的类型(要合并的另一个RDD的value的类型)
def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))] = self.withScope {

this.cogroup(other, partitioner).flatMapValues( pair =>

for (v <- pair._1.iterator; w <- pair._2.iterator) yield (v, w)

)

}

在join操作的函数中,首先是先执行了cogroup的操作,这个操作把两个RDD中相同的key对应的value集合生成为一个Pair.
接下来,通过flatMapValues操作,对相同的key的value部分进行迭代,其实一个key在flatMapValues时只包含一个value.针对这个value执行下面这个for的迭代.
for (v <- pair._1.iterator; w <- pair._2.iterator) yield (v, w)
在上面的for的迭代中,需要满足CoGroupedRDD依赖的上层的两个RDD的返回结果的iterator都存在值,
这个for的迭代返回的集合(yield表示对迭代结果记录到集合中,这个集合就是flatMapValues的结果集合),迭代按左右的pair._1的集合为基准,集合总长为为Pair._1.length*Pair._2.length.

结果示例代码:
val e = List("a","df","f")

val f = List("b","c")
val r = for(e1 <- e.toIterator; f1 <- f.toIterator) yield (e1,f1)

r.foreach(println)
上面的代码的输出结果:
(a,b)
(a,c)
(df,b)
(df,c)
(f,b)
(f,c)

LeftOuterJoin函数,
在leftOuterJoin的函数中,与join函数的区别是,join是两个RDD中相同的key的value必须都包含,
而在leftOuterJoin的函数中,下面的代码部分:
这里,先判断,如果第二个RDD的结果是空是,返回第一个RDD的结果,第二个RDD的点位的部分使用None来替代.leftOuterJoin也是按左边RDD为基准,但右边RDD可以没有对应的值.
if (pair._2.isEmpty) {

pair._1.iterator.map(v => (v, None))

}
def leftOuterJoin[W](

other: RDD[(K, W)],

partitioner: Partitioner): RDD[(K, (V, Option[W]))] = self.withScope {

this.cogroup(other, partitioner).flatMapValues { pair =>
如果右边的RDD的结果为空,不能使用下面的嵌套for迭代.直接返回第一个rdd的结果.与join的区别就在这个地方.永远包含左边的RDD的结果.

if (pair._2.isEmpty) {

pair._1.iterator.map(v => (v, None))

} else {

for (v <- pair._1.iterator; w <- pair._2.iterator) yield (v, Some(w))

}

}

}

rightOuterJoin函数,
rightOuterJoin与leftOuterJoin的结果刚好是一个相反的结果,返回的结果集中,第二个RDD的结果集总是包含.如果两个RDD都存在结果集时,还是按第一个RDD为基准结果集.
def rightOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner)

: RDD[(K, (Option[V], W))] = self.withScope {

this.cogroup(other, partitioner).flatMapValues { pair =>
如果左边的RDD的结果为空,不能使用下面的嵌套for迭代.直接返回第一个rdd的结果.与join的区别就在这个地方.永远包含右边的RDD的结果.

if (pair._1.isEmpty) {

pair._2.iterator.map(w => (None, w))

} else {

for (v <- pair._1.iterator; w <- pair._2.iterator) yield (Some(v), w)

}

}

}

fullOuterJoin函数,
这个函数相当于是leftOuterJoin与rightOuterJoin的组合体.如果两个RDD都存在结果集时,还是按第一个RDD为基准结果集.
def fullOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner)

: RDD[(K, (Option[V], Option[W]))] = self.withScope {

this.cogroup(other, partitioner).flatMapValues {
第二个RDD没有结果集时,返回左边的结果集

case (vs, Seq()) => vs.iterator.map(v => (Some(v), None))
第一个RDD没有结果集时,返回右边的结果集

case (Seq(), ws) => ws.iterator.map(w => (None, Some(w)))
两个RDD都包含结果集时,通过左边的结果集为基准,返回结果集的组合.

case (vs, ws) => for (v <- vs.iterator; w <- ws.iterator) yield (Some(v), Some(w))

}

}