[SPARK-11749][Streaming] Duplicate creating the RDD in file stream when recovering from checkpoint data #9765

jhu-chang · 2015-11-17T09:31:59Z

Add a transient flag DStream.restoredFromCheckpointData to control the restore processing in DStream to avoid duplicate works: check this flag first in DStream.restoreCheckpointData, only when false, the restore process will be executed.

zsxwing · 2015-12-11T00:29:41Z

I don't get this one. If DStream.restoreCheckpointData is not called, how to recovery the checkpoint data?

jhu-chang · 2015-12-11T06:30:56Z

@zsxwing
The DStream.restoreCheckpointData will be called only once in this PR no matter how many output paths it is in, and controlled by the new added transient variable restoredFromCheckpointData

zsxwing · 2015-12-11T20:31:34Z

I see. Could you resolve the conflicts?

zsxwing · 2015-12-11T20:31:44Z

streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala

this can be private

jhu-chang · 2015-12-14T09:10:39Z

@zsxwing
Fix the conflicts and move the restoredFromCheckpointData to private

zsxwing · 2015-12-14T18:17:11Z

Jenkins, test this please

SparkQA · 2015-12-14T18:33:11Z

Test build #47660 has started for PR 9765 at commit ac87951.

andrewor14 · 2015-12-14T18:44:49Z

retest this please

shaneknapp · 2015-12-14T19:24:44Z

jenkins test this plese

shaneknapp · 2015-12-14T19:53:54Z

jenkins test this please

SparkQA · 2015-12-14T20:04:03Z

Test build #47672 has finished for PR 9765 at commit ac87951.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2015-12-14T20:22:30Z

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

I suggest writing a simpler test that only tests this issue, such as:

