Are the Spark.NET 3.5.1 works in Polyglot notebook for UDF examples?

[GeorgeS2019]

#1199

@grazy27
4. User-Defined Functions

Previously in older version, it is not possible to work in the UDF notebook. Not sure if the latest 3.5.1 overcomes this limitation

https://github.com/Apress/introducing-.net-for-apache-spark

[grazy27]

@GeorgeS2019 It does not.

It tries to get entire declaring type, which is anonymous assembly that only exists within interactive session context.
And fails with an exception, as result:

Code expects declaring assembly for UDF to be accessible:

Notebooks without UDFs and broadcasts work fine.

[GeorgeS2019]

Do U have recommendation to the polyglot team what changes they need to make to accommodate spark.net UDF to work?

I am going to invite them to address your findings

Good job

[grazy27]

Certainly, there are ways to achieve this.
In the context of Dotnet.Spark, we can determine whether the current code is running in an interactive environment and manually include library bytes in the command payload (or alternatively, save the library as a file and load it on the worker side).

PoC:

For this PoC to work, we also need to hardcode executing assembly as 1.dll

Polyglot team could do something similar on their side as well, but it'd have to be strongly names and uploaded to GAC, or explicitly target '%DOTNET_WORKER_DIR%'.
This would only work locally, and other dotnet.Interactive wrappers would need the same fix.

This issue falls primarily within Dotnet.Spark's scope of responsibility. This issue should be taken into account when addressing migration from BinarySerializer to somewhere else, since the responsible code will have to be rewritten.

[GeorgeS2019]

dotnet/interactive#3797

[wudanzy]

I guess pyspark did save all bytecode? We can do the same in spark.NET.

[grazy27]

I’ve made some progress in understanding the interaction, though it’s still not entirely clear. Here’s what I’ve gathered so far:

Dotnet.Spark contains a separate solution for this already, Microsoft.Spark.Extensions.DotNet.Interactive. It registers into .net interactive pipeline and adds all executables via 'spark.addFile' functionality. Does pretty much the same I did in my PoC, but also handles more special cases and uses native spark function for it.
That's clever, we don't have to serialize IL code and send it to spark.worker.
On a worker side, there's dedicated code for handling this as well.

The only problem i faced is that it doesn't work, it relies on Dotnet.Interactive API Utility.DisposableDirectory API, using which throws MethodAccessException.

According to this issue in dotnet interactive, is not ready for 1.0 stable release.

Thanks @GeorgeS2019 for creating issue in interactive, let's wait and see if they can point us with this exception.

[jonsequitur]

The constructor is private. But also this class will likely be made internal as part of API stabilization. The .NET Interactive API shouldn't really be responsible for providing this kind of utility code to consumers. I'd recommend cloning the functionality as needed.

https://github.com/dotnet/interactive/blob/6ae92454e21a8793cb289e6d70508bd5a4e73418/src/Microsoft.DotNet.Interactive/Utility/DisposableDirectory.cs

[grazy27]

Hello @jonsequitur,

I've investigated it in more detail, and have a few findings and questions.
Would you mind reviewing the proposed solution for 2 and providing advice for 3? Thank you in advance!

1. DisposableDirectory

The DisposableDirectory issue is resolved by using kernel.RegisterForDisposal for manual directory cleanup, which aligns with the current API.

2. Collection of referenced nugets and their dependencies

For NuGet dependency collection, ISupportNuget has been removed, breaking the previous method of getting ResolvedPackageReferences (SupportNugetWrapper, that is utilized in PackageResolver).

Workaround: I am now subscribing to the PackageAdded event to collect ResolvedPackageReference objects, which we need for probing paths and package metadata.
Limitation: This approach requires our extension to be loaded before any other NuGet packages to guarantee a complete collection.
Question: Is this event subscription the recommended pattern, or is there a more robust method to get all resolved packages, including those potentially loaded before our extension?

3. The middleware pipeline behavior for submissions appears to have changed, there's a timing issue

In AssemblyExtension there's a middleware registration, that first propagates down the pipeline await next(....), and only then accesses ScriptContext and attaches a new compiled Submission-x.dll file to spark.

I'm not sure how it worked before, but now it does not - by the time we can access dll, submission have already failed as there was no dll.

If we add the DLL before await next(...), kernel.ScriptState is stale and provides the DLL from the previous submission (n-1), causing a version mismatch on the worker.

If I first attach dll, and then propagate the pipeline - ScriptContext contains Submission(X-1).dll - Previous cell value, and this one doesn't work as well, as dotnet.spark worker fails with version mismatch.

Question: What is the correct pattern to inject logic after compilation but before execution to prepare the Spark Context with the current submission's assembly?

[jonsequitur]

Question: Is this event subscription the recommended pattern, or is there a more robust method to get all resolved packages, including those potentially loaded before our extension?

You can see the loaded metadata references like this:

[grazy27]

This approach will only work for basic scenarios, as more complex nuget libraries may contain references to native C++ dlls, or may require additional non-binary files, like .xml\.json\.config, and they will not be copied in such case

Current implementation includes code for extracting those references from nuget metadata and sending them to remote worker as well, but this additional is only available in nuget package object

[jonsequitur]

