Connect to a remote spark cluster (Databricks Connect)

[crackityjones666]

Hi!

I need to refactor a python api that uses pyspark to execute a sql statement on Spark cluster in Azure Databricks. It uses Databricks connect (https://docs.databricks.com/dev-tools/databricks-connect.html) to initialize the connectivity to the cluster.

Is it possible to achieve something like this in c#, I would love to have an mvc core api!

Thanks!

[imback82]

We are working with Databricks engineers to be able to use C# with databricks-connect. We will update this thread when we get more info. cc: @elvaliuliuliu

[crackityjones666]

Thank you! My best bet would be to keep this python api for now? Or is there another way to have my web service trigger a SQL query against a Databricks model? Also, is there an ETA around this?

Thank you very much for such a quick reply!

[shadowmint]

I have posted about this on https://ideas.databricks.com/ideas/ as recommended here and in other threads on this topic.

However, is this really up to the databricks folk to implement?

I'm not sure it is... my understanding is that the databricks connect libraries for python, r, etc. rely on a vendored set of .jar files that are shipped when you install databricks connect.

ie. After having performed the basic setup:

python3 -m venv env
source ./env/bin/activate
pip install databricks-connect
databricks-connect configure
databricks-connect test

Then you can find the location of the installed 'modified' spark implementation using:

(env) M:tmp doug$ databricks-connect get-spark-home
/Users/doug/dev/wok/tmp/env/lib/python3.5/site-packages/pyspark
(env) M:tmp doug$ databricks-connect get-jar-dir
/Users/doug/dev/wok/tmp/env/lib/python3.5/site-packages/pyspark/jars

However, it seems currently what is lacking is an implementation in Microsoft.Spark to allow those values to be used as overrides in the library? Specifically the jars folder.

Am I mistaken in my understanding?

I've seen multiple threads saying this is something that needs to be implemented by the Databricks folk, but my understanding is that, that is not correct.

The databricks connect implementation for arbitrary consumers already exists, it is simply:

1. Not supported in Microsoft.Spark

2. Not documented how to set it up on https://docs.databricks.com/dev-tools/databricks-connect.html

tldr; Seems like game of hand ball here to me, and I'm not sure it's fair to say this is being blocked by the Databricks side...

[imback82]

@shadowmint, .NET for Apache Spark works with databricks connect as long as you don't use UDFs written in C#.

Databricks connect works by sending the logical plans to the server side and while the plans are parsed on the server side, the PythonFunction.pythonExec value (which is the .NET Spark Worker) passed was not being respected (overwritten as a Python executable).

I had a call with a Databricks engineer and got a confirmation that the server side needs to be updated. I will update this thread with the latest status.

Thanks.

[shadowmint]

I did not try that until 10 seconds ago, because I can't imagine how I was supposed to guess that I needed to do that... but I stand corrected; it does work.

Running:

./bin/spark-submit --class org.apache.spark.deploy.dotnet.DotnetRunner --master local /Users/doug/dev/dotnet-spark-test/SparkTest.App/bin/Debug/net5.0/microsoft-spark-3-0_2.12-1.0.0.jar dotnet /Users/doug/dev/dotnet-spark-test/SparkTest.App/bin/Debug/net5.0/SparkTest.App.dll

From the folder one up from that returned by databricks-connect get-jar-dir (which returns site-packages/pyspark/jars, you should run this from site-packages/pyspark) does appear to actually spin up the local driver and execute the job.

This is very useful!

It would really helpful if the documentation here:

• https://docs.microsoft.com/en-us/azure/databricks/dev-tools/databricks-connect
• https://github.com/dotnet/spark/blob/main/deployment/README.md#databricks

Was updated to explain this is what is required.

[dbeavon]

@shadowmint Did you try executing a UDF? In general databricks-connect always worked, it is just the UDF's that were problematic.

@imback82 Hopefully your patch on 6.6 was pulled into another version of the azure-databricks runtimes. That one predates spark 3.

I tested on DBR 8.1 and it is unsupported because of the spark version. DotnetRunner is disallowing spark 3.1.0

Exception in thread "main" java.lang.IllegalArgumentException: Unsupported spark version used: 3.1.0-SNAPSHOT. Normalized spark version used: 3.1.0. Supported versions: 3.0.0, 3.0.1
        at org.apache.spark.deploy.dotnet.DotnetRunner$.validateSparkVersions(DotnetRunner.scala:172)
        at org.apache.spark.deploy.dotnet.DotnetRunner$.main(DotnetRunner.scala:46)
        at org.apache.spark.deploy.dotnet.DotnetRunner.main(DotnetRunner.scala)

I tested in DBR 7.3 LTS and that fails to run the UDF as well (null reference):

JvmException: org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 10.0 failed 4 times, most recent failure: Lost task 1.3 in stage 10.0 (TID 34, 10.129.253.16, executor 1): 
org.apache.spark.api.python.PythonException: System.NullReferenceException: Object reference not set to an instance of an object.



   at Microsoft.Spark.Utils.UdfSerDe.<>c.<DeserializeType>b__10_0(TypeData td) in /_/src/csharp/Microsoft.Spark/Utils/UdfSerDe.cs:line 265
   at System.Collections.Concurrent.ConcurrentDictionary`2.GetOrAdd(TKey key, Func`2 valueFactory)
   at Microsoft.Spark.Utils.UdfSerDe.DeserializeType(TypeData typeData) in /_/src/csharp/Microsoft.Spark/Utils/UdfSerDe.cs:line 261
   at Microsoft.Spark.Utils.UdfSerDe.Deserialize(UdfData udfData) in /_/src/csharp/Microsoft.Spark/Utils/UdfSerDe.cs:line 160
   at Microsoft.Spark.Utils.CommandSerDe.DeserializeUdfs[T](UdfWrapperData data, Int32& nodeIndex, Int32& udfIndex) in /_/src/csharp/Microsoft.Spark/Utils/CommandSerDe.cs:line 333
   at Microsoft.Spark.Utils.CommandSerDe.Deserialize[T](Stream stream, SerializedMode& serializerMode, SerializedMode& deserializerMode, String& runMode) in /_/src/csharp/Microsoft.Spark/Utils/CommandSerDe.cs:line 306
   at Microsoft.Spark.Worker.Processor.CommandProcessor.ReadSqlCommands(PythonEvalType evalType, Stream stream) in D:\a\1\s\src\csharp\Microsoft.Spark.Worker\Processor\CommandProcessor.cs:line 188
   at Microsoft.Spark.Worker.Processor.CommandProcessor.ReadSqlCommands(PythonEvalType evalType, Stream stream, Version version) in D:\a\1\s\src\csharp\Microsoft.Spark.Worker\Processor\CommandProcessor.cs:line 102
   at Microsoft.Spark.Worker.Processor.CommandProcessor.Process(Stream stream) in D:\a\1\s\src\csharp\Microsoft.Spark.Worker\Processor\CommandProcessor.cs:line 43
   at Microsoft.Spark.Worker.Processor.PayloadProcessor.Process(Stream stream) in D:\a\1\s\src\csharp\Microsoft.Spark.Worker\Processor\PayloadProcessor.cs:line 82
   at Microsoft.Spark.Worker.TaskRunner.ProcessStream(Stream inputStream, Stream outputStream, Version version, Boolean& readComplete) in D:\a\1\s\src\csharp\Microsoft.Spark.Worker\TaskRunner.cs:line 144
	at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:598)
	at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$2.read(PythonUDFRunner.scala:81)

