[SPARK-3660][STREAMING] Initial RDD for updateStateByKey transformation #2665

soumitrak · 2014-10-06T03:38:20Z

SPARK-3660 : Initial RDD for updateStateByKey transformation

I have added a sample StatefulNetworkWordCountWithInitial inspired by StatefulNetworkWordCount.

Please let me know if any changes are required.

SPARK-3660 : Initial RDD for updateStateByKey transformation

AmplabJenkins · 2014-10-06T03:42:09Z

Can one of the admins verify this patch?

AmplabJenkins · 2014-10-09T20:32:36Z

Can one of the admins verify this patch?

tdas · 2014-10-21T22:39:02Z

@soumitrak This is a good addition. Here are some high level suggestions.

Please update the title as [SPARK-3660][STREAMING]....
Please add unit tests for this new functionality
Lets not create another example. Please update the StatefulNetworkWordCount exammple. That should be cool.

tdas · 2014-10-21T22:39:08Z

Jenkins, this is ok to test.

SparkQA · 2014-10-21T22:44:42Z

QA tests have started for PR 2665 at commit dde4271.

This patch merges cleanly.

SparkQA · 2014-10-21T23:54:44Z

QA tests have finished for PR 2665 at commit dde4271.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-10-21T23:54:48Z

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

tdas · 2014-10-24T18:56:52Z

@soumitrak Hey any thoughts?

soumitrak · 2014-10-25T03:27:35Z

Sorry, I was off for couple of days.

I will send a new pull request with title changed
Is there a unit test for StatefulNetworkWordCount? I can use that to create a test.
I will modify StatefulNetworkWordCount as suggested.

----- Original Message -----
From: "Tathagata Das" notifications@github.com
To: "apache/spark" spark@noreply.github.com
Cc: "soumitrak" kumar.soumitra@gmail.com
Sent: Friday, October 24, 2014 11:57:05 AM
Subject: Re: [spark] Adding support of initial value for state update. (#2665)

@soumitrak Hey any thoughts?

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

SparkQA · 2014-10-25T03:30:04Z

Test build #22190 has started for PR 2665 at commit 8f40ca0.

This patch merges cleanly.

SparkQA · 2014-10-25T05:30:04Z

Test build #22190 timed out for PR 2665 at commit 8f40ca0 after a configured wait of 120m.

AmplabJenkins · 2014-10-25T05:30:07Z

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

soumitrak · 2014-10-25T07:57:52Z

TD, I have incorporated your feedback, and add a testcase. Let me know if there is anything else.

SparkQA · 2014-10-25T07:59:55Z

Test build #22206 has started for PR 2665 at commit 4efa58b.

This patch merges cleanly.

SparkQA · 2014-10-25T08:58:13Z

Test build #22206 has finished for PR 2665 at commit 4efa58b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-10-25T08:58:17Z

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

tdas · 2014-11-07T22:07:12Z

Jenkins, test this please.

SparkQA · 2014-11-07T22:12:43Z

Test build #23074 has started for PR 2665 at commit 4efa58b.

This patch does not merge cleanly.

SparkQA · 2014-11-07T23:25:52Z

Test build #23074 has finished for PR 2665 at commit 4efa58b.

This patch fails Spark unit tests.
This patch does not merge cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-11-07T23:25:55Z

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

soumitrak · 2014-11-07T23:40:02Z

TD, Let me know how I can help.

tdas · 2014-11-10T23:15:42Z

Jenkins, test this please.

tdas · 2014-11-10T23:19:30Z

streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala

+   * Set up required DStreams to test the DStream operation using the sequence
+   * of input collections, and initial sequence.
+   */
+  def setupStreamsWithInitial[U: ClassTag, V: ClassTag](


I am not sure it is necessary to add this method.

See other comment.

SparkQA · 2014-11-10T23:20:18Z

Test build #23166 has started for PR 2665 at commit 4efa58b.

This patch does not merge cleanly.

tdas · 2014-11-10T23:44:32Z

streaming/src/test/scala/org/apache/spark/streaming/BasicOperationsSuite.scala

+        new HashPartitioner (numInputPartitions), true, initialRDD)
+    }
+
+    testOperationWithInitial(initial, inputData, updateStateOperation, outputData, true)


