Language-Agnostic UDF Protocol for Spark

Summary

The current Spark UDF execution model is tightly coupled to specific runtimes (primarily PySpark) and relies on ad-hoc inter-process communication protocols, making it insufficient for the multi-language architecture introduced by Spark Connect in version 3.4.
The proposed language-agnostic UDF model is built on three pillars: planning UDFs based on execution shape rather than programming language, a structured IPC protocol using Protocol Buffers, and a declarative worker specification that decouples the Spark engine from language runtimes—enabling remote and containerized UDF workers outside the Spark cluster.
The proposal provides a clean migration path for existing PySpark UDFs while expanding support to the growing Spark Connect client ecosystem written in Go, Rust, Swift, and .NET, with a roadmap to cover more UDF types and integrations.

Language-Agnostic UDF Protocol for Spark

User-defined functions (UDFs) are a fundamental extension mechanism in Spark, enabling users to express logic that goes beyond built-in operators. However, the current UDF execution model is not enough for the multi-language architecture in Spark Connect. Existing implementations are tightly coupled to specific runtimes (e.g., PySpark), rely on ad-hoc communication protocols, and embed language-specific logic into planning and execution.
This proposal aims to introduce a language-agnostic UDF execution model for Spark. It is built around three core ideas: (1) planning UDFs based on execution shape rather than programming language, (2) a structured execution IPC protocol and (3) a declarative worker specification to decouple the Spark engine from language runtimes.
To provide a clean and extensible foundation for IPC-based UDF execution across languages, this proposal aims to standardize multi-language support and establish a sustainable direction for the UDF ecosystem.
Talk By: Haiyang Sun, Senior Software Engineer, Databricks ; Tian Gao, Senior Software Engineer, Databricks ;

Chapters

FAQs

What is Spark Connect and why does it change how UDFs need to work?

Spark Connect, introduced in Apache Spark 3.4, separates the Spark client from the Spark server, allowing clients to send logical plans over a Protobuf-based protocol from any language. While this enables open-source Spark Connect clients in Go, Rust, Swift, and .NET, UDF support does not come automatically—the existing UDF model is tightly coupled to PySpark's runtime, so a language-agnostic UDF model is needed.

What are the three pillars of the language-agnostic UDF proposal?

The three pillars are: planning UDFs based on execution shape (scalar, batch, aggregate, table) rather than programming language; a structured IPC protocol using Protocol Buffers for stable, schema-versioned inter-process communication; and a declarative worker specification that decouples the Spark engine from language runtimes, enabling workers to run remotely or in containers outside the Spark cluster.

What types of user-defined functions does this proposal cover?

This video clarifies that the proposal covers all custom user code executed as part of a Spark query—including scalar functions, batch execution functions, group functions, aggregate functions, table functions, and some streaming APIs—not just simple functions registered with the UDF keyword. The focus is on UDFs that run outside the executor JVM and communicate through inter-process communication.

How does the proposal maintain backward compatibility with existing PySpark UDFs?

The proposal includes a migration path that keeps existing PySpark UDFs compatible during the transition to the new language-agnostic model. PySpark served as the reference design for the new protocol because it is the most mature existing UDF runtime, and the new model is designed as a clean, extensible foundation that PySpark can migrate to without breaking existing code.

Full transcript

