[SPARK-6752][Streaming] Allow StreamingContext to be recreated from checkpoint and existing SparkContext #5428

tdas · 2015-04-08T22:35:12Z

Currently if you want to create a StreamingContext from checkpoint information, the system will create a new SparkContext. This prevent StreamingContext to be recreated from checkpoints in managed environments where SparkContext is precreated.

The solution in this PR: Introduce the following methods on StreamingContext

new StreamingContext(checkpointDirectory, sparkContext)
Recreate StreamingContext from checkpoint using the provided SparkContext
StreamingContext.getOrCreate(checkpointDirectory, sparkContext, createFunction: SparkContext => StreamingContext)
If checkpoint file exists, then recreate StreamingContext using the provided SparkContext (that is, call 1.), else create StreamingContext using the provided createFunction

TODO: the corresponding Java and Python API has to be added as well.

SparkQA · 2015-04-08T22:38:23Z

Test build #29898 has started for PR 5428 at commit 36a7823.

SparkQA · 2015-04-09T00:03:46Z

Test build #29898 has finished for PR 5428 at commit 36a7823.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

AmplabJenkins · 2015-04-09T00:03:50Z

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

tdas · 2015-04-09T09:46:01Z

@jerryshao Mind taking a look at this? Its still WIP as unit tests are commented out.

zzcclp · 2015-04-09T10:00:08Z

@tdas , can this RP resolve this issue?
Restart a streaming app from checkpoint incorrectly if using accumulators .

jerryshao · 2015-04-09T12:28:42Z

@tdas Yeah, will do.

@zzcclp I'm not sure, maybe you can take a try, from my guess, this could possibly work, since accumulator is registered in SparkContext, while SparkContext is still existed.

jerryshao · 2015-04-09T13:23:28Z

streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala

Is it better to change to string interpolator style?

Yes, I will. Thanks!

Do you think that we should log a warning message in the case where we ignore the error?

jerryshao · 2015-04-09T13:31:56Z

It looks good to me. Simply curious about the scenarios of this usage, is there any situation where streaming context is failed but spark context is still existed when driver failure is met?

harishreedharan · 2015-04-09T17:53:50Z

streaming/src/main/scala/org/apache/spark/streaming/Checkpoint.scala

This change seems unrelated to this fix. I think we can simply do a null check inside this method and create a FileSystem if needed to avoid unnecessary changes to the calls (all the fs being passed in changing to Some(fs)) -- keeps git history sane.

The reason I added this is so that we should not have to handle nulls. Dealing with nulls is severely frowned upon in Sclaa, and precisely why Option was introduced. There are many places where this has been done, and slowly I was fix those. I think this is a small enough change (doesnt change functionality, or existing code paths) that is okay to do this.

BTW, this file has to change anyways in the attempt to make the semantics of read more clear.

zzcclp · 2015-04-10T01:25:44Z

@jerryshao , thanks, I will test this case later.

tdas · 2015-04-15T00:19:03Z

@zzcclp I dont think it will solve this issue directly. But it may allow the SparkContext to be re-initialized properly before the StreamingContext is recreated from checkpoints.

tdas · 2015-04-15T00:27:05Z

@ALL This is still a WIP. Adding the equivalent Java API requires refactoring the existing JavaStreamingContext.getOrCreate to not use JavaStreamingContextFactory and use o.a.s.java.api.Function0 (which needs to be added). I am going to open other PRs for them before this can be merged.

…eate

SparkQA · 2015-04-16T01:18:42Z

Test build #30389 has started for PR 5428 at commit eabd092.

tdas · 2015-04-16T01:23:14Z

@JoshRosen Please take a quick look at the Function.
@jerryshao @harishreedharan I have updated the patch with Java API and unit tests. I think I am going to create a separate JIRA for python API.

tdas · 2015-04-16T01:25:16Z

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

Some of these are unrelated to the PR, but just cleans up the formatting of the JavaAPISuite which is quite badly formatted.

SparkQA · 2015-04-16T02:24:26Z

