地图内的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两次，这是低效的。

请不要将此标记为重复。虽然有类似的问题，但没有一个真正回答这个问题。例如，这个潜在的重复会调用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)