Apache Spark：从检查点恢复状态期间的 NPE

Question

我们正在构建使用 HBase RDD 与传入 DStream 连接的简单 Streaming 应用程序。 示例代码：

val indexState = sc.newAPIHadoopRDD(
  conf,
  classOf[TableInputFormat],
  classOf[ImmutableBytesWritable],
  classOf[Result]).map { case (rowkey, v) => //some logic}

val result = dStream.transform { rdd =>
  rdd.leftOuterJoin(indexState)
}  

它工作正常，但是当我们为 StreamingContext 启用检查点时 并让应用程序从先前创建的检查点中恢复， 它总是抛出 NullPointerException。

ERROR streaming.StreamingContext: Error starting the context, marking it as stopped
java.lang.NullPointerException
        at org.apache.hadoop.hbase.mapreduce.TableInputFormat.setConf(TableInputFormat.java:119)
        at org.apache.spark.rdd.NewHadoopRDD.getPartitions(NewHadoopRDD.scala:120)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
        at scala.Option.getOrElse(Option.scala:120)
        at org.apache.spark.rdd.RDD.partitions(RDD.scala:237)
        at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239)
        at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237)
        at scala.Option.getOrElse(Option.scala:120)

有人遇到过同样的问题吗？ 版本：

• Spark 1.6.x
• Hadoop 2.7.x

谢谢！

Answer 1

Spark Streaming 检查点不能用于从以前的作业中恢复，至少在 1.6.x 中是这样。如果您的作业停止并重新提交，则无法重新使用检查点数据。在提交作业之前，您必须删除所有旧的检查点数据。

[R]无法从升级前代码的早期检查点信息重新开始。检查点信息本质上包含序列化的 Scala/Java/Python 对象，尝试使用新的、修改的类反序列化对象可能会导致错误。在这种情况下，要么使用不同的检查点目录启动升级后的应用，要么删除之前的检查点目录。

Upgrading the code - checkpointing