加入多个rdds

Question

我有4个rdd类型的RDD:((int，int，int)，values)，我的rdds是

rdd1: ((a,b,c), value) 
rdd2:((a,d,e),valueA) 
rdd3:((f,b,g),valueB)
rdd4:((h,i,c),valueC) 

如何加入RDD，比如rdd1加入“A”上的rdd2；rdd1加入“B”上的rdd2；rdd1加入“C”上的rdd3

那么在Scala中的输出是finalrdd:((a,b,c）,valueA,valueB,valueC,value))？

val newrdd2=rdd2.map{case( (a,b,c),d)=>(a,d)}.collectAsMap
val joined=rdd1.map{case( (a,b,c),d)=>(newrdd2.get(a).get,b,c,d)} 

例

rdd1: ((1,2,3),animals)
rdd2:((1,anyInt,anyInt),cat)
rdd3:((anyInt,2,anyInt),cow )
rdd 4: ((anyInt,anyInt,3),parrot)

输出应该是((1,2,3）,动物，猫，牛，鹦鹉）

Answer 1

RDDs上有一个方便的join方法，但是您需要用特定的join键来键控它，这是Spark用于分区和洗牌的。

从文档中：

join（otherDataset,[numTasks]):当调用类型为(K，V)和(K，W)的数据集时，返回一个由(K，(V，W))对组成的数据集，每个键都有所有元素对。外部连接通过leftOuterJoin、rightOuterJoin和fullouterjoin得到支持。

val rdd1KeyA = rdd1.map(x => (x._1._1, (x._1._2, x._1._3. x._2) // RDD(a, (b,c,value))
val rdd2KeyA = rdd2.map(x => (x._1._1, x._2) // RDD(a, valueA)
val joined1 = rdd1KeyA.join(rdd2KeyA) // RDD(a, ((b,c,value), valueA))

val rdd3KeyB = rdd3.map(x => (x._1._2, x._2) // RDD(b, valueB)
val joined1KeyB = joined1.map(x => (x._2._1._1, (x._1, x._2._1._2, x._2._1._3. x._2._2) // RDD(b, (a, c, value, valueA))
val joined2 = joined1KeyB.join(rdd3keyB) // RDD(b, ((a, c, value, valueA), valueB))