topic Re: Spark Streaming foreachBatch with Databricks connect in Data Engineering

Spark Streaming foreachBatch with Databricks connect

olivier-soucy — Tue, 22 Oct 2024 16:53:46 GMT

I'm trying to use the foreachBatch method of a Spark Streaming DataFrame with databricks-connect. Given that spark connect supported was added to `foreachBatch` in 3.5.0, I was expecting this to work.

Configuration:
- DBR 15.4 (Spark 3.5.0)
- databricks-connect 15.4.2

Code:

import os from databricks.connect import DatabricksSession # Setup spark = DatabricksSession.builder.clusterId("0501-011833-vcux5w7j").getOrCreate() # Execute df = spark.readStream.table("brz_stock_prices_job") def update_metrics(batch_df, batch_id): size = batch_df.count() print(f"Batch size: {size}") writer = df.writeStream.foreachBatch(update_metrics).start()

Error:

File "/Users/osoucy/miniconda3/envs/lac/lib/python3.10/site-packages/pyspark/sql/connect/client/core.py", line 2149, in _handle_rpc_error raise convert_exception( pyspark.errors.exceptions.connect.SparkConnectGrpcException: (java.io.IOException) Connection reset by peer JVM stacktrace: java.io.IOException at sun.nio.ch.FileDispatcherImpl.read0(FileDispatcherImpl.java:-2) at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39) at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223) at sun.nio.ch.IOUtil.read(IOUtil.java:197) at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379) at sun.nio.ch.SocketAdaptor$SocketInputStream.read(SocketAdaptor.java:208) at sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:103) at java.io.BufferedInputStream.fill(BufferedInputStream.java:246) at java.io.BufferedInputStream.read(BufferedInputStream.java:265) at java.io.DataInputStream.readInt(DataInputStream.java:387) at org.apache.spark.api.python.StreamingPythonRunner.init(StreamingPythonRunner.scala:206) at org.apache.spark.sql.connect.planner.StreamingForeachBatchHelper$.$anonfun$pythonForeachBatchWrapper$3(StreamingForeachBatchHelper.scala:146) [...] at org.apache.spark.sql.connect.execution.ExecuteThreadRunner$ExecutionThread.run(ExecuteThreadRunner.scala:561)

Any help would be appreciated!

Re: Spark Streaming foreachBatch with Databricks connect

daniel_sahal — Wed, 23 Oct 2024 11:07:57 GMT

@olivier-soucy
Are you sure that you're using DBR 15.4 and databricks-connect 15.4.2?
I've seen this issue when using databricks-connect 15.4.x with DBR 14.3LTS.

Anyway, I've just tested that with the same versions you've provided and it works on my end.

Re: Spark Streaming foreachBatch with Databricks connect

olivier-soucy — Wed, 23 Oct 2024 14:38:39 GMT

Hi Daniel!

Thanks for getting back to me!

Yes it DBR 15.4. It's a single user access mode if that makes a difference. Here is a screenshot of the configuration:

And here are the exact version of the pyton packages I'm using:

pyspark version: 3.5.0 grpcio version: 1.67.0 DB connect version: 15.4.2 pyspark version (from db connect): 3.5.0

I'm on a M2 mac. Can't that make a difference? Have you used exactly the same code as I provided to run your tests?

Re: Spark Streaming foreachBatch with Databricks connect

daniel_sahal — Thu, 24 Oct 2024 06:10:21 GMT

@olivier-soucy
Yup, I've used the same code. Also my setup is similiar, also running on Mac but M1.

It would be best to monitor the Driver logs when running the query, as most of the errors are logged there.

Re: Spark Streaming foreachBatch with Databricks connect

olivier-soucy — Fri, 01 Nov 2024 02:50:57 GMT

Good call! I should have thought of doing that myself. It turned out that my local machine had a different python version when compared to the workers. Updating python solved this issue. Thank you so much for the help!