apache/spark — pull request #2482 by JoshRosen (2014-09-21 19:28:19+00:00)
Title: [SPARK-3626] [WIP] Replace AsyncRDDActions with a more general runAsync() mechanism

### Background

The `AsyncRDDActions` methods were introduced in e33b1839e27249e232a2126cec67a38109e03243, the first pull request to add support for job cancelation. A follow pull request, 599dcb0ddf740e028cc8faac163303be8f9400a6, added cancelation on a per-job-group basis. Quoting from that PR:

> This PR adds a simple API to group together a set of jobs belonging to a thread and threads spawned from it. It also allows the cancellation of all jobs in this group.
>
> An example:
>
> sc.setJobDescription("this_is_the_group_id", "some job description")
> sc.parallelize(1 to 10000, 2).map { i => Thread.sleep(10); i }.count()
>
> In a separate thread:
>
> sc.cancelJobGroup("this_is_the_group_id")

In its current form, `AsyncRDDActions` seems to serve no purpose other than to enable job cancelation. `AsyncRDDActions` is marked as `@Experimental`, so users may be reluctant to depend on it. If we add new actions, then we also have to add asynchronous versions of those actions, creating a maintenance burden.

### Proposal

I propose that we remove `AsyncRDDActions` and use job groups as the only user-facing API for job cancelation. To make job groups more convenient for users, this pull request adds a `runAsync` method to SparkContext that makes it easy to run an asynchronous computation in a particular Spark job group. For example:

```scala
// Instead of countAsync(), we call the regular actions from a runAsync block:
val futureCount: RunAsyncResult[Long] = sc.runAsync {
sc.parallelize(...).map(...).count()
}

// This returns a Future:
futureCount.onSuccess(c => println(s"Got count $c!"))

// The future also supports cancellation
futureCount.cancel()

// This also works with blocks that call multiple actions (e.g. an iterative ML algorithm)
val futureResult = sc.runAsync {
val rdd = sc.parallelize(...)
val count = rdd.count()
val first = rdd.first()
first // this is the result of the block
}
```

I refactored `JobCancellationSuite` to use this new API instead of AsyncRDDActions; see that file for more examples.

### TODOS / tasks to finish before merging:

This is marked as [WIP] since there are still a number of tasks that need to be finished before this is merge-worthy:

- [ ] Add a Java interface for this; it can probably be an extension of `Runnable`.
- [ ] Extend `RunAsyncResult` to expose the job ids of jobs launched from its computation (see [SPARK-3446](https://issues.apache.org/jira/browse/SPARK-3446) / #2337).
- [ ] The thread-safety concerns here are somewhat complicated; add more comments.
- [ ] Explain the caveats with job-group-based cancellation:
- Having multiple threads submitting jobs in the same job group may lead to confusing behavior.
- Recursive runAsync() calls may have undefined / confusing behavior.
- [ ] Look through AsyncRDDActionsSuite to see whether there are any tests that need to be re-added for these APIs.
- [ ] Add example uses in the Spark examples subproject, since it may not be obvious how to use this.