spark connect and databricks connect support?

[indraniel]

Will geni support Spark 3.5's remote connections to clusters via spark connect and databricks connect ?

[behrica]

I got databricks connect working on Clojure.
But I did not manage to have the spark connect jar and the spark jars working in the same VM.

Even by carefully checking the versions.
-> I din not get Geni and databricks cnnect working together.

[alza-bitz]

Hi folks,

I managed to get Spark Connect working from a Clojure REPL, just using interop and without Geni. Based on my investigation I think I have a high level understanding of what's going on, why it's not working with Geni and some ideas for how to fix it.

Firstly, see here for two working examples:

1. Spark connect session https://github.com/alza-bitz/playpen-clojure/blob/551b675b66b1c9875beb08393a1540f38d2ba5f2/src/spark_connect_session.clj
2. Spark classic session https://github.com/alza-bitz/playpen-clojure/blob/551b675b66b1c9875beb08393a1540f38d2ba5f2/src/spark_classic_session.clj

Before trying these examples yourself, please see the comments about the classpath jars.

Investigation: Spark API

The implementation of org.apache.spark.sql.SparkSession now seems to be delegated. https://github.com/apache/spark/blob/master/sql/api/src/main/scala/org/apache/spark/sql/SparkSession.scala seems now to be a proxy/facade for one of two implementations:

1. org.apache.spark.sql.classic.SparkSession
   /home/vscode/.m2/repository/org/apache/spark/spark-sql_2.12/3.5.4/spark-sql_2.12-3.5.4.jar
   https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala

2. org.apache.spark.sql.connect.SparkSession
   /home/vscode/.m2/repository/org/apache/spark/spark-connect-client-jvm_2.12/3.5.4/spark-connect-client-jvm_2.12-3.5.4.jar
   https://github.com/apache/spark/blob/master/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala

Btw, I couldn't even find the api doc for case 2. The api doc here is for 1. only, since there is no "remote" method on the builder: https://spark.apache.org/docs/3.5.4/api/scala/org/apache/spark/sql/SparkSession$$Builder.html

Investigation: General Thoughts

1. If both jars are on the classpath, which implementation is preferred at runtime? How can we choose between them, or are you only supposed to include one or the other depending on your use case?

2. There is a related PR here [SPARK-49698][CONNECT][SQL] Add ClassicOnly annotation for classic only methods. apache/spark#49801 and it seems that Scala code is supposed to use the @classiconly annotation to handle this, for example sparkContext is not available with a Spark Connect session. But presumably this annotation solution won't work from Clojure..

So for Clojure and Geni, another approach is needed. Perhaps the simplest thing to do is not call (.sparkContext session) and then Geni would hopefully work for both cases? Although, you'd still need to make sure that only the appropriate library of the two options is on your classpath.

Investigation: Geni Specific Thoughts

Loading the zero-one.geni.core namespace tries to load the zero-one.geni.defaults namespace, but this will fail if the spark connect jar is on the classpath because as mentioned above (.sparkContext session) is not supported. If the core ns won't load, we can't use the Geni core functions e.g. read-csv even if we obtained our spark session via other means.

Suggestions

