title: Spark MLLib unable to write out to S3 : path already exists
body: <p>I have data in a S3 bucket in directory <code>/data/vw/</code>. Each line is of the form:</p>
  
  <p><code>| abc:2 def:1 ghi:3 ...</code></p>
  
  <p>I want to convert it to the following format:</p>
  
  <p><code>abc abc def ghi ghi ghi</code></p>
  
  <p>The new converted lines should go to S3 in directory <code>/data/spark</code></p>
  
  <p>Basically, repeat each string the number of times that follows the colon. I am trying to convert a VW LDA input file to a corresponding file for consumption by Spark's LDA library.</p>
  
  <p>The code:</p>
  
  <pre><code>import org.apache.spark.{SparkConf, SparkContext}
  
  object Vw2SparkLdaFormatConverter {
  
    def repeater(s: String): String = {
        val ssplit = s.split(':')
          (ssplit(0) + ' ')  * ssplit(1).toInt
    }
  
    def main(args: Array[String]) {
        val inputPath = args(0)
        val outputPath = args(1)
  
        val conf = new SparkConf().setAppName("FormatConverter")
        val sc = new SparkContext(conf)
  
        val vwdata = sc.textFile(inputPath)
        val sparkdata = vwdata.map(s =&gt; s.trim().split(' ').map(repeater).mkString)
  
        val coalescedSparkData = sparkdata.coalesce(100)
        coalescedSparkData.saveAsTextFile(outputPath)
  
        sc.stop()
    }
  }
  </code></pre>
  
  <p>When I run this (as a Spark EMR job in AWS), the step fails with exception:</p>
  
  <pre><code>18/01/20 00:16:28 ERROR ApplicationMaster: User class threw exception: org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory s3a://mybucket/data/spark already exists
  org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory s3a://mybucket/data/spark already exists
      at org.apache.hadoop.mapred.FileOutputFormat.checkOutputSpecs(FileOutputFormat.java:131)
      at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1.apply$mcV$sp(PairRDDFunctions.scala:1119)
      at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1.apply(PairRDDFunctions.scala:1096)
      at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1.apply(PairRDDFunctions.scala:1096)
      at ...
  </code></pre>
  
  <p>The code is run as:</p>
  
  <pre><code>spark-submit --class Vw2SparkLdaFormatConverter --deploy-mode cluster --master yarn --conf spark.yarn.submit.waitAppCompletion=true --executor-memory 4g s3a://mybucket/scripts/myscalajar.jar s3a://mybucket/data/vw s3a://mybucket/data/spark
  </code></pre>
  
  <p>I have tried specifying new output paths (<code>/data/spark1</code> etc), ensuring that it does not exist before the step is run. Even then it is not working.</p>
  
  <p>What am I doing wrong? I am new to Scala and Spark so I might be overlooking something here.</p>