Support Cancellation of Spark Jobs #665

harsha2010 · 2013-06-28T16:31:08Z

Hi

This patch allows us to cancel runaway Spark queries by issuing a killJob command passing in the jobId. The approach taken here is to let the DAG scheduler clean up its state about a job that is currently executing, and propagate an interrupt to the running task. Each task runs within a Future to properly respond to interrupts and cancel execution. Every Iterator that iterates over data (either via CacheManager/ or BlockFetcher or the HadoopRDD) is wrapped by an InterruptibleIterator which checks for the interrupt and cleans up state accordingly and exits.
I have tested the performance of this patch against master on a bunch of internal queries and the performance is not impacted by this patch.
I am in the process of obtaining TPCH benchmarks with and without the patch which i will attach here.
In the meanwhile, please review and let me know if the design needs changes.

AmplabJenkins · 2013-06-28T16:33:25Z

Thank you for your pull request. An admin will review this request soon.

squito · 2013-06-28T17:58:30Z

core/src/main/scala/spark/InterruptibleIterator.scala

+	extends AnyRef with InterruptibleIterator[T] {
+
+  override def hasNext(): Boolean = {
+    super.hasNext


It seems kinda weird to me that InterruptibleIterator is actually implementing hasNext, which then you override here. Maybe it should have a method exceptionIfThreadInterrupted. It seems like the trait is not actually implementing hasNext at all, its just supplying a utility methods for implementations.

Yes, this seems odd to me, too. Is there some reason why InterruptibleIterator can't be implemented more like CompletionIterator? In other words, class InterruptibleIteratorDecorator just becomes class InterruptibleIterator, which extends Iterator; and your InterruptibleIterator trait becomes object InterruptibleIterator, which defs a function called something like notInterrupted to replace all of the calls to InterruptibleIterator.hasNext -- i.e. the peculiar super.hasNext calls become something like InterruptibleIterator.notInterrupted.

squito · 2013-06-28T18:48:01Z

This looks great! I have wanted this feature for a long time. I made a couple of minor style comments.

I only have one question about the implementation -- it seems like you register the cores as freed as soon as the killTask request gets sent. Do the tasks really die immediately? Should it wait for some acknowledgement that the task really has been killed? I guess its not horrible to have too many tasks running on an executor for a little while, so if it adds a lot of complication for a rare corner case, maybe we can forget it.

harsha2010 · 2013-06-28T19:07:13Z

Thanks for your comments., I'll incorporate them in the update i will add shortly with TPCH performance #s.
The registering cores point is well taken: I couldn't find an easy way to deal with the acknowledgement that a task has been actually killed so implemented the optimistic solution. I'll take a look once more to see if i can do something better.

rxin · 2013-06-28T20:39:23Z

core/src/main/scala/spark/scheduler/Task.scala

+      f.run()
+      f.get()
+    } catch {
+      case e: Exception => throw e.getCause()


Should we be catching Throwable here?

yes, you are right., we should be catching throwable here.

shivaram · 2013-06-29T03:49:15Z

There was one major issue I ran into when I tried to do this before. I am not sure if this still applies:

When canceling tasks that were reading files from HDFS, I noticed that the sockets open in DFSClient would be interrupted and this would lead to datanodes being marked as dead. I had to add a bunch of catch InterruptedException in DFSClient to avoid this before in Hadoop 0.20. I am not sure if this is the case in later versions of Hadoop and this would be good to test.

(As a generalization, I guess we are relying on user-defined classes behaving appropriately when we interrupt a thread ?)

mateiz · 2013-06-29T04:06:38Z

Hey Ram, as I'm looking at this more closely, one question on the Iterator design: why do we need CompletionIterator to also extend InterruptibleIterator? Can't we just catch the InterruptedException in the code that's running the task and run the cleanup procedure there?

mateiz · 2013-06-29T04:08:08Z

(Or more generally, catch it on a call to next or hasNext; basically the on-complete callbacks should be called even if the task throws an exception, so InterruptedException doesn't need to be different).

harsha2010 · 2013-06-29T04:58:03Z

Hey Matei,
Now that i think about it i think you are right: the runInterruptibly methods in all tasks have a finally block that calls oncomplete callbacks. So the interrupted exception does not need to be different.

markhamstra · 2013-07-08T21:26:28Z

core/src/main/scala/spark/CacheManager.scala

@@ -53,7 +55,8 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
          elements ++= rdd.computeOrReadCheckpoint(split, context)
          // Try to put this block in the blockManager
          blockManager.put(key, elements, storageLevel, true)
-          return elements.iterator.asInstanceOf[Iterator[T]]
+            val iter = elements.iterator.asInstanceOf[Iterator[T]]


excess indentation

markhamstra · 2013-07-10T19:20:32Z

After merging this PR into master @ 7dcda9a, CancellationSuite "Cancel Task" is not completing for me.

harsha2010 · 2013-07-10T20:51:54Z

Oh ok will take a look. The test might not have been as well thought out as I imagined, it seems to work on my machine but possibly has a race condition.
Thanks will fix it

Sent from my iPhone

On Jul 10, 2013, at 12:20 PM, Mark Hamstra notifications@github.com wrote:

After merging this PR into master @ 7dcda9a, CancellationSuite "Cancel Task" is not completing for me.

—
Reply to this email directly or view it on GitHub.

markhamstra · 2013-07-19T21:57:54Z

core/src/main/scala/spark/scheduler/DAGScheduler.scala

+    }
+  }
+
+  private def killJob(job: ActiveJob, reason: String) {


@mateiz Doesn't this have the same problem discussed in #414 where more than one ActiveJob can share a stage?

Yes, that is actually true. To do this properly we'll need to do some kind of reference-counting on the stages (keep a list of which jobs currently want to run this stage). One difference here is that killJob is called by the user and for the first use case, of Shark, it's probably going to be fine. But it would be good to either track this properly or send a warning.

That's pretty much the conclusion that I was arriving at. I'll work on the reference-counting refactoring. Should be doable independently of this PR and only require a minimal change here once it is done.

Cool, that would be great to have.

AmplabJenkins · 2013-08-05T21:33:52Z

Thank you for your pull request. An admin will review this request soon.

rxin · 2013-09-17T06:13:50Z

Ram - I am closing this one because it is going to be subsumed by #935.

…in"... ... java.lang.ClassNotFoundException: org.apache.spark.broadcast.TorrentBroadcastFactory Author: witgo <witgo@qq.com> Closes mesos#665 from witgo/SPARK-1734 and squashes the following commits: cacf238 [witgo] SPARK-1734: spark-submit throws an exception: Exception in thread "main" java.lang.ClassNotFoundException: org.apache.spark.broadcast.TorrentBroadcastFactory

Support Cancellation of Spark Jobs

4084227

catch the underlying exception

70f06cc

squito reviewed Jun 28, 2013
View reviewed changes

rxin reviewed Jun 28, 2013
View reviewed changes

markhamstra reviewed Jul 8, 2013
View reviewed changes

markhamstra reviewed Jul 19, 2013
View reviewed changes

rxin mentioned this pull request Sep 17, 2013

job killing in Spark #935

Closed

rxin closed this Sep 17, 2013

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support Cancellation of Spark Jobs #665

Support Cancellation of Spark Jobs #665

harsha2010 commented Jun 28, 2013

AmplabJenkins commented Jun 28, 2013

squito Jun 28, 2013

markhamstra Jul 9, 2013

squito commented Jun 28, 2013

harsha2010 commented Jun 28, 2013

rxin Jun 28, 2013

harsha2010 Jun 29, 2013

shivaram commented Jun 29, 2013

mateiz commented Jun 29, 2013

mateiz commented Jun 29, 2013

harsha2010 commented Jun 29, 2013

markhamstra Jul 8, 2013

markhamstra commented Jul 10, 2013

harsha2010 commented Jul 10, 2013

markhamstra Jul 19, 2013

mateiz Jul 19, 2013

markhamstra Jul 19, 2013

mateiz Jul 20, 2013

AmplabJenkins commented Aug 5, 2013

rxin commented Sep 17, 2013

Support Cancellation of Spark Jobs #665

Support Cancellation of Spark Jobs #665

Conversation

harsha2010 commented Jun 28, 2013

AmplabJenkins commented Jun 28, 2013

Choose a reason for hiding this comment

Choose a reason for hiding this comment

squito commented Jun 28, 2013

harsha2010 commented Jun 28, 2013

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shivaram commented Jun 29, 2013

mateiz commented Jun 29, 2013

mateiz commented Jun 29, 2013

harsha2010 commented Jun 29, 2013

Choose a reason for hiding this comment

markhamstra commented Jul 10, 2013

harsha2010 commented Jul 10, 2013

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AmplabJenkins commented Aug 5, 2013

rxin commented Sep 17, 2013