title: Memory efficient way of union a sequence of RDDs from Files in Apache Spark
body: <p>I'm currently trying to train a set of Word2Vec Vectors on the UMBC Webbase Corpus (around 30GB of text in 400 files).</p>
  
  <p>I often run into out of memory situations even on 100 GB plus Machines. I run Spark in the application itself. I tried to tweak a little bit, but I am not able to perform this operation on more than 10 GB of textual data. The clear bottleneck of my implementation is the union of the previously computed RDDs, that where the out of memory exception comes from.</p>
  
  <p>Maybe one you have the experience to come up with a more memory efficient implementation than this:</p>
  
  <pre><code> object SparkJobs {
    val conf = new SparkConf()
      .setAppName("TestApp")
      .setMaster("local[*]")
      .set("spark.executor.memory", "100g")
      .set("spark.rdd.compress", "true")
  
    val sc = new SparkContext(conf)
  
  
    def trainBasedOnWebBaseFiles(path: String): Unit = {
      val folder: File = new File(path)
  
      val files: ParSeq[File] = folder.listFiles(new TxtFileFilter).toIndexedSeq.par
  
  
      var i = 0;
      val props = new Properties();
      props.setProperty("annotators", "tokenize, ssplit");
      props.setProperty("nthreads","2")
      val pipeline = new StanfordCoreNLP(props);
  
      //preprocess files parallel
      val training_data_raw: ParSeq[RDD[Seq[String]]] = files.map(file =&gt; {
        //preprocess line of file
        println(file.getName() +"-" + file.getTotalSpace())
        val rdd_lines: Iterator[Option[Seq[String]]] = for (line &lt;- Source.fromFile(file,"utf-8").getLines) yield {
            //performs some preprocessing like tokenization, stop word filtering etc.
            processWebBaseLine(pipeline, line)    
        }
        val filtered_rdd_lines = rdd_lines.filter(line =&gt; line.isDefined).map(line =&gt; line.get).toList
        println(s"File $i done")
        i = i + 1
        sc.parallelize(filtered_rdd_lines).persist(StorageLevel.MEMORY_ONLY_SER)
  
      })
  
      val rdd_file =  sc.union(training_data_raw.seq)
  
      val starttime = System.currentTimeMillis()
      println("Start Training")
      val word2vec = new Word2Vec()
  
      word2vec.setVectorSize(100)
      val model: Word2VecModel = word2vec.fit(rdd_file)
  
      println("Training time: " + (System.currentTimeMillis() - starttime))
      ModelUtil.storeWord2VecModel(model, Config.WORD2VEC_MODEL_PATH)  
    }}
  }
  </code></pre>