Skip to main content

Breaking Barriers: Building Custom Spark 4.0 Data Connectors with Python

Summary

  • Custom Data Connectors: This video teaches how to build custom Spark 4.0 data connectors using Python, enabling integration of any data source with the Databricks Data and AI platform.
  • Python SDK: Learn how to leverage the Spark Python SDK to develop robust, production-ready connectors for specialized data sources.
  • Extensibility: Discover how custom connectors extend the Databricks ecosystem to support unique data integration needs beyond built-in connectors.

Breaking Barriers: Building Custom Spark 4.0 Data Connectors with Python

Watch: Breaking Barriers: Building Custom Spark 4.0 Data Connectors with Python
Talk by Ashish Saraswat, Resident Solutions Architect, Databricks ; Sourav Gulati, Senior Resident Solutions Architect, Databricks
Building a custom Spark data source connector once required Java or Scala expertise, making it complex and limiting. This left many proprietary data sources without public SDKs disconnected from Spark. Additionally, data sources with Python SDKs couldn't harness Spark’s distributed power. Spark 4.0 changes this with a new Python API for data source connectors, allowing developers to build fully functional connectors without Java or Scala. This unlocks new possibilities, from integrating proprietary systems to leveraging untapped data sources. Supporting both batch and streaming, this API makes data ingestion more flexible than ever. In this talk, we’ll demonstrate how to build a Spark connector for Excel using Python, showcasing schema inference, data reads/writes and streaming support. Whether you're a data engineer or Spark enthusiast, you’ll gain the knowledge to integrate Spark with any data source — entirely in Python.
Here’s more to explore:
Production ready data pipelines for analytics and AI: https://www.databricks.com/solutions/data-engineering
The Big Book of Data Engineering: https://www.databricks.com/resources/ebook/big-book-data-engineering-2nd-edition
See all the product announcements from Data + AI Summit: https://www.databricks.com/events/dataaisummit-2025-announcements

Chapters

FAQs

How do you build custom Spark connectors?

This video provides a step-by-step guide to building custom Spark 4.0 data connectors using Python to integrate specialized data sources with Databricks.

What languages support Spark connector development?

According to this video, you can build Spark 4.0 connectors using Python, making custom data integration more accessible to Python developers.

How can developers build custom data connectors for Spark-based platforms?

Spark 4.0 provides standardized connector APIs that enable developers to implement custom data sources that integrate seamlessly with the SQL, DataFrame, and streaming interfaces. These connectors can leverage Spark's distributed execution model and query optimization, allowing organizations to connect proprietary systems to the Databricks Data and AI platform without developing separate ETL infrastructure. Building connectors enables organizations to reduce integration complexity when working with non-standard or legacy data systems.

How do you optimize performance when implementing custom Spark connectors?

High-performance connectors push filtering and projection down to source systems to minimize data transfer, implement intelligent partitioning to distribute work across Spark executors, and support caching to avoid redundant source system queries. Connectors should provide statistics that enable Spark's query optimizer to make better decisions about join ordering and resource allocation. Testing connectors against realistic workloads ensures they scale appropriately as data volumes and query complexity grow.

What governance capabilities are needed for custom data connectors?

Custom connectors must respect the Databricks Data and AI platform's access control and audit logging to ensure data access through connectors is tracked and governed consistently with other data access. Integration with Unity Catalog enables connectors to enforce column-level and row-level security, preventing sensitive data from being exposed to unauthorized users. Proper implementation ensures custom integrations do not circumvent governance controls put in place to protect sensitive data.

Full transcript

0:00 Introduction to the problem