I will keep trying but I'm not very hopeful.

What are the chances that azure-databricks will help me work thru this in a support ticket? Is there an formalized support for .Net yet from databricks and azure-databricks?

[dbeavon]

I went back to databricks v.7.3.11 to try to tackle the null reference error. I found a more detailed message in the standard error output of the worker. It appears that the not net worker daemon is running, but when it tries to load the assembly with my UDF it cannot find it.

Below are those details.

21/03/30 16:54:31 INFO CodeGenerator: Code generated in 16.945226 ms
[2021-03-30T16:54:31.6816578Z] [0220-235044-tics837-10-129-253-15] [Info] [DaemonWorker] Started .NET DaemonServer with port 43097.
[2021-03-30T16:54:31.6892143Z] [0220-235044-tics837-10-129-253-15] [Info] [DaemonWorker] New connection accepted for TaskRunner [1]
DotnetWorker PID:[1955] Args:[-m pyspark.daemon] SparkVersion:[3.0.1]
[2021-03-30T16:54:31.7292568Z] [0220-235044-tics837-10-129-253-15] [Info] [DaemonWorker] Pool snapshot: [NumThreads:1], [NumTaskRunners:1]
[2021-03-30T16:54:31.7323062Z] [0220-235044-tics837-10-129-253-15] [Info] [TaskRunner] [1] Starting with ReuseSocket[True].
[2021-03-30T16:54:31.7451367Z] [0220-235044-tics837-10-129-253-15] [Info] [ConfigurationService] 'DOTNETBACKEND_PORT' environment variable is not set.
[2021-03-30T16:54:31.7451968Z] [0220-235044-tics837-10-129-253-15] [Info] [ConfigurationService] Using port 5567 for connection.
[2021-03-30T16:54:31.7465955Z] [0220-235044-tics837-10-129-253-15] [Info] [JvmBridge] JvMBridge port is 5567
[2021-03-30T16:54:31.7474901Z] [0220-235044-tics837-10-129-253-15] [Info] [JvmBridge] The number of JVM backend thread is set to 10. The max number of concurrent sockets in JvmBridge is set to 7.
21/03/30 16:54:31 INFO CodeGenerator: Code generated in 26.104194 ms
[2021-03-30T16:54:31.7917440Z] [0220-235044-tics837-10-129-253-15] [Warn] [AssemblyLoader] Assembly 'mySparkApp, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null' file not found 'mySparkApp.dll' in 'C:\Users\dbeavon\source\repos\mySparkApp\bin\Debug\netcoreapp3.1,__dbutils_lib_,/databricks/spark/work/app-20210330165328-0000/0,/usr/local/bin/spark-dotnet/Microsoft.Spark.Worker-1.0.0/'
21/03/30 16:54:31 INFO CodeGenerator: Code generated in 21.048256 ms
[2021-03-30T16:54:31.8092221Z] [0220-235044-tics837-10-129-253-15] [Error] [TaskRunner] [1] ProcessStream() failed with exception: System.NullReferenceException: Object reference not set to an instance of an object.
   at Microsoft.Spark.Utils.UdfSerDe.<>c.<DeserializeType>b__10_0(TypeData td) in /_/src/csharp/Microsoft.Spark/Utils/UdfSerDe.cs:line 265
   at System.Collections.Concurrent.ConcurrentDictionary`2.GetOrAdd(TKey key, Func`2 valueFactory)
   at Microsoft.Spark.Utils.UdfSerDe.DeserializeType(TypeData typeData) in /_/src/csharp/Microsoft.Spark/Utils/UdfSerDe.cs:line 261
   at Microsoft.Spark.Utils.UdfSerDe.Deserialize(UdfData udfData) in /_/src/csharp/Microsoft.Spark/Utils/UdfSerDe.cs:line 160

