Py4JJavaError: An error occurred while calling o45.load. : java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/StreamWriteSupport

Question

I'm new to Kafka and pyspark. What I'm trying to do is publish some data into the Kafka and then using the pyspark-notebook to reach those data for further processing. I'm using Kafka and pyspark-notebook on docker and my spark version there is 2.4.4. to set up the environment and reaching data I'm running the following code:

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()

When I run the final step (kafka_msg) I get the following error:

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

I do not know what is the problem exactly and I really appreciat if someone can help me to understand what I should fix to solve it.

Thanks

Where are your jars placed?Please make sure it's present in the host from where you are submitting the application.I see you are using Pyspark shell and by default, it uses deploy mode as client — Subash
– Subash, Commented Oct 16, 2020 at 3:15
Since you're using Structured Streaming, you don't need streaming-kafka-0-8_2.11:2.4.4 but do you really need Spark? There's native Python libraries for Kafka — OneCricketeer
– OneCricketeer, Commented Oct 21, 2020 at 13:33

Arya · Accepted Answer · 2020-10-23 19:48:49Z

4

I found what was the problem. I need to add "kafka-client" jar file as well in my packages directory.

answered Oct 23, 2020 at 19:48

Arya

811 silver badge6 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

Py4JJavaError: An error occurred while calling o45.load. : java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/StreamWriteSupport

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related