从流式 kafka 检查点 ClassNotFoundException 恢复

Question

我正在使用 spark-streaming-kafka 检查点将处理后的 Kafka 偏移存储到 HDFS 中的文件夹中，在重新启动应用程序（使用 spark-submit）以检查恢复后，我在一个类上得到了 ClassNotFoundException属于 spark-streaming-kafka 模块并被打包到我的应用程序 uber jar 中。似乎没有在我的应用程序 jar 中查找该类。

使用 v1.5.1

15/12/02 15:42:30 INFO streaming.CheckpointReader: Attempting to load checkpoint from file hdfs://ip-xxx-xx-xx-xx:8020/user/checkpoint-1449064500000
15/12/02 15:42:30 WARN streaming.CheckpointReader: Error reading checkpoint from file hdfs://ip-xxx-xx-xx-xx:8020/user/checkpoint-1449064500000
java.io.IOException: java.lang.ClassNotFoundException: org.apache.spark.streaming.kafka.OffsetRange
    at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1163)
    at org.apache.spark.streaming.DStreamGraph.readObject(DStreamGraph.scala:188)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1017)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1893)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1990)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1915)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
    at org.apache.spark.streaming.Checkpoint$$anonfun$deserialize$2.apply(Checkpoint.scala:151)
    at org.apache.spark.streaming.Checkpoint$$anonfun$deserialize$2.apply(Checkpoint.scala:141)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1206)
    at org.apache.spark.streaming.Checkpoint$.deserialize(Checkpoint.scala:154)
    at org.apache.spark.streaming.CheckpointReader$$anonfun$read$2.apply(Checkpoint.scala:329)
    at org.apache.spark.streaming.CheckpointReader$$anonfun$read$2.apply(Checkpoint.scala:325)
    at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
    at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)
    at org.apache.spark.streaming.CheckpointReader$.read(Checkpoint.scala:325)
    at org.apache.spark.streaming.StreamingContext$.getOrCreate(StreamingContext.scala:852)
... 

Answer 1

更新：发现有一个开放的错误 - SPARK-5569 (https://github.com/apache/spark/pull/8955)。

在建议的提交中应用代码更改并构建 spark-assembly 后，它现在可以工作了。