Apache Spark无法在群集上反序列化`TopicPartition`

Question

我使用Spark 2.4和scala 2.11.12从卡夫卡中读取批处理。因此，我的build.sbt文件具有以下依赖关系。

"org.apache.spark"  %% "spark-core" % sparkVersion % "provided",
"org.apache.spark"  %% "spark-sql" % sparkVersion % "provided",
"org.apache.spark"  %% "spark-sql-kafka-0-10" % sparkVersion

我还使用了sbt-assembly插件来制作我的应用程序的胖jar。当在本地spark上部署这个jar文件时，它工作得很好，如下所示，$FAT_JAR指向我的程序集文件：

./spark-submit --class $MAIN_CLASS --master local --driver-class-path $FAT_JAR $FAT_JAR

但是当我在集群上部署它时(即使工作者和主机都在同一台机器上)，它会抛出关于TopicPartiton反序列化问题的异常。

我如何在集群上运行：

./spark-submit \
  --master spark://spark-master:7077 \
  --class $MAIN_CLASS \
  --driver-class-path $FAT_JAR \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0 \
  $FAT_JAR

我也尝试过--jars，我确信worker和master都有kafka-client的保存版本，即2.0.0

异常日志：

Caused by: java.io.InvalidClassException: org.apache.kafka.common.TopicPartition; class invalid for deserialization
        at java.io.ObjectStreamClass$ExceptionInfo.newInvalidClassException(ObjectStreamClass.java:169)
        at java.io.ObjectStreamClass.checkDeserialize(ObjectStreamClass.java:874)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2043)

为什么spark无法反序列化TopicPartition？我如何解决它？

Answer 1

我已经找到解决方案了。正如我将SPARK_DIST_PATH设置为$(hadoop classpath)一样，它包含了kafka-client-0.8，它不同于在spark-sql-kafka-0-10中使用的kafka-client-2.0.0。我刚刚使用了hadoop包含版本的spark和未设置的SPARK_DIST_PATH来解决这个问题。

无论如何，我希望spark.executor.userClassPathFirst和spark.driver.userClassPathFirst在总体上帮助解决这个问题，但现在它们还处于实验阶段。