Instead of setting up and using testOperationWithInitial just for this unit test, lets try to do the alternative.

val updateStateOperation = (s: DStream[String]) => { val initialRDD = s.context.sparkContext.makeRDD(initial) val updateFunc = (values: Seq[Int], state: Option[Int]) => { Some(values.sum + state.getOrElse(0)) } val newUpdateFunc = (iterator: Iterator[(String, Seq[Int], Option[Int])]) => { iterator.flatMap(t => updateFunc(t._2, t._3).map(s => (t._1, s))) } s.map(x => (x, 1)).updateStateByKey[Int](newUpdateFunc, new HashPartitioner (numInputPartitions), true, initialRDD) }

And then use the usual testOperation

SparkQA · 2014-11-11T20:13:54Z

Test build #23219 has finished for PR 2665 at commit 3da51a2.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- trait LeaderElectionAgent
- trait LeaderElectable
- trait PersistenceEngine
- abstract class StandaloneRecoveryModeFactory(conf: SparkConf)

AmplabJenkins · 2014-11-11T20:13:58Z

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

SparkQA · 2014-11-11T21:07:42Z

Test build #23223 has started for PR 2665 at commit 9781135.

This patch merges cleanly.

tdas · 2014-11-11T21:40:49Z

streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaPairDStream.scala

@@ -443,6 +443,23 @@ class JavaPairDStream[K, V](val dstream: DStream[(K, V)])(
    scalaFunc
  }

+  private def convertUpdateStateFunctionWithIterator[S]


Incorrect indentation

