Spark DSV2: Growing Up Fast

Summary

Spark DataSource V2 (DSV2) provides a single extensible API for Spark to read from and write to any data source — from Delta and Iceberg to JDBC, Kafka, and hundreds of connectors — replacing the fragmented original data source model.
Recent DSV2 improvements include safer schema evolution in MERGE INTO, enhanced partition filtering via Catalyst expression evaluator, a procedure catalog for richer table management, and DML observability through operation metrics surfaced in table history.
This video outlines the DSV2 roadmap including multi-statement transactions, deeper Change Data Feed integration, blob storage support, and first-class SQL features such as table constraints, complex default values, and generated columns.

Spark DSV2: Growing Up Fast

Spark’s DataSource V2 integration has taken a major step forward, beginning with the addition of a procedure catalog and row identifier support to enable richer table management and row-level operations across modern data sources. Building on this foundation, recent updates improve MERGE INTO with safer schema evolution and enhance partition filtering for more efficient query planning and execution. DML summaries now provide clearer visibility into write behavior, while key cache fixes resolve long-standing correctness issues in Spark execution. DataSource V2 has also been extended with first-class SQL features such as table constraints, complex default values, and generated columns, laying the groundwork for more advanced table semantics. We’ll also look ahead to deeper Change Data Feed (CDF) integration in Spark to support robust incremental processing, highlighting what’s available today and what’s coming next as Spark continues to close the gap with traditional warehouse systems.
Talk By: Anton Okolnychyi, Senior Staff Software Engineer, Databricks ; Szehon Ho, Software Engineer, Databricks ;

Chapters

FAQs

What is Spark DataSource V2 and why was it created?

DataSource V2 is a modern API in Apache Spark providing a clean, extensible interface for integrating any data source — from table formats like Delta and Iceberg to event systems like Kafka and databases via JDBC. It was created because the original Spark data source API could not keep up with the explosion of new formats and catalog systems over the past decade.

What improvements were made to MERGE INTO in Spark DSV2?

Recent DSV2 updates bring safer schema evolution to MERGE INTO so that schema changes in source data are handled correctly without risking data corruption. DML observability improvements also surface operation metrics — such as rows inserted, updated, and deleted — directly into table history, making MERGE INTO behavior more transparent.

What is the Transaction API in Spark DSV2?

The Transaction API extends DSV2 with ACID isolation support for multi-table DML operations, enabling coordinated writes across multiple tables within a single atomic transaction. This is foundational to supporting multi-statement transactions, which are listed on the DSV2 roadmap discussed in this video.

What SQL features are being added to Spark DSV2?

DSV2 is being extended with first-class SQL features including table constraints, complex default values, and generated columns, bringing Spark closer to the semantic richness of traditional warehouse systems. These additions lay the groundwork for more advanced table semantics and are part of the roadmap covered in this video.

Full transcript