Test build #30389 has finished for PR 5428 at commit eabd092.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

AmplabJenkins · 2015-04-16T02:24:31Z

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

tdas · 2015-04-16T08:00:41Z

Jenkins, test this again.

tdas · 2015-04-20T22:50:34Z

Jenkins, test this.

SparkQA · 2015-04-21T23:05:35Z

Test build #691 has started for PR 5428 at commit eabd092.

SparkQA · 2015-04-22T00:09:17Z

Test build #691 has finished for PR 5428 at commit eabd092.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

SparkQA · 2015-04-22T01:56:37Z

Test build #692 has started for PR 5428 at commit eabd092.

zzcclp · 2015-04-22T02:26:41Z

@tdas , I tested streaming recovering from checkpoint with this PR, it failed if it use accumulators, so this assuredly can't solve issue SPARK-5206 directly. how to solve issue SPARK-5206?

tdas · 2015-04-22T02:37:32Z

Yes, this is not intended to solve SPARK-5206.

SparkQA · 2015-04-22T08:28:37Z

Test build #30743 has started for PR 5428 at commit 524f519.

SparkQA · 2015-04-22T08:30:05Z

Test build #30743 has finished for PR 5428 at commit 524f519.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

AmplabJenkins · 2015-04-22T08:30:06Z

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

SparkQA · 2015-04-22T10:38:40Z

Test build #30751 has started for PR 5428 at commit 94db63c.

SparkQA · 2015-04-22T11:49:25Z

Test build #30751 has finished for PR 5428 at commit 94db63c.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

AmplabJenkins · 2015-04-22T11:49:31Z

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

SparkQA · 2015-04-22T13:20:07Z

Test build #693 has started for PR 5428 at commit 94db63c.

SparkQA · 2015-04-22T14:28:31Z

Test build #693 has finished for PR 5428 at commit 94db63c.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

SparkQA · 2015-04-22T19:01:21Z

Test build #694 has started for PR 5428 at commit 94db63c.

JoshRosen · 2015-04-22T20:15:56Z

LGTM pending Jenkins.

SparkQA · 2015-04-23T01:35:11Z

Test build #695 has started for PR 5428 at commit 94db63c.

SparkQA · 2015-04-23T02:41:26Z

Test build #695 has finished for PR 5428 at commit 94db63c.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

SparkQA · 2015-04-23T05:19:57Z

Test build #696 has started for PR 5428 at commit 94db63c.

SparkQA · 2015-04-23T07:25:09Z

Test build #697 has started for PR 5428 at commit 94db63c.

SparkQA · 2015-04-23T09:00:25Z

Test build #697 has finished for PR 5428 at commit 94db63c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

tdas · 2015-04-23T18:28:24Z

Merging this. Thanks Josh.

zzcclp · 2015-04-27T03:12:07Z

hi, @tdas , why this PR was reverted?

tdas · 2015-04-29T18:23:27Z

This PR was reverted because I had used MutableBoolean which does not seem to work well with Hadoop 1.0.4. I reopened the PR in #5773.

…eated from checkpoint and existing SparkContext Original PR #5428 got reverted due to issues between MutableBoolean and Hadoop 1.0.4 (see JIRA). This replaces MutableBoolean with AtomicBoolean. srowen pwendell Author: Tathagata Das <[email protected]> Closes #5773 from tdas/SPARK-6752 and squashes the following commits: a0c0ead [Tathagata Das] Fix for hadoop 1.0.4 70ae85b [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into SPARK-6752 94db63c [Tathagata Das] Fix long line. 524f519 [Tathagata Das] Many changes based on PR comments. eabd092 [Tathagata Das] Added Function0, Java API and unit tests for StreamingContext.getOrCreate 36a7823 [Tathagata Das] Minor changes. 204814e [Tathagata Das] Added StreamingContext.getOrCreate with existing SparkContext

tdas · 2015-05-12T23:57:31Z

I have revised the implementation of this PR in a followup PR #6096

