Skip to main content

Simplify Data Ingest and Egress with the New Python Data Source API

Summary

  • Custom data source simplification: The new Python Data Source API for Apache Spark makes it easy to build reusable components for non-standard data ingest and egress on the Databricks Data and AI platform.
  • Customer-driven innovation: Built directly from customer feedback across industries—IoT devices, fixed-width files, proprietary systems—the API addresses real engineering pain points.
  • Production-ready patterns: Define custom data sources once, reuse across teams, and leverage both batch and streaming patterns on the Databricks Data and AI platform.

Simplify Data Ingest and Egress with the New Python Data Source API

Watch: Simplify Data Ingest and Egress with the New Python Data Source API
Talk by Craig Lukasik, Sr. SSA, Databricks
Data engineering teams are frequently tasked with building bespoke ingest and/or egress solutions for myriad custom, proprietary, or industry-specific data sources or sinks. Many teams find this work cumbersome and time-consuming. Recognizing these challenges, Databricks interviewed numerous companies across different industries to better understand their diverse data integration needs. This comprehensive feedback led us to develop the Python Data Source API for Apache Spark™.
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

What problem does the Python Data Source API solve?

It simplifies building custom connectors for non-standard data sources (IoT, fixed-width, proprietary systems)—reducing engineering effort and enabling code reuse.

Can we use it for streaming and batch?

Yes, the Python Data Source API supports both streaming ingestion and batch processing, so you can build unified connectors for diverse integration patterns.

How does Apache Spark support custom data sources in 2026?

The Python Data Source API in Apache Spark 4.0 enables developers to write reusable connectors in Python for any custom data source. This emerged from extensive customer feedback about integration challenges with non-standard systems. The API provides standardized patterns for partitioning, offset management, and commit semantics, making streaming and batch ingestion consistent on the Databricks Data and AI platform.

When should I use Kafka, Kinesis, or direct lakehouse ingestion?

Use Kafka or Kinesis for intermediate message queuing when you need durability and topic-based partitioning. Use direct lakehouse ingestion via the Python Data Source API when your source system has its own natural partitioning and you want to avoid extra hops. The choice depends on your architecture patterns and whether you benefit from message queue semantics on the Databricks Data and AI platform.

How does Apache Spark compare to Apache Flink for streaming?

Apache Spark processes data in microbatches with strong fault tolerance through write-ahead logs and exactly-once semantics. Flink operates on individual records with different latency and consistency tradeoffs. Spark's microbatch model on the Databricks Data and AI platform integrates seamlessly with batch processing and provides excellent support for late-arriving data through watermarks and custom partition handling.

Full transcript

0:00 Introduction to custom APIs

All right. So, uh, amazing conference, flying taxis, race cars. Pretty cool. So, I hope you're having a great time and networking and getting uh, you know, super super excited about data and AI. If you haven't seen like the Genie space demo, definitely worth worth looking into. Um, now here's an oblig obligatory slide. I don't think I have any like forward looking statements, but if I do, please know they are subject to change.
And um, let me just state the obvious here. So, data bricks is a little obsessed with data. So, you know, if you at the end of this can rate me, um, that would be great. If you if you like the content, give me a high grade. And if you like the support staff who help with the display and the audio, give me a high grade. So, thank you very much. Uh, so I'm based out of the area. I'm a specialist solution architect at Data Bricks.
And today I'm excited to tell you about the uh the new Pispark custom data source API and I want to tell you how it helps the lives of not just the data engineers but also the architects and the managers. Um this new feature hopefully can help you build reusable components for those non-standard data sources. We were we were talking about that a little earlier. And um before we go into the new feature, I just want to tell you about the origin story.

1:20 Origin and customer feedback

