Update RecoverableNetworkWordCount.scala #2735

comcmipi · 2014-10-09T15:00:45Z

Trying this example, I missed the moment when the checkpoint was iniciated

AmplabJenkins · 2014-10-09T15:02:11Z

Can one of the admins verify this patch?

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

Can one of the admins verify this patch?

andrewor14 · 2014-10-14T21:24:51Z

@tdas

tdas · 2014-10-23T22:00:54Z

Jenkins, this is ok to test.

SparkQA · 2014-10-23T22:05:44Z

QA tests have started for PR 2735 at commit 96fe274.

This patch merges cleanly.

SparkQA · 2014-10-23T23:17:13Z

QA tests have finished for PR 2735 at commit 96fe274.

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

AmplabJenkins · 2014-10-23T23:17:15Z

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

tdas · 2014-11-07T21:55:07Z

examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala

@@ -116,6 +116,7 @@ object RecoverableNetworkWordCount {
      () => {
        createContext(ip, port, outputPath)
      })
+    ssc.checkpoint(checkpointDirectory)


The right place to add this is in the createContext function, as all the setting up (including the setting up of checkpoint directory) should be in that function. Could you update this PR based on that?

tdas · 2014-11-07T21:56:14Z

Thank you for catching this. The example is almost useless without that! I had one comment, please address that. Other than I tested the change manually (since unit tests dont cover this) and it is good.

tdas · 2014-11-07T21:56:51Z

Also, please make a JIRA and update the title with the JIRA (see other pull requests)

srowen · 2014-11-07T23:36:05Z

I made this change in #2564 too. You could kinda think of this as part of the SPARK-2548 JIRA as well as it includes some touch-ups to the network word count examples.

Ok, I've added ssc.checkpoint(checkpointDirectory) to createContext. First, I wasn't sure that the checkpoin is initiated when the context is recreated from checkpoinDirector. That's why I put it outside createContext.

SparkQA · 2014-11-08T10:02:35Z

Test build #23097 has started for PR 2735 at commit b6d8001.

This patch merges cleanly.

comcmipi · 2014-11-08T10:06:17Z

There is one more issue in this example. It could be the lack of my understanding, or maybe a problem.

When the checkpoint exists and you send some lines to nc while the streaming is off, after restarting streaming those line are recovered. But when you kill/quit streaming, everything within that particular cycle is lost and cannot be recovered. Is it as it should be?

SparkQA · 2014-11-08T11:26:51Z

Test build #23097 has finished for PR 2735 at commit b6d8001.

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

AmplabJenkins · 2014-11-08T11:26:54Z

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

srowen · 2014-11-09T10:24:05Z

examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala

@@ -114,7 +115,7 @@ object RecoverableNetworkWordCount {
    val Array(ip, IntParam(port), checkpointDirectory, outputPath) = args
    val ssc = StreamingContext.getOrCreate(checkpointDirectory,
      () => {
-        createContext(ip, port, outputPath)
+        createContext(ip, port, outputPath, checkpointDirectory)


@tdas Can I double-check that it's correct to call StreamingContext.checkpoint only within the "create context" function? as opposed to always calling it on the result of StreamingContext.getOrCreate? That is, if it reads checkpoint data, it already configures itself to continue using that checkpoint directory?

Yes, here is the piece of code that does it.

https://github.com/tdas/spark/blob/master/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala#L143-L150

Guess things like this should be covered in some unit test. Will add later.

tdas · 2014-11-10T19:46:23Z

@comcmipi Yes, in the current version, if you quit the streaming application (aka driver) quits, then the data received by receivers but not yet processed gets lost. Thats why we have add an experimental feature (already added to master, to be released with Spark 1.2) where the received data will be stored to a log file in HDFS, so that no data is lost.

Here's my attempt to re-port `RecoverableNetworkWordCount` to Java, following the example of its Scala and Java siblings. I fixed a few minor doc/formatting issues along the way I believe. Author: Sean Owen <[email protected]> Closes #2564 from srowen/SPARK-2548 and squashes the following commits: 0d0bf29 [Sean Owen] Update checkpoint call as in #2735 35f23e3 [Sean Owen] Remove old comment about running in standalone mode 179b3c2 [Sean Owen] Re-port RecoverableNetworkWordCount to Java example, and touch up doc / formatting in related examples (cherry picked from commit 3a02d41) Signed-off-by: Tathagata Das <[email protected]>

Here's my attempt to re-port `RecoverableNetworkWordCount` to Java, following the example of its Scala and Java siblings. I fixed a few minor doc/formatting issues along the way I believe. Author: Sean Owen <[email protected]> Closes #2564 from srowen/SPARK-2548 and squashes the following commits: 0d0bf29 [Sean Owen] Update checkpoint call as in #2735 35f23e3 [Sean Owen] Remove old comment about running in standalone mode 179b3c2 [Sean Owen] Re-port RecoverableNetworkWordCount to Java example, and touch up doc / formatting in related examples

comcmipi · 2014-11-10T20:20:56Z

@tdas Thanks for clarifying it.

tdas · 2014-11-10T20:33:13Z

Tested this manually, works perfectly. Merging this. Thanks for this!

Trying this example, I missed the moment when the checkpoint was iniciated Author: comcmipi <[email protected]> Closes #2735 from comcmipi/patch-1 and squashes the following commits: b6d8001 [comcmipi] Update RecoverableNetworkWordCount.scala 96fe274 [comcmipi] Update RecoverableNetworkWordCount.scala

Trying this example, I missed the moment when the checkpoint was iniciated Author: comcmipi <[email protected]> Closes #2735 from comcmipi/patch-1 and squashes the following commits: b6d8001 [comcmipi] Update RecoverableNetworkWordCount.scala 96fe274 [comcmipi] Update RecoverableNetworkWordCount.scala (cherry picked from commit 0340c56) Signed-off-by: Tathagata Das <[email protected]>

Trying this example, I missed the moment when the checkpoint was iniciated Author: comcmipi <[email protected]> Closes apache#2735 from comcmipi/patch-1 and squashes the following commits: b6d8001 [comcmipi] Update RecoverableNetworkWordCount.scala 96fe274 [comcmipi] Update RecoverableNetworkWordCount.scala

Here's my attempt to re-port `RecoverableNetworkWordCount` to Java, following the example of its Scala and Java siblings. I fixed a few minor doc/formatting issues along the way I believe. Author: Sean Owen <[email protected]> Closes #2564 from srowen/SPARK-2548 and squashes the following commits: 0d0bf29 [Sean Owen] Update checkpoint call as in apache/spark#2735 35f23e3 [Sean Owen] Remove old comment about running in standalone mode 179b3c2 [Sean Owen] Re-port RecoverableNetworkWordCount to Java example, and touch up doc / formatting in related examples

Update RecoverableNetworkWordCount.scala

96fe274

Trying this example, I missed the moment when the checkpoint was iniciated

tdas reviewed Nov 7, 2014
View reviewed changes

tdas mentioned this pull request Nov 7, 2014

SPARK-2548 [STREAMING] JavaRecoverableWordCount is missing #2564

Closed

srowen added a commit to srowen/spark that referenced this pull request Nov 7, 2014

Update checkpoint call as in apache#2735

0d0bf29

Update RecoverableNetworkWordCount.scala

b6d8001

Ok, I've added ssc.checkpoint(checkpointDirectory) to createContext. First, I wasn't sure that the checkpoin is initiated when the context is recreated from checkpoinDirector. That's why I put it outside createContext.

srowen reviewed Nov 9, 2014
View reviewed changes

asfgit closed this in 0340c56 Nov 10, 2014

Update RecoverableNetworkWordCount.scala #2735

Update RecoverableNetworkWordCount.scala #2735

Uh oh!

Conversation

comcmipi commented Oct 9, 2014

Uh oh!

AmplabJenkins commented Oct 9, 2014

Uh oh!

AmplabJenkins commented Oct 9, 2014

Uh oh!

andrewor14 commented Oct 14, 2014

Uh oh!

tdas commented Oct 23, 2014

Uh oh!

SparkQA commented Oct 23, 2014

Uh oh!

SparkQA commented Oct 23, 2014

Uh oh!

AmplabJenkins commented Oct 23, 2014

Uh oh!

tdas Nov 7, 2014

Choose a reason for hiding this comment

Uh oh!

tdas commented Nov 7, 2014

Uh oh!

tdas commented Nov 7, 2014

Uh oh!

srowen commented Nov 7, 2014

Uh oh!

SparkQA commented Nov 8, 2014

Uh oh!

comcmipi commented Nov 8, 2014

Uh oh!

SparkQA commented Nov 8, 2014

Uh oh!

AmplabJenkins commented Nov 8, 2014

Uh oh!

srowen Nov 9, 2014

Choose a reason for hiding this comment

Uh oh!

tdas Nov 10, 2014

Choose a reason for hiding this comment

Uh oh!

tdas Nov 10, 2014

Choose a reason for hiding this comment

Uh oh!

tdas commented Nov 10, 2014

Uh oh!

comcmipi commented Nov 10, 2014

Uh oh!

tdas commented Nov 10, 2014

Uh oh!

Uh oh!