Scala Spark фильтр внутри карты

Question

Я хочу эффективно фильтровать СДР при сопоставлении его.Это возможно?

Вот псевдокод для того, что я хочу сделать:

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

Вот хакерский способ реализации псевдокода в Scala Spark:

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)

Answer 2

Вы можете использовать mapPartitions, чтобы сделать фильтр вместе с вашими дорогостоящими вычислениями.

rdd.mapPartitions( elements => 
  elements
      .map(element => (f(element),element))
      .filter(tuple => tuple._1.isDefined)
)

Обратите внимание, что в этом коде filter - это собственный метод сбора Scala, а не фильтр Spark RDD.

Кроме того, вы также можете flatMap результат вашей функции

rdd.flatMap(element => f(element).map(result => (result,element)))