[jira] [Commented] (SPARK-1749) DAGScheduler supervisor strategy broken with Mesos

["Bouke van der Bijl (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-1749?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13993602#comment-13993602 ] 

Bouke van der Bijl commented on SPARK-1749:
-------------------------------------------

This isn't really PySpark specific, this works fine on other backends which will mark the task as failed and just keep the SparkContext running.

It shouldn't be shutting down the whole SparkContext just because a single job failed

> DAGScheduler supervisor strategy broken with Mesos
> --------------------------------------------------
>
>                 Key: SPARK-1749
>                 URL: https://issues.apache.org/jira/browse/SPARK-1749
>             Project: Spark
>          Issue Type: Bug
>          Components: Mesos, Spark Core
>    Affects Versions: 1.0.0
>            Reporter: Bouke van der Bijl
>            Assignee: Mark Hamstra
>            Priority: Blocker
>              Labels: mesos, scheduler, scheduling
>
> Any bad Python code will trigger this bug, for example `sc.parallelize(range(100)).map(lambda n: undefined_variable * 2).collect()` will cause a `undefined_variable isn't defined`, which will cause spark to try to kill the task, resulting in the following stacktrace:
> java.lang.UnsupportedOperationException
> 	at org.apache.spark.scheduler.SchedulerBackend$class.killTask(SchedulerBackend.scala:32)
> 	at org.apache.spark.scheduler.cluster.mesos.MesosSchedulerBackend.killTask(MesosSchedulerBackend.scala:41)
> 	at org.apache.spark.scheduler.TaskSchedulerImpl$$anonfun$cancelTasks$3$$anonfun$apply$1.apply$mcVJ$sp(TaskSchedulerImpl.scala:184)
> 	at org.apache.spark.scheduler.TaskSchedulerImpl$$anonfun$cancelTasks$3$$anonfun$apply$1.apply(TaskSchedulerImpl.scala:182)
> 	at org.apache.spark.scheduler.TaskSchedulerImpl$$anonfun$cancelTasks$3$$anonfun$apply$1.apply(TaskSchedulerImpl.scala:182)
> 	at scala.collection.mutable.HashSet.foreach(HashSet.scala:79)
> 	at org.apache.spark.scheduler.TaskSchedulerImpl$$anonfun$cancelTasks$3.apply(TaskSchedulerImpl.scala:182)
> 	at org.apache.spark.scheduler.TaskSchedulerImpl$$anonfun$cancelTasks$3.apply(TaskSchedulerImpl.scala:175)
> 	at scala.Option.foreach(Option.scala:236)
> 	at org.apache.spark.scheduler.TaskSchedulerImpl.cancelTasks(TaskSchedulerImpl.scala:175)
> 	at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages$1.apply$mcVI$sp(DAGScheduler.scala:1058)
> 	at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages$1.apply(DAGScheduler.scala:1045)
> 	at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages$1.apply(DAGScheduler.scala:1045)
> 	at scala.collection.mutable.HashSet.foreach(HashSet.scala:79)
> 	at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1045)
> 	at org.apache.spark.scheduler.DAGScheduler.handleJobCancellation(DAGScheduler.scala:998)
> 	at org.apache.spark.scheduler.DAGScheduler$$anonfun$doCancelAllJobs$1.apply$mcVI$sp(DAGScheduler.scala:499)
> 	at org.apache.spark.scheduler.DAGScheduler$$anonfun$doCancelAllJobs$1.apply(DAGScheduler.scala:499)
> 	at org.apache.spark.scheduler.DAGScheduler$$anonfun$doCancelAllJobs$1.apply(DAGScheduler.scala:499)
> 	at scala.collection.mutable.HashSet.foreach(HashSet.scala:79)
> 	at org.apache.spark.scheduler.DAGScheduler.doCancelAllJobs(DAGScheduler.scala:499)
> 	at org.apache.spark.scheduler.DAGSchedulerActorSupervisor$$anonfun$2.applyOrElse(DAGScheduler.scala:1151)
> 	at org.apache.spark.scheduler.DAGSchedulerActorSupervisor$$anonfun$2.applyOrElse(DAGScheduler.scala:1147)
> 	at akka.actor.SupervisorStrategy.handleFailure(FaultHandling.scala:295)
> 	at akka.actor.dungeon.FaultHandling$class.handleFailure(FaultHandling.scala:253)
> 	at akka.actor.ActorCell.handleFailure(ActorCell.scala:338)
> 	at akka.actor.ActorCell.invokeAll$1(ActorCell.scala:423)
> 	at akka.actor.ActorCell.systemInvoke(ActorCell.scala:447)
> 	at akka.dispatch.Mailbox.processAllSystemMessages(Mailbox.scala:262)
> 	at akka.dispatch.Mailbox.run(Mailbox.scala:218)
> 	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
> 	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> 	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> 	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> 	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> This is because killTask isn't implemented for the MesosSchedulerBackend. I assume this isn't pyspark-specific, as there will be other instances where you might want to kill the task 



--
This message was sent by Atlassian JIRA
(v6.2#6252)