Scala Spark: Split collection into several RDD?
apache-spark, scala
Solution
Have a look at the following question.
Write to multiple outputs by key Spark - one Spark job
You can `flatMap` an RDD with a function like the following and then do a `groupBy` on the key.
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)
But depending on the size of your input RDD you may or not see any performance gains because any of `groupBy` operations involves a shuffle.
On the other hand if you have enough memory in your Spark cluster you can cache the input RDD and therefore running multiple filter operations may not be as expensive as you think.
Problem
Is there any Spark function that allows to split a collection into several RDDs according to some creteria? Such function would allow to avoid excessive itteration. For example: ``` 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 this example I have to iterate 'logData` twice just to write results in two separate files: ``` val lineAs = logData.filter(line => line.contains("a")).saveAsTextFile("linesA.txt") val lineBs = logData.filter(line => line.contains("b")).saveAsTextFile("linesB.txt") ``` It would be nice instead to have something like this: ``` 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") ``` Any such thing?