Scala Spark: dividere la raccolta in diversi RDD?

Question

C'è qualche funzione Spark che consente di dividere una raccolta in diversi RDD secondo alcuni creteria? Tale funzione consentirebbe di evitare un'eccessiva iterazione. Per esempio:

def main(args: Array[String]) { 
    val logFile = "file.txt" 
    val conf = new SparkConf().setAppName("Simple Application") 
    val sc = new SparkContext(conf) 
    val logData = sc.textFile(logFile, 2).cache() 
    val lineAs = logData.filter(line => line.contains("a")).saveAsTextFile("linesA.txt") 
    val lineBs = logData.filter(line => line.contains("b")).saveAsTextFile("linesB.txt") 
    } 

In questo esempio devo iterare 'logData` due volte solo a dare risultati in due file separati:

val lineAs = logData.filter(line => line.contains("a")).saveAsTextFile("linesA.txt") 
    val lineBs = logData.filter(line => line.contains("b")).saveAsTextFile("linesB.txt") 

Sarebbe bello invece di avere qualcosa di simile:

val resultMap = logData.map(line => if line.contains("a") ("a", line) else if line.contains("b") ("b", line) else (" - ", line) 
    resultMap.writeByKey("a", "linesA.txt") 
    resultMap.writeByKey("b", "linesB.txt") 

Una cosa del genere?

Answer 1

Dai un'occhiata alla seguente domanda.

Write to multiple outputs by key Spark - one Spark job

È possibile flatMap un RDD con una funzione simile alla seguente e poi fare un groupBy sulla chiave.

def multiFilter(words:List[String], line:String) = for { word <- words; if line.contains(word) } yield { (word,line) } 
val filterWords = List("a","b") 
val filteredRDD = logData.flatMap(line => multiFilter(filterWords, line)) 
val groupedRDD = filteredRDD.groupBy(_._1) 

Ma a seconda delle dimensioni del vostro RDD di ingresso si può o non vedere alcun miglioramento delle prestazioni, perché una delle groupBy operazioni comporta un riordino.

D'altra parte se si dispone di memoria sufficiente nel cluster Spark è possibile memorizzare nella cache l'RDD di input e quindi eseguire più operazioni di filtro potrebbe non essere costoso come si pensa.

Answer 2

Forse qualcosa di simile potrebbe funzionare:

def singlePassMultiFilter[T](
     rdd: RDD[T], 
     f1: T => Boolean, 
     f2: T => Boolean, 
     level: StorageLevel = StorageLevel.MEMORY_ONLY 
): (RDD[T], RDD[T], Boolean => Unit) = { 
    val tempRDD = rdd mapPartitions { iter => 
    val abuf1 = ArrayBuffer.empty[T] 
    val abuf2 = ArrayBuffer.empty[T] 
    for (x <- iter) { 
     if (f1(x)) abuf1 += x 
     if (f2(x)) abuf2 += x 
    } 
    Iterator.single((abuf1, abuf2)) 
    } 
    tempRDD.persist(level) 
    val rdd1 = tempRDD.flatMap(_._1) 
    val rdd2 = tempRDD.flatMap(_._2) 
    (rdd1, rdd2, (blocking: Boolean) => tempRDD.unpersist(blocking)) 
} 

Nota che un'azione invitato rdd1 (resp rdd2.) Causerà tempRDD da calcolare e persistente. Questo è praticamente equivalente al calcolo di rdd2 (risp. rdd1) poiché il sovraccarico di flatMap nelle definizioni di rdd1 e di rdd2 è, credo, molto trascurabile.

Si potrebbe utilizzare singlePassMultiFitler in questo modo:

val (rdd1, rdd2, cleanUp) = singlePassMultiFilter(rdd, f1, f2) 
rdd1.persist() //I'm going to need `rdd1` more later... 
println(rdd1.count) 
println(rdd2.count) 
cleanUp(true)  //I'm done with `rdd2` and `rdd1` has been persisted so free stuff up... 
println(rdd1.distinct.count) 

Chiaramente questo potrebbe esteso a un numero arbitrario di filtri, collezioni di filtri, ecc