scala - 如何通过键或过滤器()使用 Spark 交点()和两个RDD？

Question

我想使用 intersection()通过 key 或filter()在 Spark 。

但是我真的不知道怎么用intersection()按键。

所以我尝试使用filter() ，但它不起作用。

示例 - 这是两个 RDD:

data1 //RDD[(String, Int)] = Array(("a", 1), ("a", 2), ("b", 2), ("b", 3), ("c", 1))
data2 //RDD[(String, Int)] = Array(("a", 3), ("b", 5))

val data3 = data2.map{_._1}

data1.filter{_._1 == data3}.collect //Array[(String, Int] = Array()

我想得到一个与 data1 具有相同键的 (key, value) 对基于 data2 的 key 拥有。
Array(("a", 1), ("a", 2), ("b", 2), ("b", 3))是我想要的结果。

有没有办法使用 intersection() 来解决这个问题？通过 key 或 filter() ?

Answer 1

对于您的问题，我认为 cogroup()更适合。 intersection()方法将同时考虑数据中的键和值，并将导致空 rdd .

函数cogroup()将两个 rdd 的值分组的关键，并给我们(key, vals1, vals2) , 其中 vals1和 vals2包含 data1 的值和 data2分别为每个键。请注意，如果某个键未在两个数据集中共享，则 vals1 之一或 vals2将返回为空 Seq ，因此我们首先必须过滤掉这些元组才能到达两个 rdd 的交集的。

接下来，我们将抓取 vals1 - 包含来自 data1 的值用于公共(public)键 - 并将其转换为格式 (key, Array) .最后我们使用 flatMapValues()将结果解压缩为 (key, value) 的格式.

val result = (data1.cogroup(data2)
  .filter{case (k, (vals1, vals2)) => vals1.nonEmpty && vals2.nonEmpty }
  .map{case (k, (vals1, vals2)) => (k, vals1.toArray)}
  .flatMapValues(identity[Array[Int]]))

result.collect()
// Array[(String, Int)] = Array((a,1), (a,2), (b,2), (b,3))