title: How to optimize Spark for writing large amounts of data to S3
body: <p>I do a fair amount of ETL using Apache Spark on EMR.</p>
  
  <p>I'm fairly comfortable with most of the tuning necessary to get good performance, but I have one job that I can't seem to figure out.</p>
  
  <p>Basically, I'm taking about 1 TB of parquet data - spread across tens of thousands of files in S3 - and adding a few columns and writing it out partitioned by one of the date attributes of the data - again, parquet formatted in S3.</p>
  
  <p>I run like this:</p>
  
  <pre><code>spark-submit --conf spark.dynamicAllocation.enabled=true  --num-executors 1149 --conf spark.driver.memoryOverhead=5120 --conf  spark.executor.memoryOverhead=5120 --conf  spark.driver.maxResultSize=2g --conf spark.sql.shuffle.partitions=1600 --conf spark.default.parallelism=1600 --executor-memory 19G --driver-memory 19G --executor-cores 3 --driver-cores 3 --class com.my.class path.to.jar &lt;program args&gt;
  </code></pre>
  
  <p>The size of the cluster is dynamically determined based on the size of the input data set, and the num-executors, spark.sql.shuffle.partitions, and spark.default.parallelism arguments are calculated based on the size of the cluster.</p>
  
  <p>The code roughly does this:</p>
  
  <pre><code>va df = (read from s3 and add a few columns like timestamp and source file name)
  
  val dfPartitioned = df.coalesce(numPartitions)
  
  val sqlDFProdDedup = spark.sql(s""" (query to dedup against prod data """);
  
  sqlDFProdDedup.repartition($"partition_column")
    .write.partitionBy("partition_column")
    .mode(SaveMode.Append).parquet(outputPath)
  </code></pre>
  
  <p>When I look at the ganglia chart, I get a huge resource spike while the de-dup logic runs and some data shuffles, but then the actual writing of the data only uses a tiny fraction of the resources and runs for several hours.</p>
  
  <p>I don't think the primary issue is partition skew, because the data should be fairly distributed across all the partitions. </p>
  
  <p>The partition column is essentially a day of the month, so each job typically only has 5-20 partitions, depending on the span of the input data set. Each partition typically has about 100 GB of data across 10-20 parquet files.</p>
  
  <p>I'm setting spark.sql.files.maxRecordsPerFile to manage the size of those output files.</p>
  
  <p>So, my big question is: how can I improve the performance here?</p>
  
  <p>Simply adding resources doesn't seem to help much.</p>
  
  <p>I've tried making the executors larger (to reduce shuffling) and also to increase the number of CPUs per executor, but that doesn't seem to matter.</p>
  
  <p>Thanks in advance!</p>