So every Apache Spark feature originates from the Spark community and many of whom are data bricks engineers. Um many uh Spark improvements originate from what we at data bricks hear from our customers. So we hear things like hey we got to ingest this streaming data or this batch data these fixed width files. Um, and it's it's not a standard source source. It's not parquet or CSV or Kafka or Kinesis. It's it's something else.
We have IoT devices, real estate devices, energy devices, and um, so getting the data and likewise, you know, there's the same story on the other side of the pipeline, you know, getting the data out to custom syncs, that was another pain point. So, we listened and our PMs and engineers, they're pretty methodical about listening to our customers. Our mission is to democratize data and AI and um we like to hear so much about uh what your pain points or frictions are that we even have, you know, uh instructions on our uh documentation on how to give feedback and add to our road map.
So like many features this Python data source origin story started with that customer feedback and it eventually took shape in Apache Spark 4. 0 although this is a old slide um as you can see because I think it's no longer preview. Um before we go into this pispark custom data source uh I want to revisit some concepts. Uh most of you probably know the difference between batch and streaming. So I'm not really going to spend time on this slide but here are some other concepts that are sort of foundational when it comes to structured streaming.

3:21 Structured streaming concepts

Uh so I'm going to go through these with you. So first of first off we have these this concept of partitions. All right the word partition is a little bit overloaded in in Apache Spark but for streaming uh for our uh demo uh think of it as the partitions in the source system um and it's kind of just like Kafka or Kinesis. So so think of this as a way to divide labor a division of labor. A data stream is sliced into smaller manageable segments and Spark can process these segments in parallel.
This is key to achieving high throughput and it's an important concept to understand when using this new API especially if you're doing the streaming or you have high volumes of data. Um so different systems have different names for uh these source partitions. Kafka has partitions, Kinesis has shards, you know, tomato tomato, right? It's all it's all the same. Um, but the mapping from the source systems partitions to Spark's input partitions isn't always one one.
So just to complicate it a little bit more, you got the source system partitions and you got Spark input partitions. Um so if you use Kafka to ingest uh into data bricks there's an option called min partitions that will actually have a single spark input partition reading from multiple Kafka partitions and that just can give you a little finer control over your job's parallelism. Um in the demo we'll show later it'll just be a one to one nothing too fancy.
So that's our uh first concept. The next concept is offsets. Uh, and this slide calls it a to-do list. So for each microbatch, Spark is quering the source system. Or you can really in this custom PIS spark data source. You could have custom logic that figures out the range of data that it will process from each partition from each source partition. So for instance it might determine for this batch I'll process records 101 through 200 from partition number one.
So this range is defined by a starting and ending offset and this to-do list is created before the batch processing typically begins and is stored in a right write ahead log. So if a single task fails mid batch, Spark can use this to retry the exact same range of data, ensuring no data is lost or recalculated. And again, different systems have different names for it. Kafka has offsets, Kinesis has sequence numbers, potato, right?
It's it's all the same. And then finally, there's the concept of commits. Uh this is the done log. You can think of it as a done log. Uh after a microbatch successfully processes its data and writes to its destination or the sync, Spark records the completed batch number in a commit log. And this log simply notes that the to-do list for that batch is finished. And its real job is job level fault tolerance. If your entire streaming application crashes or restarts or using like the available now trigger um it it'll first check that commit log when it restarts.
So if it sees that batch 123 was successfully completed it will start the new run from batch 124. And when I talk about batches this is spark operates in this microbatch mode and the microbatch mode can have you know a number of records. So some other systems are more record by record. Apache Spark I think a lot of people here are familiar with it um is microbatch based and uh so this is this mechanism really helps prevent that data duplication and it's a critical to achieving endtoend uh exactly once or at least once processing guarantees and it really depends on your sync logic um and this also enables other actions so with this custom pispark API uh data source API um you'll have an opportunity to write a method or implement a method called commit.
And in that commit method, you might do something say if you're going to a uh a wellhead to get some API data and that wellhead has you know maybe a little bit of onboard historical memory in that commit method. You can go ahead and maybe hit the API that clears that historical memory. So, those are the concepts. All right. Now, let's uh you know, before we go into a demo and look at code and whatnot, let's uh let's imagine a scenario.