[00:07] Thank you for all for being here. My name is Haiyan. I'm a software engineer from Databricks and I work in the lake or team. Our team works on sandboxing UDF execution in Databricks runtime. And I'm here today with my colleague Tian and Yitong. They are from the PySpark
[00:21] team and they uh they are also Spark committers and Tian is also the Python code developer. So, in this talk, we will present our ongoing work in Apache Spark to to extend the existing user-defined
[00:33] function support to more languages. Before getting into the architecture, I want to ask you for my a few questions. So, how many of you use Spark? Okay, that's fair enough. And how many
[00:45] of you use Spark but never use UDF? Okay, then you are in the right uh wrong track. I'm joking. So, um most of you have used UDF. It's an
[00:57] important feature in Spark. So, let me start with the motivation of this work. When I say UDF in this talk, actually I I don't mean just a simple function that you usually register with a UDF keyword and call by name in SQL.
[01:10] I mean any custom user code that Spark executes as part of the query. So, this could include scalar functions, batch execution functions, group functions, aggregate, table functions, or even some streaming APIs.
[01:24] UDFs are important because Spark's building functions cannot cover every use case. Spark already pro- provided a rich set of building functions but these are not always enough.
[01:36] So, you users often need to call external libraries, accessing external resources, run domain-specific logic, and process unstructured data. In this talk, we want to focus on UDF
[01:48] that runs outside of the executor JVM such as PySpark UDF. And this UDF run in a separate worker process. So, Spark needs to communicate with the worker through inter-process communication.
[02:05] So, another question is why we are doing this now. So, let me introduce some background about the Spark Connect project. Spark Connect was a new architecture proposed in Spark 3.4. It separates Spark Connect from a Spark
[02:18] client from the Spark server. So, the client sends logical plans to the server using protobuf based protocol. This makes it much easier to build Spark applications because you just need to
[02:31] depend on a very thin client. Today, there are already many open source Spark Connect clients for different languages such as Go, Rust, Swift, and .NET. However, UDF support doesn't
[02:43] automatically come with Spark Connect. So, a client can easily build data frames and run SQL in Spark Connect, but supporting UDF requires more than just sending a logical plan. It also requires
[02:55] a way to execute a user user code in a client's language. That is the missing piece we want to address in this talk. So, here comes the first problem. We want to support multi-language.
[03:08] So, today, if you build a new client using Spark Connect, there's no easy way to add UDF support for that language. On the left side of this slide, I'm showing the README file from the GitHub repository for the Rust implementation
[03:20] in Spark Connect. So, everything almost worked data frame, SQL, but UDF or UDGTF are explicitly marked as not possible. And this is not a
[03:32] Rust specific problem. The main uh most of most of the existing languages implementing Spark Spark Connect do not have UDF support. Except for JVM based Scala or Java UDF
[03:44] or PySpark UDF because they already have support support of UDF before Spark Connect. As a result, all the current language would have to build the whole UDF support from scratch like PySpark has
[03:57] done in the past few years. And the second problem is that we want to achieve forward compatibility. So, Spark Connect is designed so that an older client can often keep working
[04:10] against the newer server version because the logical plan is encoded in protobuf and that is designed to be version forward compatible. But a UDF is more difficult than a
[04:24] logical plan. So, to run a UDF it usually requires us a worker implementation that can execute the UDF. And that worker implementation could require the language runtime
[04:37] dependencies, serialization, and all of this are specific to every language. So, when a Spark server is upgraded, it may need to execute UDFs submitted by older clients and sometimes you need to handle
[04:50] multiple client versions at the same time. So, without a structured or version UDF protocol, this becomes very difficult to support reliably.
[05:02] And here comes the third problem is resource management. Actually, this one has been uh is not specific to Spark Connect and it has been there for long. It's about how Spark currently execute UDF nowadays. So, today you run UDF
[05:16] inside of the executor nodes. That is together with the executor uh JVM, which means the UDF shares the same resource as the Spark tasks. So, this can cause stability stability
[05:29] problems because a heavy UDF can compete with Spark tasks for memory and CPU. It also limits UDF execution to the executor environment. So, if a UDF needs a needs to load a really large in-memory
[05:42] model or it requires a special hardware like a GPU or it requires a different runtime that is not shipped with the Spark cluster, then that UDF is usually not
[05:54] uh it's hard to support. So, we need a model where UDF workers can be managed in a more flexible way without having to hardcode that execution environment into Spark.
[06:07] And before describing our architecture, it's also uh helpful to look at PySpark because PySpark is currently most complete implementation Spark has today. PySpark UDF support many evil types
[06:21] already, covers a wide range of execution shapes, and have been optimized and hardened in production over many years. So, PySpark already give us a very important starting point. But PySpark was designed for one
[06:34] language and a one worker implementation. So, the question is now that PySpark works well for Python, but how to support more languages on top of that? And what kind of new
[06:46] abstractions are needed to make UDF execution more generic and for more languages? There are currently three uh three limitations that prevent PySpark UDF
[06:58] support from becoming a general architecture for all languages. First, PySpark has planning and optimization logic that is specific to Python. And uh um second, PySpark has a protocol
[07:12] between Spark and the worker, but the protocol is currently handwritten and position-based. It works for PySpark, but it is really difficult to version or difficult to evolve safely currently, and thus it's
[07:25] really difficult for another language to implement the same protocol. And thirdly, PySpark doesn't provide a general abstraction how worker uh for worker provisioning. So, the
[07:37] Python workers is shipped with Spark engine and started by Spark engine and the worker management logic are also bounded to the engine. So this model doesn't directly support a worker provided by a
[07:50] Spark Connect that can be any language. And this worker could also have its own runtime requirement. So while PySpark give us already a valuable valuable building blocks, it doesn't by itself solve the problem we
[08:03] want to solve here from different languages. To solve this problem, we propose in three pillars. So the first pillar is very straightforward. We want language agnostic query planning.
[08:16] So here we want to generalize what PySpark does by uh executing UDF based on its execution execution execution shape instead of based on the language. So the second pillar is that we want a
[08:29] more structured and version protocol between Spark and the UDF. Here we turn uh we turn an explicit implicit worker communication pattern into an explicit contract, so that that
[08:42] can be easily picked up by different languages. And the third pillar is about the worker specification. Actually, this is a new architectural piece we introduce here. It's supposed to be a declarative
[08:54] contract that tells Spark how to obtain Spark start, connect to validate, and manage a worker without hardcoding these details into the engine. So the first two pillars generalize and
[09:06] formalize idea PySpark already have. And the third pillar introduce the missing abstraction that allows UDF workers to become a language independent, client provided, and deployable uh component.
[09:24] So we can also talk about some details about these three pillars. The first pillar is language agnostic planning. So for query planning, PySpark already uh doesn't Spark doesn't really care what a UDF is written which language
[09:37] it's written. Either Python or R or Go. What it really cares about is the execution shape of the UDF, how its input and output fit into the rest of the query. For example,
[09:49] if it is a scalar function, it takes one input in one output row out. Or a batch function that processes a whole partition at a time. Or it could be a grouped function that that's that is
[10:02] happened after group operation and then apply per key group. Or table function or other operations. So, these shapes are the actual the factor to determine how Spark plans this query. And
[10:14] uh PySpark already have planning support built around these ideas. The goal here is to generalize this planning support. So, this so the existing work in PySpark can be reused across different languages.
[10:31] And the second pillar is about the UDF protocol. How actually Spark and the worker can communicate during the UDF execution. At a high level, most of UDF execution follow a very similar pattern. First, Spark initialize a UDF execution with
[10:44] the function body and meta metadata in the init message. And Spark then streams input batches to the worker and then the worker stream result batches back. Finally, both sides finish and clean up the execution.
[10:58] Even for some more complicated types such as aggregations, you can repeat this pattern for multi multiple times for different stages. PySpark already follow a very similar pattern. But as we said before, the protocol
[11:11] existing protocol is not so well is not so generic. And we want to make that protocol more more generic by versioning with versions. So, it's easier for other
[11:23] language to adopt adopt. And so, one idea is that we want to make make sure that the protocol is defined with protocol messages, so it automatically has schema evolution and
[11:36] it can be more robust in in terms of upgrades. And also we want to separate the protocol from the transport. The protocol we define here is a message a messages and we
[11:48] uh have semantics between it it defines the semantics between the Spark and the worker how they exchange these messages, but the actual transportation is below this level. So, the transportation is a mechanism we want to use to
[12:00] exchange those messages. Currently, um and we don't want to tie to one transport transport implementation. Currently, PySpark use socket, but here we want to offer an alternative using
[12:12] gRPC. The benefits The benefits of using a gRPC based worker is that we do we no longer have to offer some framework level support such as
[12:25] um uh back pressure or uh authentication or some uh debugabilities. So, for a new language implementing a new a UDF worker becomes much easier.
[12:37] So, the language also can rely on the gRPC service we define for the transport interface and focus on the UDF execution semantics rather than rebuilding the whole communication framework from
[12:49] scratch. And uh I also want to spend a bit more time on the third pillar because this is part that goes beyond the uh generating PySpark.
[13:03] Spark uh in Spark planning tells Spark what kind of UDF execution is needed and the introduced UDF protocol tells Spark how to communicate with the worker once the worker exists. But neither of these answers how a fundamental question,
[13:16] where does the worker come from? That is the role of the worker specification we want to talk about here. So, a worker specification is a contract between the client and the Spark. It describe how Spark can obtain a
[13:29] worker, how to start it, how to connect to it, and what protocol it speaks. And uh this is slightly different from the current PySpark model where the Python
[13:41] worker is part of Spark itself. And with this worker specification, the worker no longer has to be built into the Spark. It can be now provided by the client and
[13:53] carry and carry the specification about the runtime and dependency it needs to run, and uh what's more, in addition, this worker specification can also declare its capabilities, so Spark can manage it
[14:05] through a common interface. For example, the spec can describe capabilities like whether the work supports parallel parallelism, so the Spark knows that it can schedule multiple UDF tasks to one
[14:18] single worker, or whether the worker supports uh UDF chaining of of yielding, so Spark knows whether it can do query optimization phase that can chain multiple UDF into single node.
[14:30] So, overall, the worker specification is a key abstraction we introduce here to separate the Spark engine from language-specific implementation. And because now worker provisioning
[14:42] becomes declarative, the same UDF UDF protocol can support very different deploy deployment modes models now. So, first of all, the workers can still run as a local process next to the
[14:54] executor JVM, just like how PySpark works today. But it can also be a containerized worker where stronger isolation or custom dependencies are needed. It can be a worker using a different
[15:07] runtime version than a Spark engine. It can also be even a remote worker running outside Spark cluster, for example, on dedicated on a on a remote machine with dedicated hardware. This is why the worker spec is
[15:19] important. Now, let's look at these three pillars come together end to end. Suppose we are building a new client now and you want to add UDF support. With this architecture, you do not you
[15:32] need to rebuild the UDF execution from scratch. You can just follow the common protocol we provide providing a worker and then Spark will naturally know how to plan the UDF query and run the worker
[15:45] according to the worker specification. Now, I want to come back to the exist two existing language that supports UDF. The first is the JVM UDF. As I mentioned earlier, so these Scala or Java UDF are
[15:58] mostly the auto scope for the main part of this talk because they run currently inside the JVM. And so that and they don't require extra IPC or doesn't require extra serialization. However, inlining means
[16:11] that this UDF usually share the same class path and the same privileges as Spark as a Spark executor. So, they are limited by the resource of the executor node, too. So, in some cases, it was would also be
[16:23] useful to run this JVM based UDF in a separate process. This could help avoid problems like a class path conflicts or you want to use a different JVM version, Scala version, or you want to provide strong stronger isolation.
[16:36] This is currently not supported before, but with this proposal, this is going to be possible in the future. Also about PySpark UDF, the ultimate goal is to unify all UDF execution code
[16:48] passes including PySpark. But, we already have tons of workloads running on PySpark. We don't want to breaking existing existing workloads. That's That means that we shouldn't move PySpark onto the new protocol
[17:01] immediately. So, we The idea is that we will only we will keep both code passes for now and we will only switch the code after fully validating the both the behavior and performance for PySpark to
[17:13] make the switch. And to make the transition smoother and to avoid and reinventing the whole worker in PySpark, we will try to refactor the PySpark worker to separate the transport
[17:26] layer from execution layer logic. So, instead of building the worker, this allows to us to switch just the transport layer transportation layer from for example from socket to the GRPC, which is a new
[17:38] code path. So, where are we today? So, the project was proposed as a Spark project improvement proposal or SPARK- in short. It has been accepted
[17:50] after passing the Spark community vote. We are currently actively working on the implementation. Because this is a large project with some ambitious goal, we are shipping it incrementally.
[18:02] We start by defining the protocol and the semantics of the UDF execution. From there, we are going to using one PySpark eval type as the first end-to-end validation pass. After that, we are gradually
[18:14] expanding the coverage to more eval types. In parallel, we also want to integrate the protocol with Spark Connect and onboard more language beyond Python to Spark.
[18:26] Before ending the talk, I would would like to thank the many people who have has contributed to the initial proposal, shared ideas during discussions, and help with code reviews throughout the implementation. And so, to summarize, the main takeaway
[18:39] that by separating UDF planning, communication, and worker provisioning, we can make UDF execution in Spark reusable across languages, more resilient to version changes, and more
[18:51] flexible in where and how worker runs. Thank you very much and then happy to take questions.

Learn more about the Databricks Data and AI platform.

The information provided herein is for general informational purposes only and may not reflect the most current product capabilities or configurations.