Spark Streaming 检查点异常答案

【问题标题】：Spark Streaming Checkpoint exceptionSpark Streaming 检查点异常
【发布时间】：2016-06-12 17:05:46
【问题描述】：

我有我的火花流管道与 kafka 集成，我还配置了检查点，为了测试弹性，我手动终止了作业，然后重新启动它，然后我遇到了异常

Exception in thread "main" org.apache.spark.SparkException: org.apache.spark.streaming.dstream.ShuffledDStream@1d304ac has not been initialized
    at org.apache.spark.streaming.dstream.DStream.isTimeValid(DStream.scala:267)

我使用的代码

JavaStreamingContextFactory contextFactory = new JavaStreamingContextFactory() {


      public JavaStreamingContext create() {
          final SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount");
            sparkConf.setMaster("local[2]");

        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(2000));  // new context

        jssc.checkpoint("D:\\Checkpoint");
        return jssc;
      }
    };

    // Get JavaStreamingContext from checkpoint data or create a new one
    JavaStreamingContext jssc = JavaStreamingContext.getOrCreate("D:\\Checkpoint", contextFactory);
    int numThreads = Integer.parseInt(1+"");

请指出我做错了什么

【问题讨论】：

标签： java apache-spark spark-streaming

【解决方案1】：

private static JavaStreamingContext createContext(){
    final SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount");
    sparkConf.setMaster("local[2]");

    JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(2000));  // new context

    jssc.checkpoint("D:\\Checkpoint");
    return jssc;
}

Function0<JavaStreamingContext> createContextFunc = new Function0<JavaStreamingContext>() {
    @Override
    public JavaStreamingContext call() {
        return createContext();
    }
};

JavaStreamingContext streamingContext = JavaStreamingContext.getOrCreate("D:\\Checkpoint", createContextFunc);

使用Function0，这对我有用。

【讨论】：

【解决方案2】：

http://spark.apache.org/docs/latest/streaming-programming-guide.html#checkpointing

您没有显示用于创建 DStreams/转换的代码。将它们放在 create() 方法中。

【讨论】：