scala - Spark : how to zip an RDD with each partition of the other RDD

Question

假设我有一个 RDD[U]，它总是只包含 1 个分区。我的任务是用位于 n 个分区上的另一个 RDD[T] 的内容填充这个 RDD。最终输出应该是 RDD[U] 的 n 个分区。

我最初尝试做的是:

val newRDD = firstRDD.zip(secondRDD).map{ case(a, b)  => a.insert(b)}

但是我得到一个错误:Can't zip RDDs with unequal numbers of partitions

我可以在 RDD api 中看到 documentation有一个名为 zipPartitions() 的方法。是否有可能，如果可以，如何使用此方法将 RDD[T] 中的每个分区压缩为 RDD[U] 的单个分区并执行像我上面尝试的那样映射它？

Answer 1

像这样的东西应该可以工作:

val zippedFirstRDD = firstRDD.zipWithIndex.map(_.swap)
val zippedSecondRDD = secondRDD.zipWithIndex.map(_.swap)

zippedFirstRDD.join(zippedSecondRDD)
  .map{case (key, (valueU, valueT)) => {
    valueU.insert(valueT)
  }}