加入两个 RDD，其中一个只有键，没有值答案

【问题标题】：Join two RDDs, one of which has only keys and no values加入两个 RDD，其中一个只有键，没有值
【发布时间】：2020-03-28 12:23:17
【问题描述】：

给定两个大型 RDD，a 和一组 (key, value) 对和 b 只有 keys，加入它们的最佳方法是什么，以便 a 只保留那些匹配的行b的键？

更具体地说，这就是我想要做的：

val a: RDD[(Int, Double)] = ...
val b: RDD[Int] = ...
val c: RDD[(Int, Double)] = a.myFilterJoin(b)

其中c 仅包含a 中与b 中的键匹配的行，我们可以假设a 仅包含唯一键。有类似myFilterJoin 的东西吗？

请注意，如果b 足够小，我可以简单地将其作为一组广播，然后将其用作b 上的过滤器。但是让我们假设b 足够大，以至于这个广播非常昂贵。

我通常做的是将一个虚拟变量添加到b，以便b 获得(key, dummy) 的形式以便能够进行连接，然后我在映射中删除虚拟变量。但这似乎很hacky，我想知道是否有更好的解决方案。

【问题讨论】：

标签： scala apache-spark rdd

【解决方案1】：

听起来你应该使用内连接：

  import spark.implicits._

  val a: DataFrame = spark.sparkContext.parallelize(Seq((1, 2.5), (2, 4.4), (3, 2.1))).toDF("keyA", "value")
  val b: DataFrame = spark.sparkContext.parallelize(Seq(3, 5, 1)).toDF("keyB")

  val c = a.join(b, $"keyA" === $"keyB", "inner").drop("keyB")

  c.show()

+----+-----+
|keyA|value|
+----+-----+
|   1|  2.5|
|   3|  2.1|
+----+-----+

如果你希望返回 c 为 RDD[(Int,Double)]，你可以使用：

  val d = c.rdd.map(row => (row.get(0).asInstanceOf[Int], row.get(1).asInstanceOf[Double]))

【讨论】：

【解决方案2】：

类似于 ShemTov 的回答，但通过使用 Datasets 而不是 DataFrames 来保持类型安全。
（PS：我建议您只使用 数据集而不是RDD）

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().master("local[*]").getOrCreate()
val sc = spark.sparkContext
import spark.implicits._

val a = sc.parallelize(List((1 -> 0.0), (2 -> 3.3), (3 -> 5.5), (5 -> 10.11)))
val b = sc.parallelize(List(2, 3, 4, 5))

val c = b.toDS.joinWith(a.toDS, $"value" === $"_1", "inner").map {
  case (_, (key, value)) => key -> value
}.rdd

【讨论】：