8:43 Startup scenario and reuse

All right. and the startup builds some AI based tools for developers, right? So, it's super cool, super cutting edge, super excited. You're at a startup. Startups are fun. How many people here raise your hand if you're at a startup? Uh, awesome, isn't it? Yeah. But, but the reality So, yeah. So on day one though, usually your boss gives you something that needs just sort of data engineering basics, right? So your your goofy boss says that you need to help with a bunch of teams uh that need to pull in data from GitHub.
So your boss says that you need help to help out these various teams across the company. and uh they need the pull request data from GitHub. And if if you don't know what a pull request is, I think everybody here probably does, but think of it as a change request. So when a developer wants to update a codebase, they submit a pull request or a PR and that will be reviewed and merged into the uh main codebase. Hopefully, it's reviewed and approved.
So remember, we work at a startup and we're selling these services to developers and these PRs will be used by the startup's fancy AI engine. So it might uh for instance send an automated message to a developer with a teaser analysis that looks at their PR and tells them what what they think about their code submission and what what they can change. So, you're about to tell your boss, you know, I don't have to do anything here.
So, GitHub has a I think I have a laser pointer. GitHub has an API. You almost tell your B boss, yeah, that's easy. Those teams can just use that REST API and each pull request can be retrieved. So, you can see the API there. That's the uh identifier for the pull request. And so think of it as just a sequence number or maybe an offset. So before you respond to your boss with that sort of gut reaction, you remember some of the training that you had earlier that day.
Now the founders of these startups, they're goofy folks, right? So they grew up in the 80s and 90s and they have some very clear cultural principles that were part of your onboarding training. So you remember the training video that was part of your onboarding and an image comes to mind, right? So the company values reuse. Anyone grew up in the 80s and 90s? No. A couple. All right. Pretty I'm pretty ancient. But anyway, that was a big big thing.
Hopefully you still reduce reuse and reduce and recycle. But anyway, so the company values reuse. Um, it wouldn't go over well if the teams that need that PR data had to wrestle with the G GitHub API and roll their own solution. So you want a reusable solution. Likewise, think about the effort each team would need to make their own implementation for their PR retriever. So that's development, testing, versioning, etc.
So you want to reduce that effort across the teams and your immediate goal is going to involve helping the team that needs streaming PR data with an object-oriented approach. You would allow others to extend your implementation and perhaps add support for batch workloads. So to recap in sort of more modern terms, we want to aim to build composable solutions that can be independently developed and tested and reuse across the company.
So just to recap, so some teams need batch data, some teams need maybe some streaming support, and this is a perfect sort of scenario where this custom PISpark uh data source API can come in handy. And this is what it looks like. It's it's really sort of simple at the high level. Um but it can be a little complex at the low level which is why we went through those concepts earlier of offsets and commit and uh partitions but at the high level from a programmer's perspective um the API is simple you have a top level data source class that's in the step one there uh and then you have one or more supporting classes and those classes is really depend on what you need.

13:02 API implementation overview

