根据位于另一个 RDD 中的键替换 RDD 中的值

Question

我如何 assign/map 一个 RDD 中的值基于另一个 RDD 中的键

例如开始于：

val rdd1 = sc.parallelize(Seq((1, "A"), (2, "B")))
val rdd2 = sc.parallelize(Seq(("A", "B"), ("B", "B"), ("B", "A")))

我怎样才能得到这样的东西？

(1,2)(2,2)(2,1)

Answer 1

将 rdd1 广播为查找 table（如字典），然后在 rdd2 映射

中使用广播查找 table

rdd1_local = sc.parallelize([(1,"A"),(2,"B")]).map(lambda x:(x[1],x[0])).collectAsMap()
rdd1_broadcast = sc.broadcast(rdd1_local)
rdd2 = sc.parallelize([("A", "B"), ("B", "B"), ("B", "A")])
rdd2.map(lambda x:(rdd1_broadcast.value[x[0]],rdd1_broadcast.value[x[1]])).take(2)

Answer 2

Scala 版本：

val rdd1 = sc.parallelize(Seq((1, "A"), (2, "B")))
val rdd2 = sc.parallelize(Seq(("A", "B"), ("B", "B"), ("B", "A")))

val rdd1_broadcast = sc.broadcast(rdd1.map(x=>(x._2,x._1)).collectAsMap())
val resultRDD = rdd2.map{x=> (rdd1_broadcast.value(x._1),rdd1_broadcast.value(x._2))}