[00:08] Hello everyone. Um, thanks so much for attending um, this talk uh, the last day of DICE and um, yeah I so I think uh, my title DSV2 I probably should have explained DS. DS stands for data source. Uh, hopefully everyone um, is okay uh,
[00:20] coming to this talk. So uh, my name is Sihon. I'm a software engineer at data bricks. Um, I work on the spark team but my previous background is Apache iceberg. So uh I'm iceberg PMC and spark committer. Uh unfortunately my
[00:33] co-speaker Anton he's very hard at work so he could not make it today but I would like to talk about a project we uh me Anton and actually a lot of data bricks people are contributing to now in open source called spark dsv2. Um so
[00:46] this topic is actually cross open source projects not just spark but we work on iceberg and delta lake and the goal is to make kind of a very nice user experience across the stack. So um yeah again thanks for everyone for for
[00:58] coming. Um so first I'll explain okay like a little background on this because it is a bit uh deep. So let me spend a little time to kind of set the picture. So Apache Spark I think everyone knows uh it's been around like 10 years. It's
[01:11] very core of data bricks. Uh it's very powerful both in features optimizations and like just massive distributed processing. It's probably bestin-class for open source uh processing engine. uh the problem that we face or spark faced
[01:25] is that it's like 10 years and the data has kind of moved under it. So in the early days if you remember Apache Spark uh what I call the era one of data source is when all the metadata is in hive metas store and then you have the
[01:38] data files in park or AVO or RC right um so everything was kind of stable but then era 2 came about five or maybe seven years ago with the introduction of Apache iceberg and delta lake and a lot of innovation in this kind of like data
[01:50] source space so this kind of like outgrow the uh what the data source uh needs were for spark and so they we invented a new API in spark called dsv V2 which we hope will be the future of Spark make it very extensible and along
[02:03] the way there's actually a lot of other ecosystem connectors too that Spark need to start to read and write from JDBC uh CFKA or like some Google Drive you know like there's hundreds now and so what we really want is um one API to kind of
[02:17] like Spark can plug into any data source so the way I imagine it is like um you know Spark has spark connect so users can have a stable API to interface with Spark so we're trying to do the same on the bottom was like uh Spark has a very
[02:30] stable API for any data source to be plugged into. So that's the goal of this project. Um some of the design uh uh oops sorry yeah some of the um oh I
[02:42] think I was on the wrong side. Sorry about that. Okay. Some of the design goals of of the DSV2 project is um one we want to make it very simple to implement. Um like we want a connector
[02:54] to be able to get up uh really early. So uh a little bit on the details of the API basically there's a interface that uh for every spark uh top level entity like catalog table so and then some
[03:06] operational entity like writing or scan which is reading and then the connector needs to provide only basic things like the metadata configs and obviously I think I missed it here but like how to read and write the data. So that's all that the connector needs to do. We want
[03:19] to move all the heavy lifting to spark side like anything to do with query planning, analysis, parsing, execution, all that should not be uh the connectors problem and should be in spark. So the reason I say that is actually um Apache
[03:32] iceberg and delta lake as communities before to work around the lack of a proper API in Spark. They actually implemented their own kind of hacky way to have their own procedures their own extension points which was injecting
[03:44] rules to Spark to support their own grammar to support their own um execution which is very fragile right well one it's very fragile because you know it doesn't survive usually a Spark version upgrade that those things change a lot there's a lot of duplicated logic
[03:57] and as a like data source community you're more should be focused on like how to encode data better right not really how to take care of execution or even like user experience how to come up with a good SQL, right? So, the benefits
[04:09] of having a centralized architecture in Spark DSV2 is that uh one for the user, we get a very uh consistent uh syntax. So, you can swap out like iceberg if you don't like iceberg or delta lake if you don't like delta lake. Um and you get
[04:22] the same kind of SQL commands and behaviors across all the data source. For the data source, we want to make it again I said it before easy to implement. And then for Apache Spark as a project, uh we're very invested in Spark. So we want it to be future proof
[04:34] and able to support all this innovation in the data storage space right there's a lot of innovation and new proposals being going on and uh we're able to now offer um functionality via DSV2 to the user like new functionality that was
[04:47] never before in Spark. So one um uh one example I always like to say for DSV2 why DSV2 is the DML example. Um if you know like spark for like 10 years never had any DML support but now that there's
[05:01] table formats like iceberg and delta lake they support rowle transactions spark was able to it was really cool thing where we made a DSP2 API called role level operation and then only some connectors would implement it right
[05:13] because these are like you need to support massive transactions across millions of rows and files and so Delta Lake and iceberg implemented it and we're able to offer unified SQL interface which is merge into update and delete from And I think these are now
[05:26] among the most popular spark uh uh SQL statements to um so this is a kind of a successful example of spark dsv2. Um and then the current state of the project is it actually has been a long
[05:39] time this project since I think since the you know iceberg delta lake and it came around spark 3. So the project started at that time but recent year uh two years I think it got a lot of momentum especially data bricks we contributed a lot to it and so this talk
[05:52] is only going to focus on the new stuff like spark 41 and 42 but just know that there was a lot of stuff uh before and uh if you want to know the perspective from other like the data source point of view I gave a talk in iceberg summits uh
[06:04] about that experience of spark and iceberg dsv2 and then today um our good friends from Delta Lake Rahul and ti are going to give a talk called new foundations of delta lake with kernel and spark dsv2 to how they're rearchitecting Delta to be more in line
[06:18] with DSV2 and make it more um one better user experience and more generic and uh better integration overall. So uh the first feature I'll talk about so I'll talk about a series of features we added in spark for one is about uh
[06:31] strengthening DML like this is again the kind of the poster child of DSV2 um so we strengthen the first is by obs increasing observability for the um open source spark DML operations um to
[06:44] explain a little bit maybe I will talk like maybe one minute about or like 30 seconds about what merge into does. So merge into is a very very popular statement because it does a lot of things at once. The idea is you have a
[06:56] target table that you want to do a lot of row level operations on and you do it by moving a source table or uh uh comparing it with the source table of like changes and the idea is you get three sets of rows in the comparison.
[07:08] Rows that are in the source only, rows that are in the target only and then rows that are in both. And for the rows that are in the target you have you can conditionally update or delete them and if they're not in the target you can insert them conditionally. So a lot of
[07:20] things you can do in one statement. Um uh the the idea here is like because it's one statement and we find it like kind of hard to uh be able to know what's going on because so many things happen at once. So we want to add
[07:32] something called operation metrics which is just tracking on the spark side how many rows uh went to which fates right were they inserted updated deleted and because of what condition. So again this
[07:44] is like the architecture of DSV2. We did all the heavy lifting on the spark side um like all instrumented the the operators to capture the row counts and then you may be asking what does have to do with DSV2. So I think uh the the the
[07:57] idea is like these metrics on the spark they will go to ether right like they are going in the spark UI but then they'll be kind of ephemeral after the job is done. We actually want um this to be persisted in the table history. Uh so you know like both delta and iceberg
[08:10] have this concept of like a table history either snapshots table and iceberg or describe history in delta. So now we add a small dsv2 extension to the existing right API to pass this uh information to the connector to the data
[08:23] source so they can persist it. So now uh both will be using this to populate the history in the same consistent manner. Um so that's one example of a small improvement to the API to for a small um
[08:35] uh uh operational feature and then um uh from Netherlands uh one of our good colleagues Zia he uh implemented update and delete as well. So the other two uh DML operations are updates and delete.
[08:48] Uh update you know you take a condition and you just change the rows that match the condition and delete you delete the rows that match the condition. So same thing uh spark instruments uh both the update and delete operators now and pass
[09:00] all this information to the DSV2 to um consume. I think later we may even unify the syntax for how to get the table history but I think that's like maybe for another day. Um and now I'll talk about uh so that was
[09:13] like small example. I'll talk about a little bigger feature we did in Spark 41 and 42 which is scheme evolution um in open source spark. So DML a lot of times um Yeah, so a lot of time, oop, sorry, I
[09:27] think it's uh, yeah, uh, a lot of times what happens is, uh, you have a very stable statement, but then the source schema gets out of sync with your target schema. So, uh, I think it's a very heavily requested feature that we
[09:39] support schema evolution to make this thing work uh, even if there are changes, schema changes. And the basic idea is um, we again compare source and target, but this time instead of rows, we compare the columns. So there are
[09:51] columns that may be added from the target or sorry added from the source that are not in target uh maybe missing columns the in the target not the source or they are in both but like different types. So the basic idea is just when you bring the source row into target uh
[10:04] table you want to just be more forgiving. So we either add the column if it doesn't exist we widen the column if it's a different type or we make a sensible default like if it's a new row either null or the default value. If it's an update we will just not update
[10:17] we'll keep the original value of the row. Um so that's the design for schema evolution and then uh sorry just okay yeah so we added a syntax uh merge
[10:30] using schema evolution uh again like a unified syntax for every data source now that supports merge into and then with schema evolution for the data frame API. Um and then implementation again all the
[10:42] stuff is on the spark side like heavy lifting like uh doing the analyzer we compare we resolve the table and compare the two schemas. We uh one note is like we compare the schema by name. Um and
[10:54] then we also recurse into the strrus as well to do the comparison of nested fields in the strrus to do the same uh uh column changes and we call an existing DSV2 API. So here we didn't need to invent a new API. We just invoke
[11:06] the existing API called alter table which is already used by the user when they type alter table add column. Um and then we actually let the data source decide what's a valid schema change. Like sparks in no position to dictate
[11:18] like to the table format what it can widen or not. Uh like Delta Lake and iceberg for example have very different uh or not very like somewhat different uh sets of widening that they support. And uh there's actually this project
[11:32] took a lot longer than I thought. like there's many many corner cases here. One interesting case that I think uh my colleague Johan in Amsterdam um fixed was um we found out that okay this is really nice but then the data source by
[11:46] delegating to the data source we find that data source is more strict than the spark in terms of what uh column widening is support. So to explain, imagine you have a merge into and you're bringing kind of like um int column into
[11:59] a a string column. Uh spark at runtime if you don't have schema evolution can actually inject a runtime cast and that query will work. But with schema evolution it will fail because we ask the data source and it says no you
[12:12] cannot just change like into string. So uh we did in the end have to add a new DSV2 API for the data source to tell us like what type changes support. So it's a minimal change, but it still is kind of like um uh uh some uh example of
[12:26] communication that needs to happen between Spark and data source to have a seamless user experience. Um and then uh Johan also did insert into scheme evolution which is really cool in Spark 4. So I think all the DML should be
[12:38] covered now. So there's insert with schema evolution and then data frame API uh with schema evolution. Um, one kind of like side note to insert into uh I think for legacy reason or just like
[12:50] it's a little different syntax it's the default is by position and so that the the inference for the schema changes is that we do the column uh comparison by position by default. So we like call
[13:02] compare columns at 0 one two and not regarding the name but you can still get the by name uh to match like what merge does if you put the by name clause which is already I think already there. Um yeah so um and then this I'll talk about
[13:17] is uh more Anton's thing but this is also a very crucial part of DML as well it's not um enough for you know uh table format to support merge into or something what it really needs is a
[13:29] transactionally safe uh merge into right so this is actually really important for correctness um and to explain a little bit transactional what it means for transnational support um all the DML operations right like merge to uh delete
[13:43] from update. They're data manipulation. So the idea is you read the data at some time. You transform it and you write back to the to the table. So there's two properties among the acid that we really concerned about. One is atomicity. Um
[13:55] either you should like commit all your changes or don't leave or nothing. You don't want to leave in a half state. Uh an example is schema evolution. We want schema change and data change to both go in or none. You don't want like just half of it to go in. And then it should
[14:08] be also isolated uh which is ACI the the eye of the acid. And isolation level means uh you don't want to uh conflict with the concurrent rights. Um so a good example I always think about is like
[14:21] maybe you and me we're both doing the same DML. We're both reading um the table but then maybe you're faster. So you commit first your changes and then I commit my changes after you. But I'm I'm basing on uh something I read before you
[14:34] committed. So then it's actually wrong. I'll actually overwrite something you did right so that's some isolation level is something we want to avoid um and so uh one example a concrete example is
[14:46] like insert with schema evolution this would today in spark 44 2 on iceberg uh or spark 41 sorry this is not transactional like schema evolution and the data changes don't go together and then also uh we would overwrite each
[14:58] other if we both ran this insert uh statement um so the idea for how to do this in DSV2 layer um so this is done Anton and Andreas. Um, so we would need actually a kind of a higher level DSV2
[15:11] API. The DSV2 APIs currently were like, you know, read the table, write the table, but they don't have the context that this is like one big thing. So we just need to give them the context. And then uh what happens is the data source will need to do be smarter. So when we
[15:24] do a write, it should know I don't do the right right away. I need to stage it for the commit, the final signal, right? Commit or abort. And when we do the reads, the data source needs to know um what version it read from so that at the end it can do checks to see whether
[15:36] there are conflicts in the meantime. Um so okay sorry yeah so this is the API uh Anton I think designed this is called transaction API and basically we always
[15:48] like to reuse existing API. We don't want to burden the data source too much. So it's essentially just like a wrapper on catalog but with additional commit and ab aorts procedure. Um and so the
[16:01] idea is like okay so this is the endto-end example how this would work. Uh spark issues again a set of operations to the catalog like load this table load this table. So with this example um you we're reading a table joining with another table and then
[16:14] writing back right so we would load both tables and then like your iceberg or delta lake would remember oh this was version one version one and then we would spark would do writes so it would like alter schema and then do the writes
[16:26] and then the data source would uh stage both of them. It would do the the actual transform but not commit yet. And then when spark says commit the data source needs to do two things. First is to check to reload the table and check are
[16:38] there any changes between the red version I read and the current version that I will be overwriting. So that's the isolation level and then uh two it needs to atomically do the metadata update. So one uh so I the schema change
[16:50] and the moving the data pointer should happen in one uh just one operation right that's automicity. So with that then all the DML and spark becomes uh transactional. Um and the current limitations for transaction I think we
[17:02] will continue work on this is it's currently you can't do like a begin and uh commit yourself like this is just internal to spark we will wrap all the DML with the transaction. Uh we may do it later like multi-statement transaction but it's more involved
[17:15] because it's longer live transaction and then uh we don't support cross catalog. So DSP2 has a catalog concept and usually in iceberg and delta like these table formats catalog is the unit of locking. So you cannot just do a cross
[17:27] catalog transaction yet. Um cool and then I think um and now I'll talk about a new feature I think is pretty exciting is called change data capture. Uh I'm pretty excited about this. This is I think um Anton and Gangyang another
[17:41] colleague of ours uh and Johan. So the idea is here we want to again expose another yet another table uh data storage functionality to the end user and this is the fact that all these new
[17:54] table formats be it delta lake be it iceberg are versioned and um we want to be able to query changes between versions like what are all the rows that were added between version one and two or rows deleted or updated and there's
[18:07] many use case of uh change data capture right there you can replicate tables you can incrementally uh update materialized views And I think uh just uh right before my talk, Gangyang gave a talk on CDC support. So I'll only give a very
[18:20] brief or kind of a more brief overview than I think he had more details in his talk. So um you can see the recording for that. Um the funny the interesting thing is like both iceberg and delta lake again because spark didn't have CDC
[18:33] support they implemented their own uh rules to have their own like you know remember iceberg had create change uh view and then delta had its own syntax um and this led to inconsistent behavior between two uh table formats and just
[18:46] duplicated complexity. It turns out change data capture is quite an involved execution. Um and so what we did is like we now we support a centralized spark syntax and moved execution to the spark side and uh some some of the complexity
[18:59] that spark handles now is um again gung talked more about it but there's three main things like one is um a lot of these table formats they end up with like uh on copy and write mode uh they actually read and write back the same
[19:12] row uh for every version. So we want to remove all those, right? Because otherwise you have just a lot of like no no op updates that the user didn't really intend. And then update reconciliation. These table formats, they deal with more like insert and
[19:24] delete. So we need to stitch the delete and insert together to be like um an update one update row. And then net changes. If the same row gets updated like 10 times uh even by the user, we we
[19:36] don't want to show that 10 times. We only want to show the beginning and end. So these are just some of the complication that really makes sense to centralize in Spark side. Um so the API that got added is uh uh this change log
[19:48] API. So the data source will return us raw changes like okay all for all the snapshots what all were inserted and deleted because it has all that data there and then spark uh it has to just return some configs to spark like what
[20:01] is the row ID uh what is the row snapshot version of each events and then uh whether you want spark to handle all this dduplication. So this is the um the way the architecture is laid out now for
[20:13] change logs. Um and then this is now unified syntax. Uh so Spark we debated a lot and uh you may remember another uh feature of Spark called the time travel which is taking advantage of the you
[20:25] know time travel capabilities iceberg and delta lake and so we made the syntax kind of similar to that now like we have from version from time stamp um and then you but there's a two at the end as well. So um and then you can run this
[20:38] for batch and also for streaming in terms of SQL and then also uh data frame API there's also batch and streaming version of it. And so the last one I'll talk about uh sorry it's kind of a lot but the the
[20:51] last one uh something I worked on myself called enhanced partition filtering and this is the example of like we're not only adding new features in spark in data bricks we also want to uh be able to take a look at the old APIs and say
[21:03] oh this had a some gap and we want to fix it. So partition filtering u you may know is kind of like a very old thing like data source the basic idea is data sources like iceberg or delta lake or even parquet they have data uh
[21:15] statistics to skip files right so the idea is um if you have a filter uh in your query you would push that down um via the DSV2 API like in this case the scan that's the read API and then we
[21:27] evaluate the filter expression against all the partition values of the files and only the matching ones get returned right that's the uh basic uh filter pushed So what's the problem of this? Uh when we analyzed it. Oh, sorry. Just Okay.
[21:41] When we analyzed Oh, sorry. Got too. Okay. Yeah, there's a bit of lag. uh when we analyzed it uh spark native data
[21:54] sources again DSP1 where all the code lived in spark it that had a lot of benefit because it could use spark's internal library which is catalyst to evaluate every single expression uh for partition filtering against every single
[22:07] partition value but now DSV2 is more of an abstraction right we don't want to leak catalyst to the DSV2 and also these DSV2 data sources are external to spark so uh they usually support only SQL
[22:19] standard expressions Right? They don't because they have to be cross- engine. Right? Iceberg can be read by TRO by uh you know uh other uh query engines. So we only want to expose a abstract uh SQL
[22:31] standard expression to DSV2. So then a lot of uh expressions don't get uh pushed down. And then finally um the data source communities like Apache iceberg and Delta Lake, they're not in the business of evaluating expressions,
[22:44] right? They're not like their whole focus is metadata. So it's really like no one really uh contributed a lot to evaluation expression. So um to give a exa uh more concrete numbers to this um
[22:57] like spark you can always make udfs right? So there's like infinite number of udfs you can make and none of them get pushed down because they're all like non-standard uh expressions. Spark has 500 like Spark dialect expressions and
[23:09] you can use all of them in partition filters and a lot of them don't get pushed down either because uh the SQL standard only has about 100 expressions um or at least the Spark uh DSP2 layer
[23:21] supports only 100 SQL standard expressions I should say that um so about only 100 get pushed down and then iceberg it's kind of pathetic like there's only about 10 that get evaluated in the end so uh among the whole funnel
[23:33] like you lose like a ton of uh potential show partition transforms actually or partition filtering. Um, and so it we just thought a long time about the problem. We decided, okay, why don't we clean this API up and instead of passing an expression, we pass a a blackbox
[23:46] predicate to the connector. Uh, so this might be interesting if you're writing a connector someday. Um, so this this is now um will wrap a catalyst expression and just have a eval function. So you can just put in your partition value and
[23:59] return true or false. So as simple as that. Um this provides this actually keeps the abstraction really well because um now the data source actually can even get rid of these like 10 expressions and just like use this one
[24:12] entirely um to evaluate whether a partition matches the query or not. It doesn't need to understand the query, right? It just has a partition value. Um and we didn't have to add a new API. We just use existing API to push this thing down. Um yeah. So then an example of how
[24:26] this works is um you know I create a UDF uh maybe a dumb U udf like is is number one and returns true if it's number one we push it down and before this would just get rejected at the DSP2 layer
[24:38] because it can't translate uh you can't translate UDF but now the data source for example iceberg or delta lake they will have some metadata and they can call eval and only the file with the value one in the partition column will return uh to spar. So I think I finished
[24:52] um a bit early but uh we have one more slide I think trying to okay yeah I just want to say like today
[25:04] I um I mean there's already a lot of content I think but just know that like through the history of Spark DSV2 has always been like a very uh heavily focused thing um from 3 you know when we added the initial foundations for
[25:16] catalog table uh we added partitioning and then UDFs and then more pushdowns uh DML uh storage partition join and then like 41 constraints default values variance 42 we only touched now 42 so if
[25:30] you're interested in any of the other uh features I really encourage like to to um go back to the uh previous talks on this subject I think each each is kind of like a very interesting uh field on its own takeaway though I think the
[25:43] future is kind of bright for spark dsv2 hopefully I mean we all all want to invest in the spark community um in data bricks and you know we have a lot things coming up maybe multi-statement transactions in oss spark maybe materialized views uh and but more like
[25:57] is actually driven by the data source so if you follow like parquet or iceberg or delta lake there's a lot of talks now about multimodal use cases for AI like blob storage wtables the takeaway though
[26:09] is DSV2 we're we're investing it um even though it's like a the middle layer because I think it's the past present and future for integrating data source into spark and like there's always going to be new data sources and data source
[26:21] features and it's important to continue collaborating within the community across communities to make sure this is offered like uniformly to the end user and like they have a very seamless experience. Um so yeah that's kind of my
[26:33] talk. Um yeah, thanks again for attending 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.