Clean, Correct Change Data Capture with Spark Declarative Pipelines

Summary

Standard Spark streaming output modes cannot handle updates and deletes because they are built on append-only prefix-table semantics, making CDC a fundamentally different problem that requires a dedicated abstraction.
Spark Declarative Pipelines Auto CDC uses an auxiliary staging table, a reconcile step, and a merge operation to process change events correctly, including out-of-order data and delete tombstone management.
This video walks through the Auto CDC API, showing how declaring intent such as SCD Type 2 replaces complex imperative merge logic with built-in consistency, incremental processing, and resilience across batch and streaming workloads.

Clean, Correct Change Data Capture with Spark Declarative Pipelines

Learn how to construct clean, correct, cost-conscious CDC pipelines that conquer chaotic change, column creep and complicated SCD calculations. Today, implementing CDC typically means writing complex, expensive merge logic. Handling straightforward cases is manageable — but complexity escalates quickly when you introduce out-of-order data, duplicates, schema evolution and slowly changing dimensions. What works in development often breaks under real production conditions. Spark Declarative Pipelines (SDP) addresses this with Auto CDC, a declarative API that abstracts away merge orchestration and correctness guarantees. Instead of implementing imperative patterns, you declare intent — such as applying SCD Type 2 — and SDP executes it with built-in consistency, incremental processing and resilience. In this video, we’ll walk through the Auto CDC API, examine how it automates complex merge operations and show how declarative CDC improves reliability across batch and streaming workloads.
Talk By: Andreas Neumann, Senior Staff Software Engineer, Databricks ; Joseph Torres, Software Engineer, Databricks ;

Chapters

FAQs

Why can't standard Spark streaming handle CDC updates and deletes?

Spark structured streaming is built on prefix-table semantics where new rows are appended and the table only grows; updates and deletes require rewriting earlier state, which cannot be represented as a pure append operation. This fundamental mismatch is why a dedicated CDC abstraction is needed for database replication and similar workloads.

How does Spark Declarative Pipelines Auto CDC work internally?

Auto CDC maintains an auxiliary staging table, reconciles incoming change batches against it, and then merges the net changes into the target table in the correct sequence order. This execution model handles out-of-order events and delete tombstones correctly without requiring developers to write any custom merge logic.

What is the challenge of out-of-order data in CDC pipelines?

Out-of-order data means a delete event can arrive before the insert it logically follows, or an earlier version can arrive after a later one, causing incorrect state if applied naively in arrival order. Auto CDC uses sequence-based ordering to ensure changes are applied in the correct logical order regardless of when they arrive.

What future improvements are planned for Auto CDC in Spark Declarative Pipelines?

Future work described in this video includes partial upserts, which allow updating only specific columns rather than the full row, and SCD Type 2 row splitting for more complex slowly changing dimension patterns. These additions will extend the declarative CDC API to cover additional production CDC scenarios.

Full transcript