Welcome everyone. So let's start. So have you ever been in a situation where your client asked you to process a lot of Excel files and uh they have a lot of data? Well, I was in a position where I had to process a lot of Excel files and they had a lot of data and client wanted to process them using Spark. So at that time we did not had any solution to do the distributed processing and leverage Spark's fast performance.
So what I did is I proposed them a solution that okay we can build a connector for you but that would require the Java or Scala implementation and as soon as I mentioned that we will implement in Java client wasn't happy about this one they said no no Java or any JVM based languages so well then I work with some of the workarounds and the reason was very simple for them not to have the Java all of their team they had the data scientists and data engineers they were working with Python.
Most of their technology stack was built around Python and in that moment we realized like if there was an API or something that we could do in Python it would be very nice. If you have ever been into the similar situation then this session is for you. Today we are going to show you how you can build your own custom connectors for your own data sources to read and write data to them using Spark. But that will be in pure Python implementation and this is enabled by Apache Spark's new API that has been released few days ago.
So welcome to the session. I am Ashi Saraswat. I am from datab bricks professional services team and I work from London and hello everyone gulati I'm senior resident solutions architect at datab bricks and I'm the part of datab bricks professional services team as well and I'm based in London too and along with showing you the overview of the data source API we'll also show you an in-depth demo where you can see how the API works in practice so what we are going to talk today so we will first start with the brief overview of the new API of the data sources API that has been available so far.
Then we will talk about what has been newly added into the Python API. Then we will gradually move on to the implementation details what we need to do in or what we need to do actually to if we want to build our own connector and then we will see a demo by Sor. So moving ahead. So as we all know Apache Spark is a unified data processing engine and it supports a lot of data sources. Some of them are inbuilt like you already have been processing data you from the CSV files from JSON files from park files.

2:28 Apache Spark data source API

All of these kind of a connectors they are inbuilt available from Spark. And if you ever have a need to provide support for your own data source then we can implement Apache Sparks data source API which has two versions version one and version two but for that we need to know Java or Scala because that API only supports the implementations in Java and Scala and using that API there are many vendors out there who has already built the connectors like Apache Cassandra's Spark connector Neo 4j connector that's available to us.
So whenever we write a spark program and we say spark read. format so that format is the kind of a implementation or the name for your data source connector. So now with that one what's the new thing that has been released in the pi spark API and we already have the API. So why the new API? So over the period of the time people realized that the new API people realized that the existing API it's only limited to the sky and Java developers.
So what the Apache Spark team has done they have provided the similar functionality in the Apache Spark 4 and that gives you the capability to build your connectors using Python. So there is no need to have the JVM based implementation for your own data source connectors. The new API is designed to be simple and easy to use and it encourages the Python developers community to build their own connectors. And the good thing is you can have your own existing libraries.
If you have implemented something already to read from your connectors using Python, you can reuse the same code and refactor it to be workable with the new API. And it supports all of the most of the Apache Sparks features like distributed processing of your data, the distributed scan of your data, filter push down, predicate push down. One thing to note here is that this particular API is not intended to replace the existing Java and Scala API.
If you have a very complicated requirement for your data sources, there are some intricacies in your data sources, then you can go ahead and still implement the existing data source Java and Scala API. So the question is like okay the new API has been released why we should talk about it and what kind of a benefits it brings for us. So first thing is that the with the existing API you need to have a knowledge of Java and Scala and I personally believe the technology should be inclusive.
Everyone should have an opportunity to build the software or to bring their ideas to life into their choice of technology and I believe Apache Spark is carrying forward that mission very well and with this new API what happens until now there was some kind of a limitation for Python developers and now it gives them the capability to build their own connector using the Python programming language and that's to in purely Python Python no need of having any JVM element in your connector.
So Python developers relied mostly until now some kind of a workaround using the for each or using the UDF to provide some kind of a distributed capabilities and process their own custom data sources. Now with this API we have a standard and efficient way of doing the things. So enough of the introduction. Let's go ahead and see what it needs to implement a connector. What are the components and the classes that we need to know if we have to build the connector for our own data source.
So yeah, let's start. Before we can build or start implement the API for our own connector, we need to know our data sources first. The API is quite flexible and you can implement the full API or you can have the partial implementation for your own use case. So if you have a requirement to process the data in batches collecting the files then we can implement the batch side of the things. If we have a need to process the data incrementally in a continuous fashion then we can implement the stream related components and if you want to do both then we can implement both of them.

7:15 Components of a data source

