title: org.apache.spark.SparkException: Job aborted due to stage failure - OOM Exception
body: <p>In my application I'm scooping a table of 5million rows and 151 columns using spark partitioning like below and persisting it to DISK_ONLY</p>
  
  <pre><code>   val query = "(select * from destinationlarge) as dest"
  val options = Map(
    "url" -&gt; "jdbc:mysql://IPADDRESS:3306/test?useSSL=false",
    "driver" -&gt; "com.mysql.jdbc.Driver",
    "dbtable" -&gt; query,
    "user" -&gt; "root",
    "password" -&gt; "root")
  
  val destination = spark.read.options(options).jdbc(options("url"), options("dbtable"), "0", 1, 5, 4, new java.util.Properties()).rdd.map(_.mkString(",")).persist(StorageLevel.DISK_ONLY)
  </code></pre>
  
  <p>The cluster is having 5 datanodes and 1 namenode of hardware configuration i3 4 cores and 4 GB RAM each, after sometime of execution one of the executor is dead and throwing the below ERROR</p>
  
  <pre><code>Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 6, datanode5, executor 6): ExecutorLostFailure (executor 6 exited caused by one of the running tasks) Reason: Executor heartbeat timed out after 139401 ms
  Driver stacktrace:
  at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1435)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1423)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1422)
  at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
  at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
  at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1422)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:802)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:802)
  at scala.Option.foreach(Option.scala:257)
  at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:802)
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1650)
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1605)
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1594)
  at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
  at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:628)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1925)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1938)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1951)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1965)
  at org.apache.spark.rdd.RDD.count(RDD.scala:1158)
  at com.syntel.spark.sparkDVT$.main(sparkDVT.scala:68)
  at com.syntel.spark.sparkDVT.main(sparkDVT.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:497)
  at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:750)
  at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:187)
  at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:212)
  at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:126)
  at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
  </code></pre>
  
  <p><code>lowerbound=1</code>,<code>upperbound=5</code>,<code>number of partitions is 4</code> suggested in this link (<a href="https://www.dezyre.com/article/how-data-partitioning-in-spark-helps-achieve-more-parallelism/297" rel="nofollow noreferrer">https://www.dezyre.com/article/how-data-partitioning-in-spark-helps-achieve-more-parallelism/297</a>) total number of cores is equal to number of partitions that is 4 cores in all the nodes so 4 partitions.</p>
  
  <p><code>spark-submit</code></p>
  
  <p><code>spark-submit --class "com.syntel.spark.sparkDVT" --master yarn --jars --executor-memory 512m --executor-cores 1 --num-executors 5 /root/sparkdvtmysql_2.11-1.0.jar</code></p>
  
  <p>Correct me if I'm wrong</p>
  
  <p>Thanks </p>