So if you need to support batch, there's a method that you would implement that returns a certain type. If you need to re uh implement streaming ingest and then flip side for the syncing. The second step is you register your class and that's it right there. Spark. datasource. register register and then I think the beauty of this is like once it's registered this source just acts like any other like data source that we we're used to in Apache Spark.
So this spark read uh oops spark read everybody knows that or I think everybody knows that more or less. If not, you should know that if you're using pi spark um or spark readad stream or the dataf frame write or dataf frame write stream and this format here is the name that's returned from a method of your class. So once you on ingest once you have that data frame it's just like any other data frame you work with you write it through transformations um have all sorts of fun with it.
All right. So to repeat, depending on your requirements, your custom format may support streaming and or batch and it may support reading from a source andor writing to a sync. Now in the code for this demo, we will be just looking only at the stream reader implementation. And this means that our custom format will only be usable with spark readstream. and how you use the API, it's going to depend on your use case. Um, so if you remember only one thing from this talk, uh, Whoops.
I think I got ahead of myself. How do I go backwards? Oh, yeah, it's this one. Yeah. So, if you remember only one thing from the talk, it's really this here. And that's uh the URL is not there, but that is the data brick stock. And so the data source is the base class. So that's really that top level class that you really have to imple implement in all scenarios. The name is the name that will be used in that spark read.
You know my special format or my real estate API data. Um the schema where is this little There we go. So the schema and the schema can be either in like the DDL format schema or the uh like the strruct type uh schema. So those are the really key elements and then these are the parts that are variable depending on your use case. Now there are sort of two stream readers. We're going to do the top one in in the code that we look at.
So that's the stream reader that is partition aware. But if you have like a simple data source, maybe not a ton of volume, there's another class that sort of just simplifies it for you that if you you know essentially would be pulling it down in a single Spark task. So if we go back to our concepts, um let's sort of relate that to Spark a little bit. So Apache Spark, you've got a query or some sort of action. That action gets broken down into one or more Spark jobs.
Those Spark jobs get broken down into one or more stages and then each stage is composed of one or more tasks. So, in the simplest stream reader, oops. In the simplest stream reader, um, you'd probably just get or not probably, you'd get one task to pull in the data. But if you have a lot of data, a lot of data volume, a lot of source partitions, then that stream reader, the top one will allow you to then when you look in the Spark UI, everyone here good at navigating the Spark UI.
Yeah, the Spark UI. It took me a while to figure out and there's a great um doc a data bricks doc I think maybe within the last year or two somebody put it up there I think it's like debugging and optimizing cost with the spark UI that is my best friend because it's tricky to navigate the spark UI but once you get those concepts and you can map it to this concept of job stages and tasks so the stages have one or more tasks um and then when we map in these concepts that we talked about earlier about the the partitions within our source then we can see when we look at the cluster that you know we if we're using the stream reader um we'll see multiple tasks pulling in data in parallel.

19:41 Demo and code walkthrough