…heckpoint and existing SparkContext Currently if you want to create a StreamingContext from checkpoint information, the system will create a new SparkContext. This prevent StreamingContext to be recreated from checkpoints in managed environments where SparkContext is precreated. The solution in this PR: Introduce the following methods on StreamingContext 1. `new StreamingContext(checkpointDirectory, sparkContext)` Recreate StreamingContext from checkpoint using the provided SparkContext 2. `StreamingContext.getOrCreate(checkpointDirectory, sparkContext, createFunction: SparkContext => StreamingContext)` If checkpoint file exists, then recreate StreamingContext using the provided SparkContext (that is, call 1.), else create StreamingContext using the provided createFunction TODO: the corresponding Java and Python API has to be added as well. Author: Tathagata Das <[email protected]> Closes apache#5428 from tdas/SPARK-6752 and squashes the following commits: 94db63c [Tathagata Das] Fix long line. 524f519 [Tathagata Das] Many changes based on PR comments. eabd092 [Tathagata Das] Added Function0, Java API and unit tests for StreamingContext.getOrCreate 36a7823 [Tathagata Das] Minor changes. 204814e [Tathagata Das] Added StreamingContext.getOrCreate with existing SparkContext

…eated from checkpoint and existing SparkContext Original PR apache#5428 got reverted due to issues between MutableBoolean and Hadoop 1.0.4 (see JIRA). This replaces MutableBoolean with AtomicBoolean. srowen pwendell Author: Tathagata Das <[email protected]> Closes apache#5773 from tdas/SPARK-6752 and squashes the following commits: a0c0ead [Tathagata Das] Fix for hadoop 1.0.4 70ae85b [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into SPARK-6752 94db63c [Tathagata Das] Fix long line. 524f519 [Tathagata Das] Many changes based on PR comments. eabd092 [Tathagata Das] Added Function0, Java API and unit tests for StreamingContext.getOrCreate 36a7823 [Tathagata Das] Minor changes. 204814e [Tathagata Das] Added StreamingContext.getOrCreate with existing SparkContext

…heckpoint and existing SparkContext Currently if you want to create a StreamingContext from checkpoint information, the system will create a new SparkContext. This prevent StreamingContext to be recreated from checkpoints in managed environments where SparkContext is precreated. The solution in this PR: Introduce the following methods on StreamingContext 1. `new StreamingContext(checkpointDirectory, sparkContext)` Recreate StreamingContext from checkpoint using the provided SparkContext 2. `StreamingContext.getOrCreate(checkpointDirectory, sparkContext, createFunction: SparkContext => StreamingContext)` If checkpoint file exists, then recreate StreamingContext using the provided SparkContext (that is, call 1.), else create StreamingContext using the provided createFunction TODO: the corresponding Java and Python API has to be added as well. Author: Tathagata Das <[email protected]> Closes apache#5428 from tdas/SPARK-6752 and squashes the following commits: 94db63c [Tathagata Das] Fix long line. 524f519 [Tathagata Das] Many changes based on PR comments. eabd092 [Tathagata Das] Added Function0, Java API and unit tests for StreamingContext.getOrCreate 36a7823 [Tathagata Das] Minor changes. 204814e [Tathagata Das] Added StreamingContext.getOrCreate with existing SparkContext

…eated from checkpoint and existing SparkContext Original PR apache#5428 got reverted due to issues between MutableBoolean and Hadoop 1.0.4 (see JIRA). This replaces MutableBoolean with AtomicBoolean. srowen pwendell Author: Tathagata Das <[email protected]> Closes apache#5773 from tdas/SPARK-6752 and squashes the following commits: a0c0ead [Tathagata Das] Fix for hadoop 1.0.4 70ae85b [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into SPARK-6752 94db63c [Tathagata Das] Fix long line. 524f519 [Tathagata Das] Many changes based on PR comments. eabd092 [Tathagata Das] Added Function0, Java API and unit tests for StreamingContext.getOrCreate 36a7823 [Tathagata Das] Minor changes. 204814e [Tathagata Das] Added StreamingContext.getOrCreate with existing SparkContext

