我有型RDD的4rdds:((INT,INT,INT)值)和我的RDDS是加入多个RDDS
rdd1: ((a,b,c), value)
rdd2:((a,d,e),valueA)
rdd3:((f,b,g),valueB)
rdd4:((h,i,c),valueC)
如何加入像RDD1集的RDDS上 “一” RDD1集加入RDD2在“b”上加入rdd2,在“c”上加入rdd1 rdd3
所以在Scala中的输出是finalRdd: ((a,b,c),valueA,valueB,valueC,value))
?
我试着用collectAsMap这样做,但它没有很好地工作,并抛出异常
代码只是为了RDD1集加入RDD2
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),animals,cat,cow,parrot)
你能否写一个更好的例子,数据里面的rdds? –
我添加了一个例子,它不关心anyInt字段上的数字是什么 – luis
rdd1中是否有重复的行?重复密钥? (例如,具有“(1,2,3)”和值“动物”和“另一种动物”的两个元素) –