Apache Spark RDD фильтр на два RDD

Ответ 1

Spark не поддерживает это по умолчанию. Фильтрация на одних и тех же данных дважды не так уж плоха, если вы предварительно ее кешируете, а сама фильтрация выполняется быстро.

Если это действительно два разных типа, вы можете использовать вспомогательный метод:

implicit class RDDOps[T](rdd: RDD[T]) {
  def partitionBy(f: T => Boolean): (RDD[T], RDD[T]) = {
    val passes = rdd.filter(f)
    val fails = rdd.filter(e => !f(e)) // Spark doesn't have filterNot
    (passes, fails)
  }
}

val (matches, matchesNot) = sc.parallelize(1 to 100).cache().partitionBy(_ % 2 == 0)

Но как только у вас есть несколько типов данных, просто назначьте фильтр в новый val.

Ответ 2

Spark RDD не имеет такого api.

Вот версия, основанная на тянуть запрос для rdd.span, который должен работать:

import scala.reflect.ClassTag
import org.apache.spark.rdd._

def split[T:ClassTag](rdd: RDD[T], p: T => Boolean): (RDD[T], RDD[T]) = {

    val splits = rdd.mapPartitions { iter =>
        val (left, right) = iter.partition(p)
        val iterSeq = Seq(left, right)
        iterSeq.iterator
    }

    val left = splits.mapPartitions { iter => iter.next().toIterator}

    val right = splits.mapPartitions { iter => 
        iter.next()
        iter.next().toIterator
    }
    (left, right)
}

val rdd = sc.parallelize(0 to 10, 2)

val (first, second) = split[Int](rdd, _ % 2 == 0 )

first.collect
// Array[Int] = Array(0, 2, 4, 6, 8, 10)

Ответ 3

Дело в том, что вы хотите не фильтр, а карту.

(T) -> (Boolean, T)

Извините, я неэффективен в синтаксисе Scala. Но идея состоит в том, что вы разбиваете свой набор ответов, сопоставляя его парам ключ/значение. Ключ может быть логическим значением, указывающим, прошел ли он предикат "Фильтр".

Вы можете контролировать вывод для различных целей, выполняя мудрую обработку разделов. Просто убедитесь, что вы не ограничиваете параллельную обработку только двумя разделами вниз по течению.

См. Также Как разделить СДР на два или более СДР?

Ответ 4

Если вы используете T вместо RDD[T], вы можете сделать это. В противном случае вы могли бы сделать что-то вроде этого:

val data = sc.parallelize(1 to 100)
val splitData = data.mapPartitions{iter => {
    val splitList = (iter.toList).partition(_%2 == 0)
    Tuple1(splitList).productIterator
  }
}.map(_.asInstanceOf[Tuple2[List[Int],List[Int]]])

И тогда вам, вероятно, придется уменьшить это, чтобы объединить списки, когда вы идете выполнить действие

Ответ 5

Вы можете использовать subtract function (Если операция фильтра слишком дорога).

Код PySpark:

rdd1 = data.filter(filterFunction)

rdd2 = data.subtract(rdd1)