tdas added 2 commits April 8, 2015 15:30

Added StreamingContext.getOrCreate with existing SparkContext

204814e

Minor changes.

36a7823

tdas changed the title ~~[Spark-6752]~~ [Spark-6752][Streaming] Allow StreamingContext to be recreated from checkpoint and existing SparkContext Apr 8, 2015

tdas changed the title ~~[Spark-6752][Streaming] Allow StreamingContext to be recreated from checkpoint and existing SparkContext~~ [SPARK-6752][Streaming] Allow StreamingContext to be recreated from checkpoint and existing SparkContext Apr 8, 2015

jerryshao reviewed Apr 9, 2015
View reviewed changes

harishreedharan reviewed Apr 9, 2015
View reviewed changes

Added Function0, Java API and unit tests for StreamingContext.getOrCr…

eabd092

…eate

tdas reviewed Apr 16, 2015
View reviewed changes

Many changes based on PR comments.

524f519

Fix long line.

94db63c

asfgit closed this in 534f2a4 Apr 23, 2015

tdas mentioned this pull request Apr 29, 2015

[SPARK-6752][Streaming][Reopened] Allow StreamingContext to be recreated from checkpoint and existing SparkContext #5773

Closed

tdas mentioned this pull request May 12, 2015

[SPARK-6752][STREAMING][Revised] Allow StreamingContext to be recreated from checkpoint and existing SparkContext #6096

Closed

[SPARK-6752][Streaming] Allow StreamingContext to be recreated from checkpoint and existing SparkContext #5428

[SPARK-6752][Streaming] Allow StreamingContext to be recreated from checkpoint and existing SparkContext #5428

Uh oh!

Conversation

tdas commented Apr 8, 2015

Uh oh!

SparkQA commented Apr 8, 2015

Uh oh!

SparkQA commented Apr 9, 2015

Uh oh!

AmplabJenkins commented Apr 9, 2015

Uh oh!

tdas commented Apr 9, 2015

Uh oh!

zzcclp commented Apr 9, 2015

Uh oh!

jerryshao commented Apr 9, 2015

Uh oh!

jerryshao Apr 9, 2015

Choose a reason for hiding this comment

Uh oh!

tdas Apr 15, 2015

Choose a reason for hiding this comment

Uh oh!

JoshRosen Apr 22, 2015

Choose a reason for hiding this comment

Uh oh!

jerryshao commented Apr 9, 2015

Uh oh!

harishreedharan Apr 9, 2015

Choose a reason for hiding this comment

Uh oh!

tdas Apr 15, 2015

Choose a reason for hiding this comment

Uh oh!

tdas Apr 15, 2015

Choose a reason for hiding this comment

Uh oh!

zzcclp commented Apr 10, 2015

Uh oh!

tdas commented Apr 15, 2015

Uh oh!

tdas commented Apr 15, 2015

Uh oh!

SparkQA commented Apr 16, 2015

Uh oh!

tdas commented Apr 16, 2015

Uh oh!

tdas Apr 16, 2015

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 16, 2015

Uh oh!

AmplabJenkins commented Apr 16, 2015

Uh oh!

tdas commented Apr 16, 2015

Uh oh!

tdas commented Apr 20, 2015

Uh oh!

SparkQA commented Apr 21, 2015

Uh oh!

SparkQA commented Apr 22, 2015

Uh oh!

SparkQA commented Apr 22, 2015

Uh oh!

zzcclp commented Apr 22, 2015

Uh oh!

tdas commented Apr 22, 2015

Uh oh!

SparkQA commented Apr 22, 2015

Uh oh!

SparkQA commented Apr 22, 2015

Uh oh!

AmplabJenkins commented Apr 22, 2015

Uh oh!

SparkQA commented Apr 22, 2015

Uh oh!

SparkQA commented Apr 22, 2015

Uh oh!

AmplabJenkins commented Apr 22, 2015