structured streaming

[joques]

Hi all, how do I get Spark.jl to read a stream from and write to Kafka? I need help finding documentation on that.

[dfdx]

As a general rule, you can follow the official documentation with a few context-specific modifications. I don't have a proper setup to test it myself, but I'd start with this doc. According to it, first you need to add spark-sql-kafka library to the Spark session. Something like this:

spark = SparkSession.builder()
        .appName("...")
        .master("spark://ip:7077")
        .config("spark.jars", "/path/,/path/to/another.jar")
        .getOrCreate()

then just follow the examples on the linked page (Python or Scala API):

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
  .option("subscribe", "topic1") \
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

Please, let me know if it worked for you.

[joques]

Thanks for your response. I followed your suggestion but I am still getting the same error. This is what I did.

I downloaded spark-sql-kafka into a folder. Let's call it spark-jars. Then I updated the spark session as follows

spark = SparkSession.builder.appName("Main").master("spark://IP:7077").config("spark.jars", "/absolute/path/to/spark-jars/spark-sql-kafka-0-10_2.12-3.5.2.jar").getOrCreate()

Next, I added the streaming code as follows

stream_df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "st-streaming-session").load()

But I am still getting the following exception:
Exception in thread "main" org.apache.spark.sql.AnalysisException: Failed to find data source: kafka. Please deploy the application as per the deployment section of Structured Streaming + Kafka Integration Guide.

Note that for the kafkaesque.boostrap.servers, I tried the domain name, the IP address and localhost. All three are throwing the same exception. What am I doing wrong?
Regards

[dfdx]

Can you post the full stacktrace?

Edit: or even better the full log or its pieces related to JAR sending to workers.

[joques]

Below is the full stack trace of the error I get

Exception in thread "main" org.apache.spark.sql.AnalysisException: Failed to find data source: kafka. Please deploy the application as per the deployment section of Structured Streaming + Kafka Integration Guide.
	at org.apache.spark.sql.errors.QueryCompilationErrors$.failedToFindKafkaDataSourceError(QueryCompilationErrors.scala:1567)
	at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:645)
	at org.apache.spark.sql.streaming.DataStreamReader.loadInternal(DataStreamReader.scala:158)
	at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:145)

and below is the latest log from the master for my latest run

24/09/24 05:59:24 INFO Master: Start scheduling for app app-20240924055923-0004
 with rpId: 0
24/09/24 06:02:27 INFO Master: 196.216.167.103:36600 got disassociated, removin
g it.
24/09/24 06:02:27 INFO Master: COMPSCSRV04.nust.na:33649 got disassociated, rem
oving it.
24/09/24 06:02:27 INFO Master: Removing app app-20240924055923-0004
24/09/24 06:02:27 WARN Master: Got status update for unknown executor app-20240
924055923-0004/0
24/09/24 06:02:27 WARN Master: Got status update for unknown executor app-20240
924055923-0004/1
24/09/24 06:04:53 INFO Master: Registering app Main
24/09/24 06:04:53 INFO Master: Registered app Main with ID app-20240924060453-0
005
24/09/24 06:04:53 INFO Master: Start scheduling for app app-20240924060453-0005
 with rpId: 0
24/09/24 06:04:53 INFO Master: Launching executor app-20240924060453-0005/0 on
worker worker-20240923111146-196.216.167.102-45805
24/09/24 06:04:53 INFO Master: Launching executor app-20240924060453-0005/1 on
worker worker-20240923111223-196.216.167.105-36589
24/09/24 06:04:54 INFO Master: Start scheduling for app app-20240924060453-0005
 with rpId: 0
24/09/24 06:04:54 INFO Master: Start scheduling for app app-20240924060453-0005
 with rpId: 0

[joques]

Stack trace
Here is what happened, the most recent locations are first:

geterror() @ core.jl:544
_jcall(::JavaCall.JavaObject{Symbol("org.apache.spark.sql.streaming.DataStreamReader")}, ::Ptr{Nothing}, ::Type, ::Tuple{}; callmethod::typeof(JavaCall.JNI.CallObjectMethodA)) @ core.jl:482
_jcall(::JavaCall.JavaObject{Symbol("org.apache.spark.sql.streaming.DataStreamReader")}, ::Ptr{Nothing}, ::Type, ::Tuple{}) @ core.jl:475
jcall(::JavaCall.JavaObject{Symbol("org.apache.spark.sql.streaming.DataStreamReader")}, ::String, ::Type, ::Tuple{}) @ core.jl:371
load @ streaming.jl:54
DotChainer @ chainable.jl:13

[dfdx]

For some reason, the added jar file is not propagated to your workers. Debugging this issue via Github issues isn't easy, but as a short in the night let's try to add Kafka connector as a package instead of a plain jar:

spark = SparkSession.builder()
        .appName("...")
        .master("spark://ip:7077")
        .config('spark.jars.packages', 'org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1' )   # <-- this line changed
        .getOrCreate()

[joques]

It keeps throwing the same error. Suspecting that the added jar was not propagated, I created the same folder on each worker node and added the jar. But I am still getting the same error.
Another thing I tried was to add the jar file to the spark-defaults.conf on each node as follows

spark.jars path/to/spark-sql-kafka.jar

But the error persists. I am just confused now.

[dfdx]

What's your cluster config, e.g. cluster manager, remote or local workers, etc.? Usually, Spark creates project directories on workers dynamically, so putting jar files to some directory on workers beforehand doesn't take effect.