You probably need to get at the PackageRestoreContext if you want the rest of that information. It looks a bit like the .NET for Spark implementation you pointed to was based on this code, several refactors ago.

We might need to expose an API to make the PackageRestoreContext accessible. While the type is public, the instance associated with a given kernel is scoped in such a way that getting a reference to it would be difficult:

https://github.com/dotnet/interactive/blob/c125401b58a9069d4b186600365aedba7e3d04c0/src/Microsoft.DotNet.Interactive.PackageManagement/KernelExtensions.cs#L16

[grazy27]

Aha, then i can try to get it from internal collection of disposables from Kernel using reflection magic

This will win you some time to add a way to get it properly from the context, and we can migrate to normal approach next release

[jonsequitur]

3. The middleware pipeline behavior for submissions appears to have changed, there's a timing issue

I would need more context here. Do you have a stack trace? Is the DLL that's missing the one that's supposed to be emitted from the CSharpKernel's compilation?

I'm not sure where the timing issue could be introduced. There's no parallelism in the scheduler.

Question: What is the correct pattern to inject logic after compilation but before execution to prepare the Spark Context with the current submission's assembly?

This code looks reasonable. Another approach that you could try would be to subscribe to CommandSucceeded events rather than using middleware.

[grazy27]

Got it, it's a dealbreaker from dotnet perspective.
Is there any chance to change smth in dotnet.interactive to allow that to work?
Mb events like 'OnBeforeExecutionStarted\OnAfterCompiled'?

How would you recommend to approach solving this?

[jonsequitur]

This worked before without this ability so it should still be possible.

Trying to do this in middleware might be part of the problem, though it's not clear to me. An occasional gotcha to be aware of here is that if you submit code to the CSharpKernel during the execution of code within the same kernel, you end up with two different assemblies, one of which (generated by the inner submission) is abandoned when the outer submission completes. This is because compilations are immutable and only one (the last to complete) is left available for subsequent submissions to build on.

So rather than trying to do this in middleware, I would recommend subscribing to all CommandSucceeded events at the kernel level and after each one, capture the assembly.

[grazy27]

Thanks for your help!
Two different but identical ones sounds Ok, as long as they're identical
If there's a conflicting assembly executed, and current dismissed - that should also do, as they'll have different signatures

I'll have a look at events to subscribe, and will try to understand how it worked before, as according to all logic it shouldn't have - assembly is needed before command execution, there's little help from it afterwards

The only option that comes to my mind - if it only worked with deferred execution:
Cell 1

var df = spark.Read().Parquet........UDF.....

Cell 2

df.collect()
Or df.print

Then assembly will be generated and added before execution, as execution happens in a different command

If that's the case, I'll migrate to CommandSucceeded to make it more transparent and stable

@GeorgeS2019 do you have any info on how it used to work previously?

[jonsequitur]

Two different but identical ones sounds Ok, as long as they're identical
If there's a conflicting assembly executed, and current dismissed - that should also do, as they'll have different signatures

They're not identical. They will typically be completely different. Let me give an example.

In this example, the outer submission is all of the code in the cell. The inner submission is the code on lines 8 and 9 that are being sent to the same CSharpKernel instance.

The complete sequence of events here looks like this:

1. This code (let's call it the "outer submission") is sent to the CSharpKernel:

   using Microsoft.DotNet.Interactive;
   using Microsoft.DotNet.Interactive.Commands;
   using Microsoft.DotNet.Interactive.Events;
   
   var csharpKernel = Kernel.Root.FindKernelByName("csharp");
   
   var innerResult = await csharpKernel.SendAsync(new SubmitCode("""
       var innerVariable = "Inner variable"; 
       Console.WriteLine(innerVariable);
       """));
   
   innerResult.Events

2. This code is compiled and executed (a single step within C# Scripting).

3. The executing code sends another command to the CSharpKernel which compiles and executes the following code (let's call it the "inner submission"):

   var innerVariable = "Inner variable"; 
   Console.WriteLine(innerVariable);

4. The CSharpKernel completes the handling of the SubmitCode command carrying the inner submission and stores the resulting assembly as the most-recently-executed assembly in the chain.

5. The CSharpKernel completes the handling of the SubmitCode command carrying the outer submission and stores the resulting assembly as the most-recently-executed assembly in the chain. This effectively discards the inner submission's assembly.

As you can see in the variable view at the bottom of the screenshot, the variable innerVariable is not present. It was only declared in the inner submission's assembly, which is gone by the time the outer submission is done being executed.

So the two different assemblies represent a fork. They will typically contain no overlapping code.

[grazy27]

Interesting, thanks for sharing this combination!
Dotnet.spark worker only needs an entire assembly, so if we subscribe to CommandSucceeded, in your case we'll get 2 different invocations and 2 different assemblies, right?
And one of them is discarded afterwards and global state is not updated
In such case, we can anyway attach both of them to the spark context, and discarded one will just never be referenced from worker, and cleansed after run anyway - I'll make a small test to doublecheck that

Developers will do 5 hours of debugging to save 10 min of reading the documentation 🙂
I found an official guide for UDFs in interactive .net.spark, and they specifically mentioned to run UDF creation in one cell, and its invocation in another, so we can proceed with implementation