[SPARK-3902] [SPARK-3590] Stabilize AsynRDDActions and add Java API #2760

JoshRosen · 2014-10-11T05:34:08Z

This PR adds a Java API for AsyncRDDActions and promotes the API from @Experimental to stable.

JoshRosen · 2014-10-11T05:36:29Z

/cc's for review:

@rxin, who wrote the original AsyncRDDActions
@lirui-intel, who added an experimental Java API for foreachAsync in SPARK-2636: Expose job ID in JobWaiter API #2176, and
@vanzin, who added the jobIds method to expose job ids from FutureAction in [SPARK-3446] Expose underlying job ids in FutureAction. #2337.

JoshRosen · 2014-10-11T05:37:15Z

core/src/main/java/org/apache/spark/api/java/JavaFutureAction.java

+import java.util.List;
+import java.util.concurrent.Future;
+
+public interface JavaFutureAction<T> extends Future<T> {


I think that it makes sense to expose an extended version of the Java Future API to users, since there may be a number of existing libraries for consuming these standard future types.

SparkQA · 2014-10-11T05:39:44Z

QA tests have started for PR 2760 at commit ff28e49.

This patch merges cleanly.

JoshRosen · 2014-10-11T05:40:22Z

core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala

   */
-  @Experimental
-  def foreachAsync(f: VoidFunction[T]): FutureAction[Unit] = {


Unfortunately, my PR breaks compatibility for this experimental Java API. However, the previous version of this method hasn't been shipped in any Spark releases yet.

yea i think this is fine

SparkQA · 2014-10-11T05:55:00Z

QA tests have started for PR 2760 at commit 6f8f6ac.

This patch merges cleanly.

lirui-apache · 2014-10-11T06:13:31Z

Looks great! I think it's very useful to have these async APIs in java :-)

SparkQA · 2014-10-11T06:41:56Z

QA tests have finished for PR 2760 at commit ff28e49.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class JavaFutureActionWrapper[S, T](futureAction: FutureAction[S], converter: S => T)

AmplabJenkins · 2014-10-11T06:41:59Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21628/Test PASSed.

SparkQA · 2014-10-11T07:00:38Z

QA tests have finished for PR 2760 at commit 6f8f6ac.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class JavaFutureActionWrapper[S, T](futureAction: FutureAction[S], converter: S => T)

AmplabJenkins · 2014-10-11T07:00:41Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21630/Test PASSed.

rxin · 2014-10-13T20:39:50Z

core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala

@@ -17,6 +17,7 @@

 package org.apache.spark.api.java

+import java.util


an extra import?

Good catch; fixed.

JoshRosen · 2014-10-14T18:39:29Z

@rxin I removed that extra import. Any other comments or things you'd like to discuss?

SparkQA · 2014-10-14T18:44:49Z

QA tests have started for PR 2760 at commit 7a1417f.

This patch merges cleanly.

vanzin · 2014-10-14T19:51:50Z

core/src/main/scala/org/apache/spark/FutureAction.scala

+  }
+
+  override def jobIds(): java.util.List[java.lang.Integer] = {
+    new java.util.ArrayList(futureAction.jobIds.map(x => new Integer(x)).asJava)


Hmmm... kinda sucks you have to make the conversion manually.

I'd use Collections.unmodifiableList() and Integer.valueOf() here, though.

SparkQA · 2014-10-14T19:52:56Z

QA tests have finished for PR 2760 at commit 7a1417f.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class JavaFutureActionWrapper[S, T](futureAction: FutureAction[S], converter: S => T)

AmplabJenkins · 2014-10-14T19:52:59Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21734/
Test PASSed.

vanzin · 2014-10-14T19:56:07Z

core/src/main/scala/org/apache/spark/FutureAction.scala

+      case scala.util.Success(value) => converter(value)
+      case Failure(exception) =>
+        if (isCancelled) {
+          throw new CancellationException("Job cancelled: ${exception.message}");


Missing s""? I'd also use the exception as the cause of the CancellationException.

Good catch; I'll fix the s"". Unfortunately, CancellationException doesn't have a constructor that accepts a cause exception.

Ah, but I can do this:

val ce = new CancellationException("Job cancelled") ce.initCause(exception) throw ce

Or, even more succinctly:

throw new CancellationException("Job cancelled").initCause(exception)

vanzin · 2014-10-14T20:09:06Z

core/src/test/java/org/apache/spark/JavaAPISuite.java

+    List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
+    JavaRDD<Integer> rdd = sc.parallelize(data, 1);
+    JavaFutureAction<Long> future = rdd.map(new BuggyMapFunction<Integer>()).countAsync();
+    Thread.sleep(200);


Not sure why this is necessary. The get() call below should cover this already.

vanzin · 2014-10-14T20:11:26Z

HI @JoshRosen, this looks mostly good. I have some comments on the tests because I'm generally wary of tests that rely too heavily on timeouts.

JoshRosen · 2014-10-14T22:09:57Z

Hi @vanzin,

Thanks for the review feedback. I've significantly simplified the tests and incorporated your comments.

AmplabJenkins · 2014-10-14T22:28:24Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21742/
Test FAILed.

shaneknapp · 2014-10-14T22:28:35Z

Jenkins, test this please.

SparkQA · 2014-10-14T22:34:42Z

QA tests have started for PR 2760 at commit e8e2867.

This patch merges cleanly.

vanzin · 2014-10-14T23:03:07Z

core/src/test/java/org/apache/spark/JavaAPISuite.java

+    JavaRDD<Integer> rdd = sc.parallelize(data, 1);
+    JavaFutureAction<Long> future = rdd.map(new BuggyMapFunction<Integer>()).countAsync();
+    try {
+      long count = future.get(2, TimeUnit.SECONDS);


nit: variable is unused.

vanzin · 2014-10-14T23:04:11Z

Latest changes LGTM, thanks!

SparkQA · 2014-10-14T23:41:35Z

QA tests have finished for PR 2760 at commit e8e2867.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class JavaFutureActionWrapper[S, T](futureAction: FutureAction[S], converter: S => T)

AmplabJenkins · 2014-10-14T23:41:39Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21743/
Test PASSed.

SparkQA · 2014-10-15T00:19:41Z

QA tests have started for PR 2760 at commit c0153a5.

This patch merges cleanly.

SparkQA · 2014-10-15T01:31:26Z

QA tests have finished for PR 2760 at commit c0153a5.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class JavaFutureActionWrapper[S, T](futureAction: FutureAction[S], converter: S => T)

AmplabJenkins · 2014-10-15T01:31:29Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21754/
Test PASSed.

…n-java Conflicts: project/MimaExcludes.scala

SparkQA · 2014-10-19T07:24:38Z

QA tests have started for PR 2760 at commit 0d45fbc.

This patch merges cleanly.

JoshRosen · 2014-10-19T07:26:52Z

@rxin Did you have any other feedback here? If not, I'd like to merge this.

SparkQA · 2014-10-19T08:32:46Z

QA tests have finished for PR 2760 at commit 0d45fbc.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class JavaFutureActionWrapper[S, T](futureAction: FutureAction[S], converter: S => T)

AmplabJenkins · 2014-10-19T08:32:49Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21883/
Test PASSed.

pwendell · 2014-10-20T00:18:42Z

LGTM - we discussed some details of this offline last week.

JoshRosen · 2014-10-20T03:01:05Z

Alright, I'm going to merge this. Thanks!

rxin · 2014-10-20T22:15:39Z

Reviewed again after the fact. LGTM.

JoshRosen added 2 commits October 10, 2014 19:16

[SPARK-3902] Stabilize AsyncRDDActions; add Java API.

346e46e

Add MiMa excludes and fix a scalastyle error.

ff28e49

JoshRosen reviewed Oct 11, 2014
View reviewed changes

Fix import ordering.

6f8f6ac

JoshRosen changed the title ~~[SPARK-3902] Stabilize AsynRDDActions and add Java API~~ [SPARK-3902] [SPARK-3590] Stabilize AsynRDDActions and add Java API Oct 13, 2014

rxin reviewed Oct 13, 2014
View reviewed changes

Removed unnecessary java.util import.

7a1417f

vanzin reviewed Oct 14, 2014
View reviewed changes

Updates based on Marcelo's review feedback

e8e2867

vanzin reviewed Oct 14, 2014
View reviewed changes

Remove unused variable.

c0153a5

JoshRosen added 2 commits October 19, 2014 00:18

Merge remote-tracking branch 'origin/master' into async-rdd-actions-i…

ad3ae53

…n-java Conflicts: project/MimaExcludes.scala

Whitespace fix.

0d45fbc

asfgit closed this in d1966f3 Oct 20, 2014

		@@ -17,6 +17,7 @@

		package org.apache.spark.api.java

		import java.util

[SPARK-3902] [SPARK-3590] Stabilize AsynRDDActions and add Java API #2760

[SPARK-3902] [SPARK-3590] Stabilize AsynRDDActions and add Java API #2760

Conversation

JoshRosen commented Oct 11, 2014

JoshRosen commented Oct 11, 2014

Choose a reason for hiding this comment

SparkQA commented Oct 11, 2014

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Oct 11, 2014

lirui-apache commented Oct 11, 2014

SparkQA commented Oct 11, 2014

AmplabJenkins commented Oct 11, 2014

SparkQA commented Oct 11, 2014

AmplabJenkins commented Oct 11, 2014

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JoshRosen commented Oct 14, 2014

SparkQA commented Oct 14, 2014

Choose a reason for hiding this comment

SparkQA commented Oct 14, 2014

AmplabJenkins commented Oct 14, 2014

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

Choose a reason for hiding this comment

vanzin commented Oct 14, 2014

JoshRosen commented Oct 14, 2014

AmplabJenkins commented Oct 14, 2014

shaneknapp commented Oct 14, 2014

SparkQA commented Oct 14, 2014

Choose a reason for hiding this comment

vanzin commented Oct 14, 2014

SparkQA commented Oct 14, 2014

AmplabJenkins commented Oct 14, 2014

SparkQA commented Oct 15, 2014

SparkQA commented Oct 15, 2014

AmplabJenkins commented Oct 15, 2014

SparkQA commented Oct 19, 2014

JoshRosen commented Oct 19, 2014

SparkQA commented Oct 19, 2014

AmplabJenkins commented Oct 19, 2014

pwendell commented Oct 20, 2014

JoshRosen commented Oct 20, 2014

rxin commented Oct 20, 2014