[joques]

I am running spark in a standalone mode on a 3-node cluster. I have one master and two workers. Each spark instance is on a different node. The kafka instance is running on the same node as the master. That's pretty much it.

[joques]

I made a few changes. I reverted the session back to spark.jars as follows:

spark = SparkSession.builder.appName("SoftwareTools").config("spark.jars", "/path/to/spark-sql-kafka-0-10_2.12-3.5.2.jar").master("spark://IP:7077").getOrCreate()

I also upgraded the version of scala to 2.12.2 to match the version of the jar file. But I am now getting the error below. Any hint?

Exception in thread "main" java.lang.NoClassDefFoundError: Could not initialize class org.apache.spark.sql.kafka010.KafkaSourceProvider$
	at org.apache.spark.sql.kafka010.KafkaSourceProvider.org$apache$spark$sql$kafka010$KafkaSourceProvider$$validateStreamOptions(KafkaSourceProvider.scala:338)
	at org.apache.spark.sql.kafka010.KafkaSourceProvider.sourceSchema(KafkaSourceProvider.scala:71)
	at org.apache.spark.sql.execution.datasources.DataSource.sourceSchema(DataSource.scala:233)
	at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute(DataSource.scala:118)
	at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo(DataSource.scala:118)
	at org.apache.spark.sql.execution.streaming.StreamingRelation$.apply(StreamingRelation.scala:36)
	at org.apache.spark.sql.streaming.DataStreamReader.loadInternal(DataStreamReader.scala:169)
	at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:145)

[dfdx]

It looks like Kafka jar is now being picked up, but versions of libraries still mismatch. Let's try to align them. In your local installation of Spark.jl, find a file called pom.xml and change this:

    <java.version>1.11</java.version>
    <scala.version>2.13</scala.version>
    <scala.binary.version>2.13.6</scala.binary.version>

    <spark.version>[3.2.0,3.2.1]</spark.version>

to this:

    <java.version>YOUR ACUTAL JAVA VERSION</java.version>
    <scala.version>2.12</scala.version>
    <scala.binary.version>2.12.2</scala.binary.version>

    <spark.version>3.5.2</spark.version>

Then in Julia console type:

] build

or

] build Spark

Once the build is complete (and successful), try to test your code again.

[joques]

I made the suggested change. But my code is in a Pluto notebook. Do you think it will rebuild automatically if I restart the notebook? Spark is not available in the global space in Julia.

[joques]

After following the suggestion above and successfully building the Spark package, I am now observing two types of error:

1 - When I use the spark.jars.packages config option I get this error

Exception in thread "main" java.lang.NoClassDefFoundError: Could not initialize class org.apache.spark.sql.kafka010.KafkaSourceProvider$
	at org.apache.spark.sql.kafka010.KafkaSourceProvider.org$apache$spark$sql$kafka010$KafkaSourceProvider$$validateStreamOptions(KafkaSourceProvider.scala:338)
	at org.apache.spark.sql.kafka010.KafkaSourceProvider.sourceSchema(KafkaSourceProvider.scala:71)
	at org.apache.spark.sql.execution.datasources.DataSource.sourceSchema(DataSource.scala:233)
	at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute(DataSource.scala:118)
	at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo(DataSource.scala:118)
	at org.apache.spark.sql.execution.streaming.StreamingRelation$.apply(StreamingRelation.scala:36)
	at org.apache.spark.sql.streaming.DataStreamReader.loadInternal(DataStreamReader.scala:169)
	at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:145)

When I revert back to the configuration using spark.jars with the path to the jar file, I get the following error

xception in thread "main" java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArraySerializer
	at org.apache.spark.sql.kafka010.KafkaSourceProvider$.(KafkaSourceProvider.scala:599)
	at org.apache.spark.sql.kafka010.KafkaSourceProvider$.(KafkaSourceProvider.scala)
	at org.apache.spark.sql.kafka010.KafkaSourceProvider.org$apache$spark$sql$kafka010$KafkaSourceProvider$$validateStreamOptions(KafkaSourceProvider.scala:338)
	at org.apache.spark.sql.kafka010.KafkaSourceProvider.sourceSchema(KafkaSourceProvider.scala:71)
	at org.apache.spark.sql.execution.datasources.DataSource.sourceSchema(DataSource.scala:233)
	at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute(DataSource.scala:118)
	at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo(DataSource.scala:118)
	at org.apache.spark.sql.execution.streaming.StreamingRelation$.apply(StreamingRelation.scala:36)
	at org.apache.spark.sql.streaming.DataStreamReader.loadInternal(DataStreamReader.scala:169)
	at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:145)
Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.serialization.ByteArraySerializer
	at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
	... 10 more

[joques]

I added the Kafka clients package and it finally worked. Thanks for your assistance on that.
Now I am getting an error with the df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") code. I looked up the code, but the function is not in there. Is it inherited? I'd like to extract the key and values from the topic

PS: This is the error

UndefVarError: `selectExpr` not defined

[joques]

May I please have a concrete example of how to read and write from and to the kafka stream based on the Spark.jl implementation? I can't find any example in the package doc.

I did the following but it is not working.

rec_evts = stream_df.select(Column("key"), Column("value"))
query = rec_evts.writeStream().outputMode("complete").format("console").start()
query.awaitTermination()

Thanks