RDD的join和Dstream的join有什么区别？
RDD的join和Dstream的join有什么区别？
有人在知识星球里问：
浪院长，RDD的join和Dstream的join有什么区别？
浪尖的回答：
DStream的join底层就是rdd的join。
下面，我们就带着疑问去验证以下，我们的想法。
2. DStream -> PairDStreamFunctions
Dstream这个类实际上支持的只是Spark Streaming的基础操作算子，比如： map, filter 和window.PairDStreamFunctions 这个支持key-valued类型的流数据
，支持的操作算子，如，groupByKeyAndWindow，join。这些操作，在有key-value类型的流上是自动识别的。
对于dstream -> PairDStreamFunctions自动转换的过程大家肯定想到的是scala的隐式转换。具体代码在Dstream的object内部。
implicit def toPairDStreamFunctions[K, V](stream: DStream[(K, V)])
(implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K] = null):
PairDStreamFunctions[K, V] = {
new PairDStreamFunctions[K, V](stream)
}假如，你对scala的隐式转换比较懵逼，请阅读下面文章。
3. PairDStreamFunctions的join
PairDStreamFunctions的join API总共有三种
/**
* Return a new DStream by applying 'join' between RDDs of `this` DStream and `other` DStream.
* Hash partitioning is used to generate the RDDs with Spark's default number of partitions.
*
* 通过join this和other Dstream的rdd构建出一个新的DStream.
* Hash分区器，用来使用默认的分区数来产生RDDs。
*/
def join[W: ClassTag](other: DStream[(K, W)]): DStream[(K, (V, W))] = ssc.withScope {
join[W](other, defaultPartitioner())
}
/**
* Return a new DStream by applying 'join' between RDDs of `this` DStream and `other` DStream.
* Hash partitioning is used to generate the RDDs with `numPartitions` partitions.
*
* 通过join this和other Dstream的rdd构建出一个新的DStream.
* Hash分区器，用来使用numPartitions分区数来产生RDDs。
*/
def join[W: ClassTag](
other: DStream[(K, W)],
numPartitions: Int): DStream[(K, (V, W))] = ssc.withScope {
join[W](other, defaultPartitioner(numPartitions))
}
/**
* Return a new DStream by applying 'join' between RDDs of `this` DStream and `other` DStream.
* The supplied org.apache.spark.Partitioner is used to control the partitioning of each RDD.
* 通过join this和other Dstream的rdd构建出一个新的DStream.
* 使用org.apache.spark.Partitioner来控制每个RDD的分区。
*/
def join[W: ClassTag](
other: DStream[(K, W)],
partitioner: Partitioner
): DStream[(K, (V, W))] = ssc.withScope {
self.transformWith(
other,
(rdd1: RDD[(K, V)], rdd2: RDD[(K, W)]) => rdd1.join(rdd2, partitioner)
)
}上面所示代码中，第三个PairDStreamFunctions的join api 体现了join的操作，也即是函数：
(rdd1: RDD[(K, V)], rdd2: RDD[(K, W)]) => rdd1.join(rdd2, partitioner)上面是两个RDD的join过程，并且指定了分区器。后面我们主要是关注该函数封装及调用。
其实，看过浪尖的Spark Streaming的视频的朋友或者度过浪尖关于Spark Streaming相关源码讲解的朋友应该有所了解的是。 这个生成RDD的函数应该是在 DStream的compute方法中在生成RDD的时候调用。假设你不了解也不要紧。 我们跟着代码轨迹前进，验证我们的想法。
DStream.transformWith
/**
* Return a new DStream in which each RDD is generated by applying a function
* on each RDD of 'this' DStream and 'other' DStream.
*/
def transformWith[U: ClassTag, V: ClassTag](
other: DStream[U], transformFunc: (RDD[T], RDD[U]) => RDD[V]
): DStream[V] = ssc.withScope {
// because the DStream is reachable from the outer object here, and because
// DStreams can't be serialized with closures, we can't proactively check
// it for serializability and so we pass the optional false to SparkContext.clean
val cleanedF = ssc.sparkContext.clean(transformFunc, false)
transformWith(other, (rdd1: RDD[T], rdd2: RDD[U], time: Time) => cleanedF(rdd1, rdd2))
}
/**
* Return a new DStream in which each RDD is generated by applying a function
* on each RDD of 'this' DStream and 'other' DStream.
*/
def transformWith[U: ClassTag, V: ClassTag](
other: DStream[U], transformFunc: (RDD[T], RDD[U], Time) => RDD[V]
): DStream[V] = ssc.withScope {
// because the DStream is reachable from the outer object here, and because
// DStreams can't be serialized with closures, we can't proactively check
// it for serializability and so we pass the optional false to SparkContext.clean
val cleanedF = ssc.sparkContext.clean(transformFunc, false)
val realTransformFunc = (rdds: Seq[RDD[_]], time: Time) => {
assert(rdds.length == 2)
val rdd1 = rdds(0).asInstanceOf[RDD[T]]
val rdd2 = rdds(1).asInstanceOf[RDD[U]]
cleanedF(rdd1, rdd2, time)
}
new TransformedDStream[V](Seq(this, other), realTransformFunc)
}经过上面两个 TransformWith操作，最终生成了一个TransformedDStream。需要关注的是new TransformedDStream[V](Seq(this, other), realTransformFunc) 第一个参数是一个包含要进行join操作的两个流的Seq。
那么，TransformedDStream 的parents 就包含了两个流。我们可以看到其 compute 方法的第一行。
override def compute(validTime: Time): Option[RDD[U]] = {
// 针对每一个流，获取其当前时间的RDD。
val parentRDDs = parents.map { parent => parent.getOrCompute(validTime).getOrElse(
// Guard out against parent DStream that return None instead of Some(rdd) to avoid NPE
throw new SparkException(s"Couldn't generate RDD from parent at time $validTime"))
}compute的第一行就是获取parent中每个流，当前有效时间的RDD。然后调用，前面步骤封装的函数进行join。
val transformedRDD = transformFunc(parentRDDs, validTime)以上就是join的全部过程。也是，验证浪尖所说的，DStream的join底层就是RDD的join。
腾讯云开发者