So to implement these to provide the capability and to build your connector there are certain classes that we can implement and all of these classes are available into a new module which is pispark. SQL data source. So within that module we will find all of the relevant classes and the main component of this implementation is your data source class. It has some methods and based on your requirement you can either choose for batch related processing or you can implement the stream related methods.
So you have the classes accordingly the for the reader writer and the stream reader and stream writer. So let's start with the reader implementation and for this one we are going to take an example of Excel data source. So how we can read the Excel files and how we can write data to the Excel files. So to build a Excel data source as I mentioned earlier what we need to do is we need to implement the data source class and within that class there are some methods.

8:46 Implementing batch and stream

So first thing first if I'm doing the implementation how spark will know and it will detect the classes that provide that implementation. So there is the method name which should be unique for the spark and you can implement that method. You can return any kind of string but just make sure that it is a unique string and it doesn't conflict at the runtime with other data sources. Then the next thing is I want to provide a schema because spark data frame is a structured API.
I want to attach some kind of a schema to it. So we have two options. So whenever we say spark read dot format and then we call the schema method, that's where you are providing your own custom schema. If you don't want to do that, you know your data sources and you have some capability to infer the schema, that kind of a logic goes here. So within the schema method either you can return a hard-coded schema if your data source always has fixed schema or what we can do is we can read some sample files infer the schema and return that one.
After we have name schema what we need to do is we need to implement the read method. This is the actual time how I'm going to read the my read my data from my data source. So for that one there is a read reader method within the data source API and as I mentioned data source class is the binding API binding class for all of your readers and writers. So within that method we just return the object of your data source reader class which is used for the batch read and this read reader class is responsible for reading the each partition's data as we see on the next slide.
So here what we need to do is we are creating this Excel data source reader class extending the data source reader the data source reader class and there we have two key methods. First one is the partition method and second one is the read method. Partition method is really critical and important here. So if you want to provide the distributed capability. So let's say if you have a data source and you want spark to process the data parallelly you can write your logic here.
So in our case we have excel and if I have multiple files and each file being processed by the single partition by the single executor in parallel I can in this method write some logic determine all the list of all the files and return the path of all the list. So within the read method that will be passed and then you can get the path and start processing it. reader method is quite simple. So before that also we need to make sure that if we don't provide any implementation for the partition method that's fine it will still work.
The problem the only thing is that all of the data will be read into one partition and it will slow down your data processing. You won't get the benefit of extra executors available into your cluster. For the read method the implementation is simple. Think about it how you will read one single file and write the logic for that particular one. If you have another data source, if you are making chunks of that data source, think about it how you will independently read the one chunk of your data source and that kind of a logic goes into the reader.
So we have built the batch reader and I am bit more ambitious. I want to implement the stream reader as well. I want to provide streaming capabilities. So there are few additional methods. We already implemented the data source class. Within the same class we extend the wait. Yeah. Sorry. So within the same class we implement or we we extend that stream reader method and in that one we return the stream reader class object.
And what we do in that stream reader class is we have the read method. We have the other methods like partition. But there are two additional methods into that class which in the batch they are not available. So for my stream I need to know when I'm processing the data the stream might stop it might fail. I need to have a way to start the processing from the previous point of failure. That's called the checkpointing mechanism which is available in this part.
So these two methods initial offset and the latest offset helps you with that one. When the stream start first time the initial offset method is called. Whenever the stream fails and the next time you restart it then the latest offset is called based on your data sources. You can determine what is the thing or like what you want to return from this method so that stream can proceed further. And then after that you have the same two methods partition and read.
So the same kind of a logic can go here but if you have some different requirement for the stream then you can provide your own custom logic here. More or less the similar logic works everywhere. So we have now seen the how you can build the stream reader and batch reader as well. So let's move on with the how you can write your data to the Excel files. So for writer as well. So we start with the same data source you see on the screen the class it has the same name schema reader stream reader methods for the writer we add additional method writer and that writer method returns the object of your data source writer class.

14:14 Building custom writers

