Py4JJavaError：调用 o45.load 时出错。 ：java.lang.NoClassDefFoundError：org/apache/spark/sql/sources/v2/Strea

Posted 2023-03-10

【中文标题】Py4JJavaError：调用 o45.load 时出错。 ：java.lang.NoClassDefFoundError：org/apache/spark/sql/sources/v2/StreamWriteSupport

【英文标题】：Py4JJavaError: An error occurred while calling o45.load. : java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/StreamWriteSupport

【发布时间】：2021-01-30 09:56:32

【问题描述】：

我是 Kafka 和 pyspark 的新手。我想做的是将一些数据发布到 Kafka 中，然后使用 pyspark-notebook 获取这些数据以进行进一步处理。我在 docker 上使用 Kafka 和 pyspark-notebook，我的 spark 版本是 2.4.4。要设置环境并获取数据，我正在运行以下代码：

os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars dependency/elasticsearch-hadoop-7.6.0.jar  --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.4.4,org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.4 pyspark-shell'

spark = SparkSession.builder \
.master("local[*]") \
.appName("reactor_ds_data_streaming") \
.config("es.nodes", "http://10.29.18.124") \
.config("es.port","9200") \
.getOrCreate()

kafka_msg = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "10.29.18.124:9092") \
        .option("subscribe", "reactor-raw") \
        .option("startingOffsets", "latest") \
        .option("failOnDataLoss", "False") \
        .load()

当我运行最后一步 (kafka_msg) 时，我收到以下错误：

Py4JJavaError: An error occurred while calling o45.load.
: java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/StreamWriteSupport
    at java.base/java.lang.ClassLoader.defineClass1(Native Method)
at java.base/java.lang.ClassLoader.defineClass(ClassLoader.java:1017)
at java.base/java.security.SecureClassLoader.defineClass(SecureClassLoader.java:174)
at java.base/java.net.URLClassLoader.defineClass(URLClassLoader.java:550)
at java.base/java.net.URLClassLoader$1.run(URLClassLoader.java:458)
at java.base/java.net.URLClassLoader$1.run(URLClassLoader.java:452)
at java.base/java.security.AccessController.doPrivileged(Native Method)
at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:451)
at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:589)
at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522)
at java.base/java.lang.Class.forName0(Native Method)
at java.base/java.lang.Class.forName(Class.java:398)
at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.nextProviderClass(ServiceLoader.java:1209)
at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNextService(ServiceLoader.java:1220)
at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNext(ServiceLoader.java:1264)
at java.base/java.util.ServiceLoader$2.hasNext(ServiceLoader.java:1299)
at java.base/java.util.ServiceLoader$3.hasNext(ServiceLoader.java:1384)
at scala.collection.convert.Wrappers$JIteratorWrapper.hasNext(Wrappers.scala:43)
at scala.collection.Iterator.foreach(Iterator.scala:941)
at scala.collection.Iterator.foreach$(Iterator.scala:941)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at scala.collection.TraversableLike.filterImpl(TraversableLike.scala:255)
at scala.collection.TraversableLike.filterImpl$(TraversableLike.scala:249)
at scala.collection.AbstractTraversable.filterImpl(Traversable.scala:108)
at scala.collection.TraversableLike.filter(TraversableLike.scala:347)
at scala.collection.TraversableLike.filter$(TraversableLike.scala:347)
at scala.collection.AbstractTraversable.filter(Traversable.scala:108)
at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:649)
at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:194)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.sources.v2.StreamWriteSupport
at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:471)
at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:589)
at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522)
... 43 more

我不知道究竟是什么问题，如果有人可以帮助我了解我应该解决什么来解决它，我真的很感激。

谢谢

【问题讨论】：

您的 jars 放在哪里？请确保它存在于您提交应用程序的主机中。我看到您使用的是 Pyspark shell，默认情况下，它使用部署模式作为客户端

由于您使用的是结构化流媒体，因此您不需要 streaming-kafka-0-8_2.11:2.4.4 但您真的需要 Spark 吗？ Kafka 有原生 Python 库

【参考方案1】：

我发现了问题所在。我还需要在我的包目录中添加“kafka-client”jar 文件。