CODEWITHSUNDEEP
scalaapache-sparkapache-spark-sql
Pyd
Pyd

Reputation: 6159

org.apache.spark.SparkException: Job aborted due to stage failure - OOM Exception

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

   val query = "(select * from destinationlarge) as dest"
val options = Map(
  "url" -> "jdbc:mysql://IPADDRESS:3306/test?useSSL=false",
  "driver" -> "com.mysql.jdbc.Driver",
  "dbtable" -> query,
  "user" -> "root",
  "password" -> "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)

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

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)

lowerbound=1,upperbound=5,number of partitions is 4 suggested in this link (https://www.dezyre.com/article/how-data-partitioning-in-spark-helps-achieve-more-parallelism/297) total number of cores is equal to number of partitions that is 4 cores in all the nodes so 4 partitions.

spark-submit

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

Correct me if I'm wrong

Thanks

Upvotes: 1

Views: 3556

Answers (1)

mrsrinivas
mrsrinivas

Reputation: 35444

I would recommend you use DataFame(in Spark 2.0 i.e DataSet[Row]) as is, because DataSet uses Encoders so that it will have very little memory footprint than RDD. 

val destination = spark.read
    .options(options)
    .format("jdbc")
    .load()

If you want concat columns by delimiter you can use concat_ws() - example here

destination
  .withColumn("column", concat_ws(", ", 
     destination.columns.map(destination.col(_)).toSeq : _*))
  .select("id, column") // id will be used for subtraction with other df
  .persist(StorageLevel.DISK_ONLY)

Check this SO post - Comaparing RDD/DF/DS which you the idea of how Dataset diffrent from RDD and it's advantages.

This may not answer your question entirely. I will update the asnwer as per my comment response

Upvotes: 4

Related Questions