So now I'm going to try to bring some coat up and I might wrestle with this for a moment. So bear with me. Take a stretch. And let's see here. All right. So, let's see here. Is this running still? All right. Good. So, for this um demo, uh I'm actually not going to access the uh GitHub API. It does have a rate limit. You can be crafty and implement logic in your custom PISpark data source to deal with that rate limit, but then it would make for a pretty uh sort of gnarly demo code.
So, I've got a little simulator that's just faking poll request data. And I was just making sure that's running. Um, and then here's our trusty data bricks workspace. Oh, sorry about that. I'm just going to give us a little bit of real estate here and we will look at the code Should I zoom in? All right, let me zoom in a little bit. Good enough. Good enough. All right. So, we don't have to read all this stuff, but basically it kind of explains our scenario.
We're at this AI startup and we need something that's reusable and sort of brings elegance to the end user. So, with that Spark data frame, I think it's elegant. Um, so we'll start with like the imports. So, these are really the three sort of key parts that we'll be demoing. Um, I'm not importing the writer. Um, in our demo, we're just going to look at the streaming source. So, the data source is our top level class.
And the name for our custom source is going to be GitHub. So I mentioned the sort of the key methods earlier on the on the slide and with that GitHub then we will be able to use this within our notebook further down I mentioned the schema is an essential part. So uh most of these sources have a schema. So I think if it's if you're using Kafka as your source it's got like a key and a value and I think a couple other color up other uh columns.
So for us, we're going to return the GitHub repository, the ID of the pull request, the title, author, and then some dates. And it's all going to be fake data, so it'll be goofy. Um, and then this is the end of what we're going to implement for this top level class. So our implementation of data source, we're going to call it GitHub data source. And it's really these three methods. So this stream reader is uh what we're going to implement first well in this demo.
And it's just saying this custom data source is only going to support streaming reads. So you'll get an exception if you try to do like a spark read. GitHub. I I don't know what the exact exception would be but it uh it won't work because I didn't implement the other method. So we mentioned that this will be a partition aware example. So there's that other method that we could have implemented called simple stream reader where if you have a low volume uh source then you can kind of take the easy button that way.
Um, so I'm going to do a quick question. So a number of people wrote uh raised their hand when I asked if they had a custom data source. So I am curious for your custom source that you're hitting the API or if you have to use some SDK, does it support the concept of partitions and does it support the concept of um offsets or sequence number or maybe even time stamp? Is that I forget who raised their hands, but but anyone?
No. No. Okay. So, that's an interesting pickle. That's an interesting pickle. And do you want to do you know why it's a pickle? So, if you think about Spark and if you're familiar with like the Kafka source, you can give it a starting offset. So if you're like ah I need to you know my job is messed up. I need to start back over and rehydrate the sink you can tell it like and again if you're using Kafka you can say my starting offset is you know 1 2 3 4 5.
So if your custom source does not support that offset con concept you can still like use this data source but your use cases will have some limitations. So like you won't be able like so for instance let's just say you have a source that pulls in current temperature and that's it and you can't go back in time. Well if you need to join with historical data you're going to be out of luck. I mean you're gonna have you're not gonna be out of luck but you're going to have nullles up until you know your event time matches the event time of your weather source.
So the the nuance the API is fairly simple, but sometimes the reality of what you need to connect to can can make it uh challenging, but it you should still be able to use the API. But if your source doesn't allow you to go back in time or back sequence number, well, it's going to limit sort of the applications or use cases. And if your source doesn't have a concept of partitions, um then you'll use that simple uh stream reader and you know that one task might just have to be on a beefy core.
All right, so let me keep going here. So the next class we define is our input partition. So for the GitHub source we do have this concept of repos. So in our um implementation what we're going to do is we'll allow the user of the source to supply a list of different GitHub repos. And that's a way for us to partition the retrieval across many cores on the cluster. And then likewise that GitHub API, I don't know if you remember it, but the last part of it was like the PR number.
So that's a sequential number. So that gives us the offset concept. So for us, we can implement the partition and offset aware uh logic to its fullest. And then if we have like a stream to stream join, well, I don't know if we would need that for this use case, but then we could rewind it back in time um and so forth. And for our partition, it's pretty simple. It's just what is the repo, what is the start. All right.
So now the rest of the code, I will not try to go too fast. I'll try not to go too slow. This is the class that we need to inherit that data source stream reader. We're not not using the simple one. Uh when it gets initialized, um I'm processing the options that are passed in. Actually, this token's not needed for this implementation, but the repos, that's the list of the GitHub repos that we're going to want to stream PR data in for.
And then we're going to initialize the offsets uh to zero. Now in this implementation I have here, it's just a toy implementation. I'm not actually going to be querying the source system to determine what the uh available offsets are. I've just got some logic that increments it by three. kind of it assumes there's infinite amount of PRs which frankly there probably are close to infinite um and then that initial offset is a method that we override and then we already have set those initial offsets and then this is the logic that in this toy example um this is where in your in a real implementation you would go out and maybe see what the range of available offsets are and then calculate what you want to process in that microbatch.
Um, in this implementation, we're just saying, oh, we're we're going to add three. This batch size is just a hard-coded value of three. So that way, this demo can go forever. And oh, I forgot to do one thing. Let me just run these cells and pray to the demo gods. Uh, all right. I got to backtrack a little bit. All right. So then there's a a method to implement called partitions. And we saw that class that we defined earlier called repo partition.
So that is just stating like for this repo or partition here's a starting and ending offset. There's a function here that I'll just sort of fly through. So assume this is actually accessing the API and it's hitting my simulator. So it's not the real API and then returning the data in the structure that we defined in that schema. So earlier in that first class that we implemented, there was a name method that we implemented and there was a schema method that we implemented.
So this data matches that schema that we defined. And then the uh the read method is something you will implement as well. And that is using that like API that fake API and then it just is yielding the uh the output. And then this last part it doesn't have an implementation but this method allows you uh going back I think a couple people were in the energy sector. So some of those devices might have onboard historical memory.
So if you wanted to reach out back to that IoT device and say you know clear clear the memory you would do it in this commit method because by this point you know your data has been committed to the sync. So it's safe to go back and say yeah wipe that on board memory and then this registers it. And then this code here is uh well it's going to be hard to see. So I've got an a number of repos that we're going to monitor and then I will display here.
And actually I'm going to run a couple more cells real quickly and I'll come back to them. So, we got about I think 20 different repos that we're going to monitor and then we're going to display it. And then we're also going to sync this data to a table. All right. So, it's starting up. Let's see what we got here. So, in this raw data, it's kind of cool. We can start to see our work. Ah, look at this. So, in our source, we've got a custom data source and you can see these offsets.
So, when it's first starting up, that's where I've got the sort of logic that sort of hardcodes a, you know, you know, it takes the initial offset, which is initially zero, and adds three to it. But we can see for this first micro batch, we can see what those offsets are for these various different partitions. And let's see what else we got. So we got some data flowing through. And let me just close this up. And now we can see our fake data.
All right. So, this is sort of I would say the beauty right here is like this spark. stream. format and then instead of using one of these built-in formats, we've got this one that we just created. Now, I'm going to jump down a little bit and I'll I'll skip some of this, but we've got logic that is al is basically doing the same read stream, but this time it's syncing it to a delta table. And what I'm going to show here is, let me just make sure we've got some stuff in that table.
Stuff. Um, yeah. So, we have some data in our table. So, I've got a little helper directory. Um, so if you're familiar with Spark streaming when you sync to a a sync, be it a table or a Kafka or whatever, you provide a checkpoint directory. That checkpoint directory includes these like subdirectories. So there's a commits, there's an offset. If it's a stateful stream, you'll see another directory for state. And then if we look at the example offset files, we can see uh I ran this earlier, but for the microbatch 18, we can see what the offsets were for our partitions.
And then I had run this later. And we can see for the microbatch 25 the offsets had changed. So that is the custom PIS spark API but in a way like I sort of we've sort of failed meeting the cultural goals of our startup to reduce reuse and recycle. So does anybody have an idea what we should do next? All right, perfect. So, publish the data source. So, in this example, I I put it together in a notebook. It's sort of easy to see and we can have the display and fun stuff like that.
But to really meet our mission of read, what is it? Reuse, reduce, and recycle. you know, we can put that Python file in a volume or a repo so that others can use it and pip install it or or import it. Um, and then actually I'm going to go back to my slides. I've got four more slides if I can find them. And I want to get back to Let's see if this works. This might not work. I clicked the wrong button. So, I've just got a couple final thoughts.

36:25 Conclusion and best practices

There we go. All right. So, for the developer, so the micro perspective, so decisions you need to make, read or write, stream or batch. Um, does your source support offset awareness? If not, well then that's going to limit sort of the ways you can use the data. Is the data volume high? Then you're going to want to parallelize that ingest using like the class that we we used. Um and if not, you can use that simple one for the macro perspective.
If you're a manager or architect and you're thinking of how you know how do we get this reused across the enterprise and may maybe it doesn't need to be reused but uh in some industries you might want to roll this once. So you want to think like how many other teams need this and how can we test this as an independent component put it at its own CI/CD pipeline and version it and then how do you manage changes so at the macro level you know and it really depends on the company and situation but if there you know I work with some clients where there's you know 10 different data engineering teams and if you all need to hit the same data source if you approach this uh you know with some thought and planning you can just basically roll that once let the rest of the teams use it and then you got independent versioning and uh make your your startup happy.
Um please remember to uh leave a little rating for me and uh I'll be sticking around. I have to be somewhere at 7, but other than that, I'll be here to chat. And 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.