So for writing the data there are few things that we need to take care of. First thing is that think about it how you will write one single file. So when you write the one single file that's pretty s straightforward but when you want to write multiple files what happens based on the partitioning for each partition the right method is called but there are few things that can happen during the production your right process might fail it might have some corrupt data introduced into your system to handle that one we have two additional methods commit and the abort.
So when everything has been processed successfully by your data source then the commit method is called. If something fails in between or let's say you have five executors processing five files four are successful one fails in that scenario you have the abort method that's called. So you can provide your logic in the commit for any kind of a post cleaning operations and in the about method you can provide the logic to handle any kind of a failure scenario.
Let's say you wrote some files you want to clean them up because one of the tasks failed. So you can do that one as well and you might have observed when working with park other data sources in your directory when you write UC success file that file is being like being created by this commit method usually. So we have seen that we can build the reader, we can build the writer as well. So now my implementation is done.
I might implement it as a library and give it to someone. But the question is how I start using that one into my code. How I read the files using that particular data source. So it is fairly simple. We need to first register. We need to let Spark know that these are the classes that I have implemented at the runtime. You need to look for them. So I just say spark dod data source dotregister and you provide the class name.
Please note this is the class name not the string and that's it at the runtime your data source is available to the spark. So as soon as you start doing spark read dot format and you provide excel spark will be able to recognize that it has the implementation and use all those methods to read and write data from your data source. So the on the next slide as well you can see for the right we have the similar way. So as we saw the API is really very simple and that's where the title of this talk comes in.

17:11 Practical demonstration

