map 内的Scala Spark过滤器

Question

我想在映射 RDD 时有效地过滤它。这可能吗？

这是我想做的伪代码:

for element in rdd:
    val opt = f(element)
    if (opt.nonEmpty) add_pair(opt.get, element)

这是在 Scala Spark 中实现伪代码的一种 hacky 方法:

rdd.map(element => (
    f(element).getOrElse(99),
    element
)).filter(tuple => tuple._1 != 99)

我无法找到干净的语法来执行此操作，因此我首先映射了所有元素，然后过滤掉了我不想要的元素。请注意，可能昂贵的调用 f(element) 只计算一次。如果我要在映射之前过滤元素(看起来更干净)，那么我最终会调用 f 两次，这是低效的。

请不要将此标记为重复。虽然有类似的问题，但他们都没有真正回答这个问题。例如，this潜在重复将调用 f 两次，效率低下，因此不回答此问题。

Answer 1

你可以只使用 flatMap:

//let's say your f returns Some(x*2) for even number and None for odd
def f(n: Int): Option[Int] = if (n % 2) Some(n*2) else None 

val rdd = sc.parallelize(List(1,2,3,4))
rdd.flatMap(f) // 4,8

// rdd.flatMap(f) or rdd.flatMap(f(_)) or rdd.flatMap(e => f(e))

如果您需要进一步传递元组并进行过滤，则只需使用嵌套 map:

rdd.flatMap(e => f(e).map((_,e))) //(4,2),(8,4)