title: Using Spark to write a parquet file to s3 over s3a is very slow
body: <p>I'm trying to write a <code>parquet</code> file out to <code>Amazon S3</code> using <code>Spark 1.6.1</code>.  The small <code>parquet</code> that I'm generating is <code>~2GB</code> once written so it's not that much data.  I'm trying to prove <code>Spark</code> out as a platform that I can use.  </p>
  
  <p>Basically what I'm going is setting up a <code>star schema</code> with <code>dataframes</code>, then I'm going to write those tables out to parquet.  The data comes in from csv files provided by a vendor and I'm using Spark as an <code>ETL</code> platform.  I currently have a 3 node cluster in <code>ec2(r3.2xlarge)</code> So <code>120GB</code> of memory on the executors and 16 cores total.  </p>
  
  <p>The input files total about 22GB and I'm extracting about 2GB of that data for now.  Eventually this will be many terabytes when I start loading the full dataset.</p>
  
  <p>Here is my spark/scala <code>pseudocode</code>:</p>
  
  <pre><code>  def loadStage(): Unit = {
      sc.hadoopConfiguration.set("fs.s3a.buffer.dir", "/tmp/tempData")
      sc.hadoopConfiguration.set("spark.sql.parquet.output.committer.class","org.apache.spark.sql.parquet.DirectParquetOutputCommitter")
      sc.hadoopConfiguration.set("spark.sql.hive.convertMetastoreParquet","false")
      var sqlCtx = new SQLContext(sc)
  
  
      val DataFile = sc.textFile("s3a://my-bucket/archive/*/file*.gz")
  
      //Setup header table/df
      val header_rec = DataFile.map(_.split("\\|")).filter(x=&gt; x(0) == "1")
      val headerSchemaDef = "market_no,rel_date,field1, field2, field3....."
      val headerSchema = StructType(headerSchemaDef.split(",").map(fieldName =&gt; StructField(fieldName, StringType,false)))
      val headerRecords = header_rec.map(p =&gt; Row(p(3), p(8), p(1), p(2), p(4), p(5), p(6) ))
      val header = sqlCtx.createDataFrame(headerRecords, headerSchema)
      header.registerTempTable("header")
      sqlCtx.cacheTable("header")
  
  
      //Setup fact table/df
      val fact_recs = DataFile.map(_.split("\\|")).filter(x=&gt; x(0) == "2")
      val factSchemaDef = "market_no,rel_date,field1, field2, field3....."
      val factSchema = StructType(factSchemaDef.split(",").map(fieldName =&gt; StructField(fieldName, StringType,false)))
      val records = fact_recs.map(p =&gt; Row(p(11), p(12), p(1), p(2), p(3), p(4), p(5), p(6), p(7), p(8), p(9), p(10)))
      val df = sqlCtx.createDataFrame(records, factSchema)
      df.registerTempTable("fact")
  
      val results = sqlCtx.sql("select fact.* from header inner join fact on fact.market_no = header.market_no and fact.rel_date = header.rel_date")
  
  
      println(results.count())
  
  
  
      results.coalesce(1).write.mode(SaveMode.Overwrite).parquet("s3a://my-bucket/a/joined_data.parquet")
  
  
    }
  </code></pre>
  
  <p>The count takes about 2 minutes for 465884512 rows.  The write to parquet takes <strong>38 minutes</strong></p>
  
  <p>I understand that the <code>coalesce</code> does a shuffle to the driver which does the write....  but the amount of time it's taking is making me think I'm doing something seriously wrong.  Without the <code>coalesce</code>, this still takes 15 minutes, which IMO is still too long and gives me a ton of small <code>parquet</code> files.  I'd like to have one large file per day of data that I'll have.  I have code to do the <em>partitioning by a field value</em> as well, and it is just as slow.  I've also tried to output this to <code>csv</code> and that takes ~1 hour.</p>
  
  <p>Also, I'm not really setting run time props when I'm submitting my job. My console stats for one job are:</p>
  
  <ul>
  <li>Alive Workers: 2</li>
  <li>Cores in use: 16 Total, 16 Used</li>
  <li>Memory in use: 117.5 GB Total, 107.5 GB Used</li>
  <li>Applications: 1 Running, 5 Completed</li>
  <li>Drivers: 0 Running, 0 Completed</li>
  <li>Status: ALIVE</li>
  </ul>