apache-spark - 为什么 Spark 从检查点恢复时会抛出 "SparkException: DStream has not been initialized"？

Question

我正在从 HDFS 检查点恢复流(例如，ConstantInputDSTream)，但我不断收到 SparkException: <X> has not been initialized .

从检查点恢复时我需要做一些具体的事情吗？

我可以看到它想要 DStream.zeroTime设置但当流恢复时zeroTime是 null .由于它是私有(private)成员 IDK，它可能无法恢复。我可以看到 StreamingContext恢复的流所引用的值确实为 zeroTime .
initialize是私有(private)方法，在 StreamingContext.graph.start 处调用但不是 StreamingContext.graph.restart ，大概是因为它期望 zeroTime一直坚持。

有人有一个流的示例，它从检查点恢复并且具有zeroTime 的非空值吗？ ?

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(...)

Answer 1

问题是我在从检查点重新创建 StreamingContext 之后创建了 dstream，即在 StreamingContext.getOrCreate 之后。 .创建 dstream 和所有转换应该在 createStreamingContext 中.

问题填写为 [SPARK-13316] "SparkException: DStream has not been initialized" when restoring StreamingContext from checkpoint and the dstream is created afterwards .