Given that it can't find my assembly (within /FileStore/DotNetForSpark) I think that the problem is almost certainly an issue with the DOTNET_ASSEMBLY_SEARCH_PATHS environment variable.

[imback82]

@imback82 Hopefully your patch on 6.6 was pulled into another version of the azure-databricks runtimes. That one predates spark 3.

Yes, it seems that the patch is there (without the patch, it wouldn't even reach to the . NET worker).

DotnetRunner is disallowing spark 3.1.0

We don't support Spark 3.1.0 yet, but it will come in 2.0 release in coming weeks.

I think that the problem is almost certainly an issue with the DOTNET_ASSEMBLY_SEARCH_PATHS environment variable.

Yes, most likely.

[dbeavon]

@imback82 Thanks for the feedback.

I'm still stumped. Going to take a break for now. I set DOTNET_ASSEMBLY_SEARCH_PATHS in a million places with no luck

DOTNET_ASSEMBLY_SEARCH_PATHS=/FileStore/DotNetForSpark

I'm not convinced this environment variable is making a difference. For some reason the code is ignoring it (eg. see the path that is referenced in message - [Warn] [AssemblyLoader])

I am still able to submit work via the jobs API and I do that regularly from ADF.
https://docs.databricks.com/dev-tools/api/latest/jobs.html

I've never had problems with my environment variable when using the jobs API. The environment variable is necessary and is properly respected. I'm using the same value of the environment variable that I use with the jobs API.

I also briefly tested the uploading of my UDF assemblies with "--files" but can't get that working. I'm not sure what to expect with that option, since I've never used it.

The messages we get from the worker aren't very helpful to help figure out what is going wrong. In the very least it might be nice to print the value of DOTNET_ASSEMBLY_SEARCH_PATHS (possibly in the [Error] [TaskRunner] ). Also it might be nice if the System.NullReferenceException wasn't displayed so frequently. By now everyone knows that implies a problem with the search paths, but we should take things a step further and actually say something specific along those lines. IE. Why can't we preceed it with an explicit assertion that has a better failure message?

This may be a problem that only affects me, or only affects windows. I will probably try to pick this back up again another time.

[MikeRys]

To answer your second question: We are working on making .NET for Spark usable for all Spark engines and experiences. However, each company that offers Spark services has to make its own decision on whether or not to include it out of the box. If they do so, we are happy to help. At this point, I am aware of Azure HDInsight and Azure Synapse workspaces to offer .NET for Spark out of the box. I do like the Synapse experience, especially since it provides nice integration with the nteract based notebooks in Synapse. Best regards Michael

[briancuster]

how do you interact with the spark cluster on synapse using .Net for spark application? I'm trying to get started using .Net for spark and am unsure how to get things going with a synapse cluster from my Windows 10 device.