1. Drop the default session entirely
2. Keep the default session, but you have to explicitly require the default ns to load it
3. In the Spark Connect case, any session creation cannot call (.sparkContext session) as explained above. Further, because it would be difficult or impossible (?) to determine the SparkSession implementation, and SparkContext seems to be deprecated anyway (read https://spark.apache.org/docs/3.5.4/spark-connect-overview.html at the end of the page), then just don't call it?

[behrica]

Thanks for this investigation.

I think as well there is a problem with the design of the default session, but not sure how to solve it.

I got as well Databricks connect working in Clojure using pure Java classes. Even with a Azure Databricks cluster incl. authentication. and I am open to any PR on improving GENI on this.

I would like to get (your ?) feedback on the overall "suitability" of the Geni approach...

Before "Databricks connect" I was not sure at all, how even to use Geni with Azure Databricks which would be my target.
(lets assume it would work as designed)

I have no idea how to get "REPL driven development" with Geni working. I cannot "login" into a Azure Databricks compute cluster to install clojure or a repl.

I managed at one point in time to "blindly" create a uber-jar and got it executed as a job.
But to get the "dependencies" of that was a nightmare.... neither did I find any documentation how to even do a "job jar" which has access to the spark session of databricks when executed. (neither did I find this for Java)
Specially documentation which depencies such Jar should have (and not have !!!)
I only found this page: https://learn.microsoft.com/en-us/azure/databricks/jobs/how-to/use-jars-in-workflows
-> Is this even still supported or "legacy" ?

The "default session" of Geni makes to me any experimenting very difficult, as it somehow "goes in the way"
On "Azure Databricks" specifically, this is a "blocker":
#357

which I fixed in a small fork:
https://github.com/behrica/geni

Now with Databricks connect the situation improves in my view, at least REPL driven working seems to be feasable.

For your information:
I got this scala instruction working, using Clojure with Java Interop:
https://learn.microsoft.com/en-us/azure/databricks/dev-tools/databricks-connect/scala/

I managed as well to convert the Spark Dataframe into a tech ml dataset, so I can further work on it in pure clojure (only feasable for small enough data, which is my case)
https://github.com/techascent/tech.ml.dataset/blob/master/src/tech/v3/libs/spark.clj

The code in that ns does not work as-is neither, it was developed against an older Spark version.
But a small change was enough, I have it somewhere.

This latest approach I tried does work without the need for Geni at all.
(but indeed only for small data)

[behrica]

"Databricks connect" and "Spark connect" are not entirely the same. I think bot give at the end an interface of "org.apache.spark.sql.SparkSession"

[behrica]

One other remark, again for "future of geni" ?
A big part of Geni makes the scala based spark API integrated better in Clojure, correct ?
But Spark has now as well a Java API.

[alza-bitz]

Hey @behrica thanks for the response.

It seems there are multiple overlapping problems, may I suggest these are separated to move forward?

Apologies in what follows, I'm not knowledgeable on Databricks so please forgive me.

Here is a plan to move forward with four steps:

1. Get your default config change PR approved and merged to allow classic mode to work with Databricks.

2. Decide what to do about the default session.

Keeping it will require a solution that works for both classic and connect. However, after merging point 1. the default session doesn't hold any special config any more, so it doesn't have as much purpose? It only saves one function call to create a session. So, maybe the default session can just be removed entirely? This also makes sense with classic vs connect. Any default session can only be for one or the other mode, and this choice depends on the classpath (unless we try and be clever). So overall based on these constraints, I would say to remove it (and also update the docs).

If however there is some strong opinions to keep the default session (for example to help beginners or for setup speed in exploring data), my suggestion is to separate it completely from the core ns. So it would only be loaded by doing a require for a new default-session ns. What you get when that is loaded would depend on what is on the classpath, either the classic or connect jar (I am assuming it's illegal to have both). From my experiments it seems the same session creation code could work for both, just don't call .sparkContext on it.

The var holding the atom is defined and initialised as a helpful string in a separate ns default. If the default-session ns is loaded, the default session is created and reset! as the new value of the atom. This way, the user can get a helpful error if trying to use the non-session form of the core functions and they forgot to include default-session ns.

I wrote some example code to illustrate this idea:

https://github.com/alza-bitz/playpen-clojure/blob/spark-connect-investigation/src/geni/client.clj

3. After the default session situation is resolved, add support for spark connect.

4. Then add support for Databricks connect. Sorry, I didn't investigate yet how similar or different the Databricks connect is to plain spark connect. But in any case, I would suggest to add support for plain spark connect first, then Databricks.

I would appreciate your thoughts on the above plan @anthony-khong 🙏

Re: your point about the Java spark api instead of Scala spark api, AFAIK the Java api doesn't work with spark connect yet.

[behrica]

@alza-bitz
Not sure what you mean by "Java API not working".

public static Dataset<Row> query (SparkSession spark) {

Dataset<Row> table = spark.read().table("xxxxx");    
    return table.filter("`epublication-year`==2024").select("doi");
        }

}

This work for me where spark session is comming from 'DatabricksConnect'
Of course again an other topic, but I am getting really hooked into using Java / Clojure together as explained here:
https://calva.io/virgil/

This (to me) reduces the need of "Clojure libraries which are only wrappers of existing Java libraries".
(= the motivation to work on such wrappers ...)

[alza-bitz]

Not sure what you mean by "Java API not working".

@behrica the phrase I used was "AFAIK the Java api doesn't work with spark connect yet."

I'm only talking about Spark connect here, not Spark classic. My previous understanding was that the Spark connect client is only supported from the Scala, PySpark or Golang apis. I didn't think it was implemented for Java yet, because when I searched I could only find an example for Scala: https://stackoverflow.com/questions/76209004/using-spark-connect-with-scala

However I since realised this is most likely wrong - if it's possible from Scala (and Clojure), then it's almost certainly also possible from Java using the same connect jar on the classpath.

Also, I'm not talking about Databricks either since I don't know about that (yet). I'm only discussing vanilla spark connect. In an earlier comment you mentioned "Databricks connect and Spark connect are not entirely the same" so perhaps we should have two separate Github issues, one for vanilla spark connect and another for Databricks connect?

[alza-bitz]

Hi @behrica @anthony-khong

Some further thoughts that didn't occur to me until after I wrote my earlier message "plan to move forward with four steps"..

1. Apologies but it didn't register with me until now, I noticed that the builder's build method isn't build() but getOrCreate()

After reading the doc, it seems that yes indeed Spark is doing it's own caching and management of a default. It would appear to be the singleton pattern although it doesn't explicitly say that: https://spark.apache.org/docs/3.5.4/api/scala/org/apache/spark/sql/SparkSession$$Builder.html#getOrCreate():org.apache.spark.sql.SparkSession

This lower-level singleton is an extra factor I hadn't considered. Based on this learning, if there is to be another higher-level singleton in Geni (for the default session) then there is more to consider..

2. Further, on the use of a Clojure atom for managing the default session.

I was reminded after re-reading the Clojure docs that Clojure atoms are only working as expected when the state being managed is one of Clojure's own persistent data structures e.g. seq or map: https://clojure.org/reference/atoms

"As with all reference types, the intended use of atom is to hold one of Clojure’s immutable data structures."

"Atoms are an efficient way to represent some state that will never need to be coordinated with any other"

The api doc for Builder.buildOrCreate states "In case an existing SparkSession is returned, the non-static config options specified in this builder will be applied to the existing SparkSession."

Although this implies that buildOrCreate will always return the same (but maybe modified) instance, any calls to it can also modify the Geni default session outside the scope of the atom reset! function (since the atom is just holding the same Java object instance). This defeats the point of using an atom for the default session, because any external changes can't be co-ordinated. This general problem is discussed in another context here: https://stackoverflow.com/questions/48332746/how-to-guard-a-mutable-object-against-concurrent-read-write-access-in-clojure

So, in conclusion, based on points 1. and 2. I think there is a compelling case to drop the default session implemented with atom.

However, there are still use cases for a default session (simpler for beginners, quick data explorations) and there is also the unit tests to consider.

With this in mind, I made another branch with a simpler, modified solution:

https://github.com/alza-bitz/playpen-clojure/blob/spark-connect-investigation-part-two/src/geni/core.clj

Please read the comments in the code, but in general I believe this simpler solution covers all the requirements (unless I missed something).

Thanks! 🙏

[behrica]

@alza-bitz
Do ypu have a working deps.edn.
I can not find any combnation of

         com.databricks/databricks-connect {:mvn/version "16.1.0"}
         org.apache.spark/spark-core_2.12      {:mvn/version "3.5.4"}
         org.apache.spark/spark-sql_2.12      {:mvn/version "3.5.4"}
         org.apache.spark/spark-mllib_2.12      {:mvn/version "3.5.4"}
         org.apache.spark/spark-hive_2.12      {:mvn/version "3.5.4"}
         org.clojars.behrica/geni {:mvn/version "0.0.42-CB.2"}

versions which is not failing wit " no-classdef found" or similar exceptions on start.
I use it against a Cluster with Databricks version 16.1, so databricks-connect with same version is suggested.
The cluster has spark 3.5.x which is therefore needed as well.

(without geni, I have some sucesss)

[alza-bitz]

@behrica

https://github.com/alza-bitz/playpen-clojure/blob/spark-connect-investigation-part-two/deps.edn

https://github.com/alza-bitz/playpen-clojure/blob/spark-connect-investigation-part-two/src/spark_classic_session.clj

https://github.com/alza-bitz/playpen-clojure/blob/spark-connect-investigation-part-two/src/spark_connect_session.clj

Note 1. these cover vanilla spark connect, not Databricks

Note 2. please read comments in the code about what you have on the classpath and making sure that the spark connect server is started or stopped, if you want to switch between classic and spark connect modes.

Note 3. The above code has Geni on the classpath, but I am not loading it, because it doesn't currently work with Spark connect (I already explained why in my earlier comments).

[behrica]

Thanks for the info.
As I need to use databricks connect, it did not really help me.

As my needed use case is "databricks" and I have working code without geni,
I give up at this point.
I will use java API, which does work for me with databricks connect.

Changing randomly versions, I get various errors all pointing into "incompatible versions",
so typically "AbstractMethodError" or "NoClassDefFound".

As the potential solution depens as well completely on "my Cluster" (and its versions) I do not expect to find a "solution" somewhere.
And if so, it would likely break on the next update.

Leaving "geni" out, gives me a least the assurance that either the spark developers or databricks developers "deliver and maintain" something working.
While the Geni project is largely unsupported and quite old as well.

And it would likely support the open source spark, and open source spark connect , which is not my use case.

[behrica]

I have now the "impression" that Geni cannot be made work with Spark/Databricks connect.
The "foreseen way" to work with databrics connect is to not have any of the usual spark-dependencies on the classpath, but only the "databricks-connect jar" which contains itself the spark core classes.
but only a subset, for example not the ml related classes which Geni needs.

So to me Geni needed to be shrinked in order to support Databricks connect