private def convertUpdateStateFunctionWithIterator[S]( in: JFunction2[JList[V], Optional[S], Optional[S]] // 4 space indent ): (Iterator[(K, Seq[V], Option[S])]) => Iterator[(K, S)] = { // 2 space indent

tdas · 2014-11-11T22:19:50Z

This is starting to look good. Please address the comments (mainly Scala API addition), and we will good to go. This is a good addition to updateStateByKey.

SparkQA · 2014-11-11T22:33:55Z

Test build #23223 has finished for PR 2665 at commit 9781135.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-11-11T22:33:58Z

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

SparkQA · 2014-11-11T22:50:24Z

Test build #23225 has started for PR 2665 at commit 304f636.

This patch merges cleanly.

SparkQA · 2014-11-12T00:01:13Z

Test build #23225 has finished for PR 2665 at commit 304f636.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- trait LeaderElectionAgent
- trait LeaderElectable
- trait PersistenceEngine
- abstract class StandaloneRecoveryModeFactory(conf: SparkConf)

AmplabJenkins · 2014-11-12T00:01:17Z

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

SparkQA · 2014-11-12T00:40:15Z

Test build #23231 has started for PR 2665 at commit ee8980b.

This patch merges cleanly.

tdas · 2014-11-12T01:44:04Z

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

+
+    JavaPairDStream<String, Integer> updated = pairStream.updateStateByKey(
+        new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {
+        @Override


Nit (do this if you have to make more changes): incorrect indentation. should be indented by more than the line with new Function2...

I don't have any other change to do. Actually I copied the previous method.

SparkQA · 2014-11-12T02:13:04Z

Test build #23231 has finished for PR 2665 at commit ee8980b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-11-12T02:13:08Z

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

tdas · 2014-11-12T20:24:48Z

Alright I am merging this.

tdas · 2014-11-12T20:27:23Z

Thanks very much for the changes!

soumitrak · 2014-11-12T22:40:05Z

TD, Thanks for getting this through.

SPARK-3660 : Initial RDD for updateStateByKey transformation I have added a sample StatefulNetworkWordCountWithInitial inspired by StatefulNetworkWordCount. Please let me know if any changes are required. Author: Soumitra Kumar <kumar.soumitra@gmail.com> Closes apache#2665 from soumitrak/master and squashes the following commits: ee8980b [Soumitra Kumar] Fixed copy/paste issue. 304f636 [Soumitra Kumar] Added simpler version of updateStateByKey API with initialRDD and test. 9781135 [Soumitra Kumar] Fixed test, and renamed variable. 3da51a2 [Soumitra Kumar] Adding updateStateByKey with initialRDD API to JavaPairDStream. 2f78f7e [Soumitra Kumar] Merge remote-tracking branch 'upstream/master' d4fdd18 [Soumitra Kumar] Renamed variable and moved method. d0ce2cd [Soumitra Kumar] Merge remote-tracking branch 'upstream/master' 31399a4 [Soumitra Kumar] Merge remote-tracking branch 'upstream/master' 4efa58b [Soumitra Kumar] [SPARK-3660][STREAMING] Initial RDD for updateStateByKey transformation 8f40ca0 [Soumitra Kumar] Merge remote-tracking branch 'upstream/master' dde4271 [Soumitra Kumar] Merge remote-tracking branch 'upstream/master' fdd7db3 [Soumitra Kumar] Adding support of initial value for state update. SPARK-3660 : Initial RDD for updateStateByKey transformation

Adding support of initial value for state update.

fdd7db3

SPARK-3660 : Initial RDD for updateStateByKey transformation

Merge remote-tracking branch 'upstream/master'

dde4271

Merge remote-tracking branch 'upstream/master'

8f40ca0

[SPARK-3660][STREAMING] Initial RDD for updateStateByKey transformation

4efa58b

soumitrak changed the title ~~Adding support of initial value for state update.~~ [SPARK-3660][STREAMING] Initial RDD for updateStateByKey transformation Oct 25, 2014

tdas reviewed Nov 10, 2014
View reviewed changes

Fixed test, and renamed variable.

9781135

tdas reviewed Nov 11, 2014
View reviewed changes

Added simpler version of updateStateByKey API with initialRDD and test.

304f636

Fixed copy/paste issue.

ee8980b

tdas reviewed Nov 12, 2014
View reviewed changes

asfgit closed this in 36ddeb7 Nov 12, 2014

[SPARK-3660][STREAMING] Initial RDD for updateStateByKey transformation #2665

[SPARK-3660][STREAMING] Initial RDD for updateStateByKey transformation #2665

Conversation

soumitrak commented Oct 6, 2014

AmplabJenkins commented Oct 6, 2014

AmplabJenkins commented Oct 9, 2014

tdas commented Oct 21, 2014

tdas commented Oct 21, 2014

SparkQA commented Oct 21, 2014

SparkQA commented Oct 21, 2014

AmplabJenkins commented Oct 21, 2014

tdas commented Oct 24, 2014

soumitrak commented Oct 25, 2014

SparkQA commented Oct 25, 2014

SparkQA commented Oct 25, 2014

AmplabJenkins commented Oct 25, 2014

soumitrak commented Oct 25, 2014

SparkQA commented Oct 25, 2014

SparkQA commented Oct 25, 2014

AmplabJenkins commented Oct 25, 2014

tdas commented Nov 7, 2014

SparkQA commented Nov 7, 2014

SparkQA commented Nov 7, 2014

AmplabJenkins commented Nov 7, 2014

soumitrak commented Nov 7, 2014

tdas commented Nov 10, 2014

tdas Nov 10, 2014

Choose a reason for hiding this comment

tdas Nov 10, 2014

Choose a reason for hiding this comment

SparkQA commented Nov 10, 2014

tdas Nov 10, 2014

Choose a reason for hiding this comment

SparkQA commented Nov 11, 2014

AmplabJenkins commented Nov 11, 2014

SparkQA commented Nov 11, 2014

tdas Nov 11, 2014

Choose a reason for hiding this comment

tdas commented Nov 11, 2014

SparkQA commented Nov 11, 2014

AmplabJenkins commented Nov 11, 2014

SparkQA commented Nov 11, 2014

SparkQA commented Nov 12, 2014

AmplabJenkins commented Nov 12, 2014

SparkQA commented Nov 12, 2014

tdas Nov 12, 2014

Choose a reason for hiding this comment

soumitrak Nov 12, 2014

Choose a reason for hiding this comment

SparkQA commented Nov 12, 2014

AmplabJenkins commented Nov 12, 2014

tdas commented Nov 12, 2014

tdas commented Nov 12, 2014

soumitrak commented Nov 12, 2014