java - Spark : Cogroup RDDs fails in case of huge group

Question

下午好!我有一个问题:

val rdd1: RDD[(key, value)] = ...
val rdd2: RDD[(key, othervalue)] = ...

我要过滤rdd1并丢弃所有不在 rdd2 中的元素。我知道有两种方法可以做到这一点。

第一:

val keySet = rdd2.map(_.key).distinct().collect().toSet
rdd1.filter(x => keySet.contains(x))

它不起作用，因为 keySet太大，无法装入内存。

另一个:

rdd1.cogroup(rdd2)
  .filter(x => x._2._2.nonEmpty)
  .flatMap(x => x._2._1)

这里发生了一些事情，我收到两种错误(在代码的不同位置):java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE和java.lang.ArrayIndexOutOfBoundException

我认为这是因为我的组太大了。

那么我该如何解决这个问题呢？有没有通用的方法来解决这个问题？

Answer 1

您是否考虑过使用subtractByKey ？

类似的东西

rdd1.map(x => (x, x))
    .subtractByKey(rdd2)
    .map((k,v) => k)