So until now people felt some limitations that okay I don't know Java or I don't know Scala but I have the idea I want to build the connector for my own data source and they find it bit difficult either they have to learn the language or rely on something else. So with this new API, Apache Spark is breaking that barrier. You can implement your own connector while staying while being stay into the same Python eco system and I know the theory is bit boring so I won't take much time and I invite my colleague Sorup for the demo.
All right. Uh thank you so much. I really enjoyed the overview of this feature. I hope you did as well. So now let me switch over to the demo so that we can validate that it's not just a set of slides or it actually works. U all right here I am in my datab bricks workspace and as we discussed in this example we are going to process excel files with the pispark data source API. uh just would like to call out one thing why Excel so because it's one of the most prominent data format that you see in the market lot of business users would like to use it and until now I have seen lot of examples where developers has to do workarounds to process it in pispark it could be converting it to CSV or doing some udfs but now we can do it in the native way and that's what we're going to show you now so let's start with looking at the data it.
So for the demo purpose, we have created a set of files and I'll download one of them and show it to you how it looks. But before that, I'll just run this to in the interest of time uh I would like to mention one thing. Doesn't matter how many times you run it. Uh whenever you're on the stage, anxiety just kicks in. Okay. Uh so I I'll go to this file and I'll show you how it looks like. Let me zoom in a bit. So it's it's an Excel file with few set of columns and couple of rows and most of the files looks like the same.
And in this example, we're going to process them. However, you can actually argue that I do not need a new API to process it in distributed manner. we can actually read this uh Excel files using a Python library for example pandas dataf frame and I can convert it to spark data frame which will allow me to do the distributed processing on it. However, if you think about it that implies that you are reading all the data set on the driver first and then shuffling it over to the executors.
Which means if you have a very large data set, you need a very heavy driver and then you'll be shuffling all that data which is something that we always try to avoid in Spark world. And we know that Spark actually supports distributed scan. So why not build a capability where we can read all the files directly on the executors and we can process them in parallel. And that's what we're going to show you now how we achieve this.
So let's start with an example first and then I'll go to the implementation. So I just executed this uh you you have seen this in slide I'm uh registering my data source with this method and I'll show you the implementation of this Excel data source class and then we are we are starting with the batch reader files here I'm using spark readme uh spark read method and I'm using the format to provide my data source in this case it's excel I'm passing couple of u uh options And I'm using the partition ID method to see in what partitions my files were read.
So in my case the fi there were five files and I can see they were read in separate partitions. I don't have to shuffle the data. They I got five partitions to start with. So now let's go and see the implementation of it. So I'll start with the data source itself. You already saw it in the slides but I'll go over a couple of key methods again. So the first one is name and we called it Excel and this is what I used in my example here.
And the other method is schema. So this returns the default schema. In case user does not provide any schema, this is what will be used. But in my case, the schema differed a bit. So I provided that in my example. Also we have seen that there are some data sets that provide schema inference for example CSV. So in case we want to enable that we can let user pass that option here and all the options that user passes are available inside the data source and you can use that option if user has asked for it and can build your inference logic and then for reader stream reader writer and stream writer there are separate methods which return their respective objects and those objects comes from the pispark data sql data source module.
So now the example that I executed is the batch read and I'll show you how I we were able to partition the data set while reading. So let me go to the implementation of batch read. So here is how the batch reader looks like. Two key methods partitions and read and partitions is the method that has enabled us to distribute the data while reading. So in this example I'm assuming user will be providing a list of the files from the source and for each file in the list I'm creating one input partition and what's an input partition it's an another class that comes from the same module we implemented it and we called it Excel input partition in this case.
So now in this example because Excel is not a splitable format I'm creating one partition per file but imagine you have some other data sources source for example CSV if you can further split the file you can implement that logic over here can and can return one partition per split so that you can enable you can process every split in parallel and then the read method which runs per partition and in this case I'm using pandas to read the sheet in the excel and as we already mentioned uh if we do not implement the partitions method then all the data set will be read in one single partition so you probably have to uh accommodate that logic in your read method all right so this is how the batch reader looks like now it uh I want to write my data frame into excel file as well so imagine you got some data frame like is and you have done some computation on top of it.
So I can use the dataf frame. right method and here I'm using my Excel data source to write it to a location that I have provided and you can use uh you can you can fi look at the data source and you can you can get all these options there. And if you see here this one thing I would like to mention because I had five files in my source I can see the five tasks were executed on that. So it is processing my data in parallel.
Now how it has done that we can go to the data source writer implementation and see it. So let me scroll down and go to the data source writer. So this is how data source writer looks like. one key method write this method runs once per partition. So in my case I had five partitions and this has executed five times in parallel on executors and I got five files uh and I'm naming the file as with a random file name. So if I go to my uh directory where I'm writing the data, I can see five files are created and then there are a couple of other key methods like commit and abort.
So commit runs when all the tasks in your uh in your processing have been succeeded and a bot runs if one or more task fails. So in our case all the me uh tasks were successful. And then there's one thing I would like to mention. You also get a list of writer commit message in commit and a bot. And what is writer commit message? It's an object that you can return with your write me write method. So if you want to see the implementation of it, this is how we have implemented it and we have called it Excel commit message.
So now you can imagine that once your executors have executed their task you can send a message or or values back to the driver and just to call out commit and about runs on driver. So after you uh your processing is complete you can use commit and abort to take decisions on your processing. So in this case since all the tasks were executed if I go to the cluster I can see the messages are printed from the commit me method right.
So this is how the data source reader and data source writer looks like for batch processing. Now let's switch to streaming a bit right. So stream processing as we know is used for uh incremental uh processing or continuous processing. So there is a there's a difference in batch processing and stream processing and that is checkpointing. So when you run the stream for the very first time, it processes all the data available.
It checkpoints that location and then when the next trigger runs, it picks the data from there and then processes what was ingested since the last run. To enable that, there are couple of methods in the stream reader and I'll go to that now. So if you see here stream reader implementation looks a bit similar to the batch reader. We have partitions method for partitioning the data and we have read method to read it part each partition.
But there are a couple of extra methods. So initial offset and latest offset. So when the stream will run for the very first time the value that will be returned by the initial offset it will start from that and start processing data from that point and it will process the data to the point of the value returns by the latest offset method and then from there onwards the latest offset value will be checkpointed and every following run will start from the checkpointed offset and get get the new latest offset from this method and run up till that point.
So in this example I'm from the initial offset I'm just returning a time stamp which is like 1st of January 1985 and latest offset I'm returning the current time stamp. So essentially I'm processing all the files which are ingested in my lifetime. Uh so and then if I go to the uh stream uh read example here I here you can see this is a familiar read stream method. We are using the format excel and then I'm passing some set of options.
I'm passing the spark partition I'm using the spark partition ID method again to see in which partitions the data is read and in this case I'm reading the stream in memory. So I can query this and see what all data is read. And you can see it's similar to the batch. You have read all the data in different partitions. And in the real world it's uh when you run the stream processing you will be reading data uh from a source and for example you are writing data to the delta table.
So you can use a delta format here that will allow you that will allow you to keep processing your stream data and keep appending data to the Excel table. So, so sorry to the delta table and I would like to show you how the checkpoints look like for a moment. So, here I'll go I'll click on my offsets. If I download this file and open it, I I'll zoom in a bit on this as well. Yeah. So, here I can see that the uh the checkpoint I'm returning from my latest offset is already checkpointed and from there the stream will continue.
All right. So this is uh how the stream reader looks like and it will be very rare in uh in a case where you will be writing Excel files from the stream. But I would still like to show you how the data stream writer looks like. So here is the example of data stream writer and if you see it looks quite similar to the data source writer which is the batch writer. You have a write method that runs every partition and then you have commit and about methods.
The only difference is you get a a parameter for the batch of the trigger. So then you can take decisions per batch. So that completes our demo. We have seen the examples of batch reader, batch writer, stream reader and stream writer. And we have seen how easy it is to implement a data source for the data that is not supported out of box in spark. So now let's switch back to the slides. Perfect. Now we have seen it working and we already mentioned that it it it is recently built and Scala data source API was already available.

