计算 RDD 中 Array 元素的出现次数

Question

我有一个带有 [(String, Array[String])] 类型键值对的 RDD1（我将它们称为 (X, Y)）和一个数组 Z[String]。 我正在尝试让 Z 中的每个元素计算在 Y 中有多少 X 实例有 Z。我希望我的输出为 ((X, Z(i)), #ofinstances)。

RDD1= ((A, (2, 3, 4), (B, (4, 4, 4)), (A, (4, 5)))
Z = (1, 4)

那我想得到：

(((A, 4), 2), ((B, 4), 1))

希望这是有道理的。 正如你所看到的，如果至少出现一次，我只想要一个元素。

到目前为止我已经尝试过了：

val newRDD = RDD1.map{case(x, y) => for(i <- 0 to (z.size-1)){if(y.contains(z(i))) {((x, z(i)), 1)}}}

我这里的输出是一个 RDD[Unit]

我不确定我所要求的是否可能，或者我是否必须以其他方式进行。

Answer 1

所以这只是另一个字数

val rdd = sc.parallelize(Seq(
   ("A", Array("2", "3", "4")), 
   ("B", Array("4", "4", "4")),
   ("A", Array("4", "5"))))

val z = Array("1", "4")

为了提高查找效率，将z 转换为Set：

val zs = z.toSet

val result = rdd
  .flatMapValues(_.filter(zs contains _).distinct)
  .map((_, 1))
  .reduceByKey(_ + _)

在哪里

_.filter(zs contains _).distinct

过滤掉z 中出现的值并进行重复数据删除。

result.take(2).foreach(println)
// ((B,4),1)
// ((A,4),2)