GitHub user mwws opened a pull request:
https://github.com/apache/spark/pull/11753
[SPARK-13316][Streaming]add check to avoid registering new DStream when
recovering from CP
When creating a recoverable streaming job, it must be no new DStream
registered after a StreamingContext has been recreated from checkpoint. Or you
will see following exception:
```
org.apache.spark.SparkException:
org.apache.spark.streaming.dstream.ConstantInputDStream@724797ab has not been
initialized
at
org.apache.spark.streaming.dstream.DStream.isTimeValid(DStream.scala:311)
at
org.apache.spark.streaming.dstream.InputDStream.isTimeValid(InputDStream.scala:89)
at
org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:332)
at
org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:332)
at scala.Option.orElse(Option.scala:289)
at
org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:329)
at
org.apache.spark.streaming.dstream.ForEachDStream.generateJob(ForEachDStream.scala:48)
at
org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:117)
at
org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:116)
```
This PR is to add meaningful error message to make it obvious at first.
I manually tested the PR with following repo code
```scala
def createStreamingContext(): StreamingContext = {
val ssc = new StreamingContext(sparkConf, Duration(1000))
ssc.checkpoint(checkpointDir)
ssc
}
val ssc = StreamingContext.getOrCreate(checkpointDir),
createStreamingContext)
val socketStream = ssc.socketTextStream(...)
socketStream.checkpoint(Seconds(1))
socketStream.foreachRDD(...)
````
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mwws/spark SPARK-13316
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/11753.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #11753
----
commit 51a03994ae3f66fa80b41a488f27b3fe94abe0ff
Author: mwws <[email protected]>
Date: 2016-03-16T04:20:43Z
add check to avoid registering new DStream when recovering from Checkpoint
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