31:15 Performance and considerations

So one obvious question is how it compares to the Scala data source API in terms of performance. This API is built on top of Scala data source API and it uses Apache arrow to for the data exchange. So it's relatively slower. And if you look at the graph at the top right, it talks about the constant overhead between these two APIs. So when I say constant overhead, it's basically a time to process one row in Scala versus Python without doing any external computation.
So I'm not calculating any time reading from the source, writing to the sync, any computation on the data frame. This is just onetoone mapping between processing a row in Python versus Scala. And it includes all the time for the all the overhead for the communication between Python and Java uh Python and JVM also the graph uh but we do not have to process the data row by row right so if you look at the second graph uh you can all return the data in arrow record batches and it is referring to the Python colummer current row in this graph and you can see the data has it has already improved Right?
If you compare it to Python, row by row, there is a significant improvement. Also, there is some work being done on this which will enhance the performance further. And just to note, this is in order of e to the minus 7 which so we are talking about tens of milliseconds. So unless your data processing is very very sensitive to the SLS, Python data source will work for you. Which brings me to this slide. what are some highle considerations that I should have while building this API.
So we need to define the use case as we discussed whether we are reading from it we are writing or we are writing using this API we are doing batch processing stream processing or both and based on that we have to implement the respective methods we also have to decide the partition strategy how to partition and when to partition. So for example, most of the batch use cases partitioning makes sense. But let's say you're doing a continuous processing and the data between two triggers is not significant then you can probably skip it.
We already touched upon the performance SLA in the previous slide and imagine you have an implementation in Python which is processing some complex data set with this API. Now you can write you reuse that code instead of rewriting it in Scala or Java. So you can you can decide based on that as well. And if you are wondering when we can use this, it's available in Spark 4 which is recently released. But in the databicks environment, we do not have to wait for DBR to support Spark 4.

34:05 Closing and resources

It's already available 15. 3 onwards. And the DBR I have used to run my demo is 15. 4. Also, it is available in serverless compute from environment v2. So, you can use it even now. And that's all we have to share with you today. Here is some set of documentation that you would like to read if you want to know more about it. And I would like to personally recommend looking at the community developed data sources. There are some really good examples.
So, if you want to start, you can go here, go there and look at it. And this is uh the QR code for the demo source code that we have shared. We'll be do making some improvements to it but you can have a look. And finally, I would like to uh end with saying this is an API that has given freedom to Python developers to build their own data source. So if you're struggling with a data set that is not supported out of box in spark instead of doing the workarounds try this API and it will do wonders for you.
So finally thank you for your time. I hope you enjoyed the session and enjoy the summit as well and if you want to give any review for us you can you'll probably get get a notification for it. Thank you so much.

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.