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) 

誰もが同じ問題に直面しましたか？ バージョン：

• スパーク1.6.xの
• のHadoop 2.7.35

ありがとう！

Answer 1

スパークストリーミングチェックポイントは、少なくとも1.6.x以前のジョブからのリカバリには使用できません。ジョブが停止して再実行された場合、チェックポイントデータは再利用できません。ジョブを送信する前に、古いチェックポイントデータを削除する必要があります。

[R]アップグレード前コードの以前のチェックポイント情報から開始することができません。チェックポイント情報には、基本的にシリアル化されたScala/Java/Pythonオブジェクトが含まれており、新しいクラスを使用してオブジェクトを逆シリアル化しようとするとエラーが発生する可能性があります。この場合、別のチェックポイントディレクトリを使用してアップグレードしたアプリケーションを起動するか、前のチェックポイントディレクトリを削除します。

Upgrading the code - checkpointing