[00:08] My name is Jose Torres. Uh I'm an engineer on the declarative pipelines team and today we are going to be talking about autocd in Apache Spark. So to begin to begin I have to go into some
[00:21] deep theory here on what exactly is a stream. Now it sounds like a it sounds like a simple question but as we will discuss there are a lot of complications under the hood here. The historical
[00:33] model of of streaming within the spark structured streaming project has been based on what we call pref prefix tables. So a structured stream at any given point represents a logical prefix
[00:46] of an unbounded table. So as new rows come in the prefix gets long gets longer. uh you know they're just appended onto the onto the end of the table. Uh the prefix plus the
[00:58] transformations that the stream applies to it get materialized into the target you know the thing that the stream is writing into. Uh there are some caveats around stateful operations here where
[01:11] they kind of maintain outofband state that is you know depicted nowhere in this in this diagram but it still fits the fits the intuition and really any appendbased workstream can fit pretty
[01:23] well into the into this model. The problem we encounter though is that some use cases and many many useful ones are not really appendon. Uh one that one
[01:37] the kind of the kind of easiest to reason about is database replication. You know I had I have a remote database you know just a myql or postquest somewhere I want to maintain copies of it within
[01:50] my lakehouse. So it is true that as the stream advances uh new rows will arrive and get appended to the end of this table but there are also updates and deletes that are going to happen to the
[02:02] to this table and though those can't be represented as a purely you know add to the end kind of thing. They fundamentally have to go back and rewrite things that have already happened. Uh you also have cases where
[02:17] you just kind of want to do more ad hoc deletes. You know, perhaps there's some data that for compliance reasons you need to delete or there's that there's this information you got that you now discovered is wrong and you want to go
[02:30] correct it. Obviously, what you can do is do that as a kind of outofband pro process. But it's kind of inconvenient to manage that. You know, it's a whole separate thing. It's it's a separate code path that you only execute
[02:43] sometimes and isn't part of your core testing. It's really a lot more convenient to have it be part of your core processing logic. Uh there are some things that go even further than that where your source data doesn't really
[02:57] represent a concrete thing in the world by it by itself. You might have you might have you know the status of an order where each row you receive is a change made to that order. You know you
[03:10] might have snapshots of device output where you're reconstructing the state of a device. You might have uh accounts where you're tracking the balance of the account across various transactions that happen to it. Uh these events do kind of
[03:25] comprise a stream in you know in in most meaningful senses. You know each event is a thing that comes that comes in and then the next thing comes in and so on and so forth. But they don't really
[03:38] represent any kind of table in and of themselves. there are things that are happening to some underlying table. So in order to the in order to handle these cases, we kind of need a different
[03:51] abstraction. One of the one aside I'll note here for people who are familiar with structured streaming. We do have that something that we call output modes today that sounds that it sounds like it it works
[04:03] and does work to some degree, but it's not quite as expressive as what we need in the in the general case here. So we have our update output mode which it can express inserts and and updates in the
[04:17] incoming stream. The there there are caveats here around how do you identify what is the key that you are updating but you know as I mentioned earlier there there there's always caveats it mostly works.
[04:31] What it can't do though is express deletes at all. There's just nothing in an output mode update stream stream that can effectively ind indicate that and there's no concept of nonprocessing time
[04:43] sequencing when a new batch comes in and it results go out. You know the effective time of those results is when the when they're output there's there's no way to there's no way to really say
[04:56] you know my event time is a bit different and that's the one that should control. We have output mode complete as well which it can it can express anything but the way that it does that is it says the
[05:09] output of each stream of each streaming batch is all of the of the data which has obvious performance problems in and of itself. It also has the problem that sometimes it's just not possible. you
[05:21] know, if your s if your source doesn't have infinite retention and you don't have the storage capacity to maintain an infinite upstream snapshot, you know, basically you can't home roll your own
[05:33] retention, you can't you can't use it. It just won't work. So, the general pattern that we use to resolve this is what's called the what's called chain change data capture. Uh
[05:47] this is a this is a different kind of streaming abstraction that exists in a number of different systems. Uh each system tends to like frame it slightly differently and use slightly different names for for things. Uh the schematic I
[06:02] have here is is kind of optimized for clarity of presentation in the diagrams that I have that I have drawn. You won't typically you you probably won't see in whatever system you have in mind these exact
[06:14] fields with these exact names. But all the all change data capture systems have basically the same concepts. You know there's a key which represents the identity of the row to which a change is
[06:26] happening. There's an operation that says what's happening to this row. Is it inserted? Is it updated? Is it deleted? There's there's a sequence which is the event time sequencing. you know when
[06:39] logically did this event happen and the payload that defines what is the change that is happening. So in the in this in this example diagram here uh what this
[06:52] represents is at version one there were two inserts for row rows A and B. At version five there was an update to row A that changed some of the data into
[07:04] into new data. And at version 9 there was a delete to row a that removed it from the from the table. Obviously the next question is okay you know you have a different streaming
[07:16] abstraction you know what do you do what do you do with this? How do you ingest this and turn it into the tables you want to see. Uh before we get into the details, I kind of want to motiv motivate here why we want to do this as
[07:29] part of Spark declarative pipelines in in particular and the re the reason basically is the components of the of declarative pipelines give us the right tools to do this effectively. Uh you
[07:42] know obviously we get Spark which is a very powerful distributed data data execution engine. You know, I I really try not to take that for granted as I work with it, as I work with Spark that,
[07:54] you know, I can just define a simple job job specification and it can run on two machines or 20 machines and it all just works under the hood and it gives me it gives me nice little pictures in the Spark UI of here's what's happening and
[08:07] how, you know, it really it really it really is very powerful and very very nice. And like I said, I try hard not to forget that uh the declarative nature here is also very is also very helpful
[08:21] because it means that we don't have to jump through hoops to express the logic we need as just one just one query or and we don't have to store it in just one in just one object. uh you know we
[08:35] do we do in fact leverage that second part pretty pretty heavily here as we'll get into uh the store the storage representation we use includes both a main table that has like the you know
[08:47] real air quote results of the of your data and an auxiliary table that that maintains some of the version tracking info that doesn't really fit well within the within the main table.
[09:00] uh the fact that the pipelines part of it also that also is very very helpful. uh we don't because because it allows streaming and batch semantics to kind of coexist you know we don't have to do the
[09:12] normal stuff of oh you know I'm using streaming so all these batch operations they no longer you know they're no longer relevant I'm not allowed to use them in declarative pipel in declarative pipelines you don't have to worry about
[09:24] that that so much and despite all this complexity the user surface remains very very hands-off you know you don't you don't have to worry about about managing
[09:36] the multiple jobs or how to you know how do I keep track of all of the of all of the underlying stuff you know you define what you want to happen and it all did it all just works
[09:52] so uh what what is there that what is the result of get of doing this ingestion on the surface with just with just the change the change feed that we've discussed so far it seems relatively simple you know at version one you have these two rows and the and
[10:06] the table has those two rows. At version five, there's an update and the table has those rows plus the update. At version 9, one of the rows got deleted. So, you know, there's a delete for row
[10:18] B. It's no longer in the table. It's now just A with the updated state. The problem is there there's a difficulty I'm eliding here. Here we have the event time order and the
[10:31] processing order matching precisely. And when that's not the case, it gets much harder to deal with. So, you know, in the in this order, you know, we ingest an update for A at version five and we
[10:45] store something like, you know, we have that we have this data for the for row A, but it's an update that's being applied on top of a payload we don't have yet. So, you know, what do what do
[10:57] we do with that? Then we get to a delete of the of row B, but row B doesn't exist. So what does it mean to delete this row that doesn't exist? Then we finally ingest version version one that
[11:09] has the that has these inserts that in the event time order were supposed to come first. And now it's no longer as simple as saying okay you know we get the inserts we insert them into the table. We have to somehow that somehow
[11:22] know that okay now that we have these inserts we can we can finally apply that stuff that we were going to apply earlier but we couldn't because we didn't have the the right stuff. And
[11:34] this is this is in fact the simplest of the out of order of the out of order difficulties that we have. There's a there's a lot of there's a lot of detailed edge cases here. Uh you know with a couple a couple I'll go over
[11:47] here. How long do we keep the deleted the delete tombstones? So, uh, if we go Oh, going back doesn't work on my clicker.
[12:02] There we go. All right. So, what one obvious one obvious way to implement the the ingestion at row 9 here is to say, okay, well, B is deleted. So, let's just keep a record that B was deleted at version that at version 9. And then
[12:14] whenever the insert comes in, we'll just we'll just, you know, we'll we'll know that it really should be deleted. But how long do you how long do you retain this tombstone for row for row B? You
[12:26] know, if you retain it forever, that means that none of your deletes are ever physically effective. You know, you're going to maintain maintain a row for everything that was ever deleted since the beginning of of time. If you delete
[12:38] it too soon, you might delete it before you ingest the insert that it was supposed to match to. and you will end up having be B just stuck in your table forever even though it was supposed to be deleted.
[12:51] Uh there's there's similar there's similar issues even without considering deletes when you have what we what we call partial updates. Uh we'll get we'll get into a bit more detail on that at the end of the at the end. But the TLDDR
[13:05] is when when you're updating only specific columns within a row, there are situations where the result of performing a m multiple updates is order dependent. And and then you know how do
[13:17] you how do you get them to be in the in the right order. you even more importantly all these all of these things and other other sub features of this kind of multiply on top
[13:29] of each other that you know despite despite our passionate investment in getting that getting the abstractions to be as simple as possible yeah that we have I I pasted the copy in here of our of our nice what what is this like 12
[13:42] line edge case where you know in in four hyperspecific conditions there's like an internal aggregate that certain rows need to be excluded from and with it we've verified with our fuzz testing
[13:54] that all of this is in fact required. If you remove any of these conditions, you will get an ingestion result that that is not order in order independent. So how do we resolve all this? Uh our
[14:08] solution is well what we what we said at the beginning of the talk auto CDC. So uh the primary abstraction here or I don't know if I would say primary the
[14:20] first abstraction here is we have the auxiliary table I I mentioned uh this is this is a bit of glossing over some of this stuff is stored as like uh columns within the within the main table but you
[14:34] know there's auxiliary information that contain that contains stuff about for the you know for each row when is the latest upsert we know about when is the latest delete within delete we know
[14:46] about uh one corlary that I should me that I should mention here because it will come up a lot in my further explanations is because of our event time event time versus processing time
[14:58] that time thing basically in order to maintain the invariant that that the results do not depend on uh the results don't depend on processing time ordering we inserts and updates end up both
[15:12] collapsing in behavior into an upsert case. So we don't actually have to treat them differently which is very convenient from an API perspective if you don't have deletes but also we'll get it we'll get to that any anyway what
[15:24] do you do once you have this when each new batch of change data arrives uh you get you go you go into the change data you see what key what keys are being modified here like what is the full key
[15:37] set that this change data batch is the is affecting we grab that batch out of the out of the target table that are not part of that part of that batch and we
[15:49] recombine them using the target data and this auxiliary data that I mentioned we recom we recombine them into the correct event time order state. So basically we we rearrange things so that we get the
[16:03] results that we would have gotten had we ingested every every change event we know about in event time order. And once we did it, once we do that, we take those results and we put them back into
[16:15] the target table, back into the auxili auxiliary table, and then then we're done. Uh, one, one more thing I want to me mention about the API before we get into a a kind of detailed execution
[16:28] example here is with the declarative pipelines really does let us get something that's a lot simpler and and a lot clearer here. you know to do the kind of things that I that I'm
[16:40] describing out the outside of autocd requires you know setting up a stream with a for with a for each batch call back and then within that for each batch call back you implement this hyper complex merge that you can kind of see
[16:53] is spilling off the end of the page the end of the page here in practice what what we see that what we see and what we saw pretty regularly which prompted us to build this in the first place is it's not even it it's not even this
[17:06] straightforward what happens s is the merge that the merge that people write and even that their agents write ends up being wrong. It end it ends up having cases where that where that the result of some outof order ingestion isn't
[17:19] right or there's some kit some partial updates that that put the wrong value put the wrong values in and you know we would routinely run into issue run into issues where we had to where we would went in and debugged and we're like oh I
[17:33] okay the problem here is that this merge clause needs to be it needs to be up here in autocd pack we package all of that into just very straight straightforward declarative.
[17:45] You know, I want to auto CDC. It's going into this target. It's reading from this stream. You know, here's my keys. Here's how I know when it's a delete. And like
[17:57] I said, if if you have a workload that doesn't have deletes, you just don't have to worry about operation types at all. You know, you have your event time sequencing and you have how it and you have how you how you want to store it. It's SD type one is a bit of a spoiler
[18:11] for uh second to last slide here. No, last slide. Last slide here. We will get into more detail about that later. Uh for now, I want to step the step through what this execution looks looks like for
[18:23] a for a simple example here. So we have our target table here with you rows A through rows A through F ingested at a variety of different versions. The motivation for these numbers will become
[18:35] clear as we work through the work through the the example. And we have we have an incoming source batch. In this source batch, uh rows A and B have new have new updates coming in at versions 9
[18:48] and two respectively. And rows C and D are deleted at 9 and 9 and two respectively. So our first step over the going going back to the you know three-step three-step slide our first step is to
[19:01] reconcile the source badge uh with the the relevant segment of the the current target table state. So we have the row row A uh the incoming that the incoming
[19:15] upsert is newer than the one we know about. So that stays the same. for row B the incoming upsert is actually a lower version than the one we know about. So we have to rec we reconcile that by
[19:27] saying no you know the up the update that's already in the table the one at version 7 that's the one that should control for row B and C and D go similarly you know the delete the delete
[19:39] the delete at C really have it really happens because it's at a newer version the delete of D does not does not happen because we that happened at version two so when we look at this we infer okay
[19:52] you know that when we had we have upsert for D at version 7. That means that someone between version two and version 7 must have must have reinserted D. So even though we even though there's a
[20:05] delete at version two, the result of our ingestion should have it. And row E, obviously it there was no row E in the source batch, so we don't have to worry about it. Row F went missing in one of
[20:18] the slide revisions, so don't worry too much about that. Obviously in a realistic case there would be there would be a bunch of other row rows kind of spill spilling off the bottom here. You know everything here works if the
[20:30] the source batch is touching most or all of the target table. But in typical usage it's going to be a relatively small subset that's being affected. Anyway, once we have that reconciliation
[20:42] done, we can then we can then take the reconciled source source data and merge it back into the back into the table which again this is an abstraction over the t over the table and the auxiliary
[20:55] theit thing. So you know row row a gets the new upsert. Row B gets the result from the from the reconciled batch which is just its original value. Same you
[21:07] know same thing for row for row D. uh row C is deleted is deleted. That actually is one of the things that will not go into the main table. That goes, you know, that goes as a tombstone as a
[21:19] tombstone record into the auxiliary into the auxiliary table. And as we discussed earlier, we'll hold on to it for a bit in case there's any inserts for C at below version 9 that we later adjust and
[21:31] know that we need that we need to not reflect in the in the final results. So this what we've discussed so far is auto
[21:43] the auto auto CDC as it as it exists in or as it will as it will exist in what is currently the release candidate of the of Spark 4.2. It has not been voted on yet. So it is not officially Spark
[21:56] 4.2 that we're continuing to do to do a lot of active work on this. And there's a few things I want to touch on because they heavily they heavily motivate a lot of the abstractions that we're that we're using here.
[22:08] Uh one is partial upsherts which we mentioned a bit we mentioned a bit earlier. Uh some sometimes your change feed has what are called partial upserts where you have you have an update that's
[22:21] coming in but it's only effective for some of the for some of the columns. So you know you might say you know my row A it has column columns uh where where's a good place to start from? It has columns
[22:33] D E FG and I have an update for columns D and E. I don't have one for columns F and G and I want to do and I want those to be unchanged. You know, whatever whatever is currently there I want to
[22:46] stay there which then it sounds fine on its own terms and it and it is but it's currently there in the event time dimension and that introd that
[22:58] introduces a lot of a lot of complexity here because there are cases where you have to where you have to know. Yeah. So you can you can have you can have a row that a row in column A where it has
[23:11] value X at time at time 10 but that X has been has been propagated onwards by up to by updates to other columns and that that value X was actually ingested
[23:23] at version five. In that in that case, it's no longer that it's no longer true that when you get it when you get another update at version seven, you can just discard it because we're already at version 10. You have to you have to know
[23:37] this one specific value hasn't been that you know it hasn't been ingested that re that recently. So that should control and and you know you have to split the you have to split the rows the rows
[23:49] apart in the in in the way that it's mostly depicted depicted here. You know, but basically what you end up having to do is maintain per column ver version sequencing to understand you know I am I
[24:02] may it may currently be version 10 and I may have this column value but when did I actually get that column value another another future thing is what we
[24:14] what we called SD type two. So our previous discussion has been based on the premise that what the user wants to see is just you know a replica of their sort of their source table. You know it
[24:27] may be a real source table in a database. It may be you know I want a replica of the account to which these to which these change events have been applying. But you know that is what they want what what they want to see. What
[24:39] people often want to see instead is a version history of the of a row. So you know they might want to say okay from versions one to three row A had that had these values but then in version four
[24:53] this value came in and then in version seven this other val this other value came in and then finally in version 9 it got deleted and I haven't and I haven't seen it since. Uh this means that our
[25:05] version tracking is now two versions instead of one version and they're they're us they're more directly user that user relevant right like the versioning is itself part of what the of what the users want to ingest obviously
[25:19] as an effect of that uh out of order handling gets much more complicated and the that is something that I could like talk about for an hour straight so I'm going to attempt to restrain myself the the core motivating case I would say is
[25:33] that This produces scenarios where you have to split rows apart. So in the in this uh example type two table at the bottom here you know you have where that you have a row a with such and such
[25:46] state from version one to version three. If you get an update at version two then there's no single thing to which that applies. You have to did it you have to dig into this row and split it apart
[25:58] into two different into two different rows. you know, one that was one that's relevant from version 1 to two, one that's relevant from version two to two to three. And as you as you might expect, having to do that then
[26:11] multiplies the complexity even further across a lot of the across a lot of the other the other stuff and all the out all the out of order cases that we've otherwise discussed.
[26:28] So uh yeah that's that's about the extent of that I stand my material here. Uh like I mentioned this is an act this is an active area of you know discussion investigation implement implementation in the in Apache spark. So it is
[26:42] available we're always on the on the mailing list and happy happy to discuss future development.

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.