title: Spark - how to write ~20TB of data from a DataFrame to a hive table or hdfs?
body: <p>I am using Spark to process 20TB+ amount of data. 
  I'm trying to write the data into a Hive table, using the following:</p>
  
  <pre><code>df.registerTempTable('temporary_table')
  sqlContext.sql("INSERT OVERWRITE TABLE my_table SELECT * FROM temporary_table")
  </code></pre>
  
  <p>where <code>df</code> is the Spark DataFrame. Unfortunately it doesn't have any dates I can partition over. When I ran the above code, I encountered the error message: </p>
  
  <blockquote>
    <p>py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.sql.execution.EvaluatePython.takeAndServe.
    : org.apache.spark.SparkException: Job aborted due to stage failure: Total size of serialized results of 95561 tasks (1024.0 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)</p>
  
  <pre><code>at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1433)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1421)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1420)
  at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
  at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
  at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1420)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:801)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:801)
  at scala.Option.foreach(Option.scala:236)
  at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:801)
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1642)
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1601)
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1590)
  at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
  at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:622)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1831)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1844)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1857)
  at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:212)
  at org.apache.spark.sql.execution.EvaluatePython$$anonfun$takeAndServe$1.apply$mcI$sp(python.scala:126)
  at org.apache.spark.sql.execution.EvaluatePython$$anonfun$takeAndServe$1.apply(python.scala:124)
  at org.apache.spark.sql.execution.EvaluatePython$$anonfun$takeAndServe$1.apply(python.scala:124)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:56)
  at org.apache.spark.sql.DataFrame.withNewExecutionId(DataFrame.scala:2087)
  at org.apache.spark.sql.execution.EvaluatePython$.takeAndServe(python.scala:124)
  at org.apache.spark.sql.execution.EvaluatePython.takeAndServe(python.scala)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:498)
  at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
  at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
  at py4j.Gateway.invoke(Gateway.java:259)
  at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
  at py4j.commands.CallCommand.execute(CallCommand.java:79)
  at py4j.GatewayConnection.run(GatewayConnection.java:209)
  at java.lang.Thread.run(Thread.java:745)
  </code></pre>
  </blockquote>
  
  <p>The error message seems to also depends on the amount of data. With slightly smaller data, I encountered the following error message</p>
  
  <blockquote>
    <p>Map output statuses were 395624469 bytes which exceeds spark.akka.frameSize (134217728 bytes). </p>
  </blockquote>
  
  <p>What's a more practical way to achieve this (if the task is feasible)? I'm using Spark 1.6.</p>
  
  <p>Below are the config variables when submitting the spark job:
  <code>
  spark-submit --deploy-mode cluster --master yarn 
  --executor-memory 20G 
  --num-executors 500 
  --driver-memory 64g
  --driver-cores 8
  --files 'my_script.py'
  </code></p>
  
  <p>BTW, naively I would imagine that when the write operations happen, Spark will write data from the executors to hdfs. But the error message seems to imply that there are some data transfers between the executors and the driver?  </p>
  
  <p>I only have shallow knowledge in Spark so please pardon me for the dumb questions!</p>