如何仅在rdd的第二个字段中存在的情况下从一个字段rdd中选择值

Question

我有一个rdd，其中有3个字段，如下所述。

1,2,6
2,4,6
1,4,9
3,4,7
2,3,8

现在，从上面的rdd，我想得到下面的rdd。

2,4,6
3,4,7
2,3,8

结果rdd没有以1开头的行，因为1不在输入rdd的第二个字段中。

Answer 1

好的，如果我正确理解你想做什么，有两种方法：

1. 将您的RDD一分为二，其中第一个RDD包含"第二个字段"的唯一值，第二个RDD以"第一个值"为键。然后把rdd连在一起。这种方法的缺点是CCD_ 2和CCD_。

   val r: RDD[(String, String, Int)] = sc.parallelize(Seq(
     ("1", "2", 6),
     ("2", "4", 6),
     ("1", "4", 9),
     ("3", "4", 7),
     ("2", "3", 8)
   ))
   val uniqueValues: RDD[(String, Unit)] = r.map(x => x._2 -> ()).distinct
   val r1: RDD[(String, (String, String, Int))] = r.map(x => x._1 -> x)
   val result: RDD[(String, String, Int)] = r1.join(uniqueValues).map {case (_, (x, _)) => x}
   result.collect.foreach(println)
   

2. 如果您的RDD相对较小，并且第二个值的Set可以完全适合所有节点的内存，那么您可以在内存中创建该集合作为第一步，将其广播到所有节点，然后只过滤您的RDD:

   val r: RDD[(String, String, Int)] = sc.parallelize(Seq(
     ("1", "2", 6),
     ("2", "4", 6),
     ("1", "4", 9),
     ("3", "4", 7),
     ("2", "3", 8)
   ))
   val uniqueValues = sc.broadcast(r.map(x => x._2).distinct.collect.toSet)
   val result: RDD[(String, String, Int)] = r.filter(x => uniqueValues.value.contains(x._1))
   result.collect.foreach(println)
   

两个示例输出：

(2,4,6)
(2,3,8)
(3,4,7)