withStreamingContext(new StreamingContext(conf, batchDuration)) { ssc => ssc.checkpoint(checkpointDir) val inputDStream = new CheckpointInputDStream(ssc) val checkpointData = inputDStream.checkpointData val mappedDStream = inputDStream.map(_ + 100) val outputStream = new TestOutputStreamWithPartitions(mappedDStream) outputStream.register() /// do more two times output mappedDStream.foreachRDD(rdd => rdd.count()) mappedDStream.foreachRDD(rdd => rdd.count()) assert(checkpointData.restoredTimes === 0) val batchDurationMillis = ssc.progressListener.batchDuration generateOutput(ssc, Time(batchDurationMillis * 3), checkpointDir, stopSparkContext = false) assert(checkpointData.restoredTimes === 0) } withStreamingContext(new StreamingContext(checkpointDir)) { ssc => val checkpointData = ssc.graph.getInputStreams().head.asInstanceOf[CheckpointInputDStream].checkpointData ssc.start() ssc.stop() assert(checkpointData.restoredTimes === 1) }

jhu-chang · 2015-12-15T04:21:04Z

retest this please

andrewor14 · 2015-12-15T20:59:37Z

retest this please

andrewor14 · 2015-12-15T21:00:07Z

Looks good, merging into master once tests pass.

andrewor14 · 2015-12-15T21:19:39Z

retest this please

SparkQA · 2015-12-15T22:20:10Z

Test build #47752 has finished for PR 9765 at commit dc38d0f.

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

zsxwing · 2015-12-16T00:24:10Z

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

It should be TestOutputStreamWithPartitions if using generateOutput.

jhu-chang · 2015-12-16T03:52:29Z

@zsxwing
Thanks for the comments. I should run the test on local, but due the bad connection issues, I still can not get the full dependency jars from maven after merged master. Hope this time, the unit test will be ok...

jhu-chang · 2015-12-16T03:52:51Z

retest this please

andrewor14 · 2015-12-16T18:14:15Z

retest this please

SparkQA · 2015-12-16T19:25:36Z

Test build #47832 has finished for PR 9765 at commit 625e20a.

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

zsxwing · 2015-12-16T19:41:56Z

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

nit: remove this unused line

zsxwing · 2015-12-16T19:42:34Z

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

You can remove outputBuffer and change val outputStream = new TestOutputStreamWithPartitions(mappedDStream, outputBuffer) to val outputStream = new TestOutputStreamWithPartitions(mappedDStream) since the output buffer is not necessary now.

zsxwing · 2015-12-16T19:42:52Z

Just two nits. Otherwise LGTM

jhu-chang · 2015-12-17T14:47:29Z

@zsxwing
Thanks for your comments, could you help to check this again?

zsxwing · 2015-12-17T17:58:27Z

retest this please

SparkQA · 2015-12-17T19:04:21Z

Test build #47936 has finished for PR 9765 at commit 8131033.

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

andrewor14 · 2015-12-17T19:16:29Z

retest this please?

SparkQA · 2015-12-17T22:42:29Z

Test build #47942 has finished for PR 9765 at commit 8131033.

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

jhu-chang · 2015-12-18T00:35:48Z

OOM happens in the PySpark, Env issue?

zsxwing · 2015-12-18T00:38:56Z

retest this please

SparkQA · 2015-12-18T01:48:11Z

Test build #47966 has finished for PR 9765 at commit 8131033.

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

zsxwing · 2015-12-18T01:52:13Z

@jhu-chang thanks, merging to master and 1.6

…en recovering from checkpoint data Add a transient flag `DStream.restoredFromCheckpointData` to control the restore processing in DStream to avoid duplicate works: check this flag first in `DStream.restoreCheckpointData`, only when `false`, the restore process will be executed. Author: jhu-chang <[email protected]> Closes #9765 from jhu-chang/SPARK-11749. (cherry picked from commit f4346f6) Signed-off-by: Shixiong Zhu <[email protected]>

jhu-chang added 2 commits November 17, 2015 17:17

Add a flag to control the restore in DStream to avoid dup works

02e50c2

merge main branch

268dfd7

zsxwing reviewed Dec 11, 2015
View reviewed changes

streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala Outdated

Copy link

Member

zsxwing Dec 11, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this can be private

jhu-chang added 2 commits December 14, 2015 16:52

merge the main source and fix conflicts

4baa83b

move the restoredFromCheckpointData to private to DStream

ac87951

zsxwing reviewed Dec 14, 2015
View reviewed changes

fix style and only test checkpoint invoke times

dc38d0f

zsxwing reviewed Dec 16, 2015
View reviewed changes

Using TestOutputStreamWithPartitions instead of TestOutputStream

625e20a

zsxwing reviewed Dec 16, 2015
View reviewed changes

streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala Outdated

Copy link

Member

zsxwing Dec 16, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: remove this unused line

zsxwing reviewed Dec 16, 2015
View reviewed changes

remove un-used variable in test case

8131033

asfgit closed this in f4346f6 Dec 18, 2015

jhu-chang deleted the SPARK-11749 branch January 5, 2016 07:57

[SPARK-11749][Streaming] Duplicate creating the RDD in file stream when recovering from checkpoint data #9765

[SPARK-11749][Streaming] Duplicate creating the RDD in file stream when recovering from checkpoint data #9765

Uh oh!

Conversation

jhu-chang commented Nov 17, 2015

Uh oh!

zsxwing commented Dec 11, 2015

Uh oh!

jhu-chang commented Dec 11, 2015

Uh oh!

zsxwing commented Dec 11, 2015

Uh oh!

zsxwing Dec 11, 2015

Choose a reason for hiding this comment

Uh oh!

jhu-chang commented Dec 14, 2015

Uh oh!

zsxwing commented Dec 14, 2015

Uh oh!

SparkQA commented Dec 14, 2015

Uh oh!

andrewor14 commented Dec 14, 2015

Uh oh!

shaneknapp commented Dec 14, 2015

Uh oh!

shaneknapp commented Dec 14, 2015

Uh oh!

SparkQA commented Dec 14, 2015

Uh oh!

zsxwing Dec 14, 2015

Choose a reason for hiding this comment

Uh oh!

jhu-chang commented Dec 15, 2015

Uh oh!

andrewor14 commented Dec 15, 2015

Uh oh!

andrewor14 commented Dec 15, 2015

Uh oh!

andrewor14 commented Dec 15, 2015

Uh oh!

SparkQA commented Dec 15, 2015

Uh oh!

zsxwing Dec 16, 2015

Choose a reason for hiding this comment

Uh oh!

jhu-chang commented Dec 16, 2015

Uh oh!

jhu-chang commented Dec 16, 2015

Uh oh!

andrewor14 commented Dec 16, 2015

Uh oh!

SparkQA commented Dec 16, 2015

Uh oh!

zsxwing Dec 16, 2015

Choose a reason for hiding this comment

Uh oh!

zsxwing Dec 16, 2015

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Dec 16, 2015

Uh oh!

jhu-chang commented Dec 17, 2015

Uh oh!

zsxwing commented Dec 17, 2015

Uh oh!

SparkQA commented Dec 17, 2015

Uh oh!

andrewor14 commented Dec 17, 2015

Uh oh!

SparkQA commented Dec 17, 2015

Uh oh!

jhu-chang commented Dec 18, 2015

Uh oh!

zsxwing commented Dec 18, 2015

Uh oh!

SparkQA commented Dec 18, 2015

Uh oh!

zsxwing commented Dec 18, 2015

Uh oh!

Reviewers