Better Together: Change Data Feed in a Streaming Data Flow

Summary

• CDC streaming solution: 84.51 built a streaming data flow using DLT and Delta Lake that processes inserts, updates and deletes while maintaining streaming checkpoints for efficient data ingestion.
• Massive scale: The platform handles 8 million sales transactions daily with 80 million items purchased, processing both new transactions and updates to previous transactions reliably.
• Materialize and refresh: By combining change data feed with materialized views and Enzyme incremental refresh, the solution delivers a low-code, efficient and robust end-to-end data flow.

Better Together: Change Data Feed in a Streaming Data Flow

Talk by Mattias Moser, Data Engineer & Architect, 84.51 LLC ; Scott Gordon, Lead Data Engineer, 84.51˚
Traditional streaming works great when your data source is append-only, but what if your data source includes updates and deletes? At 84.51 we used DLT and Delta Lake to build a streaming data flow that consumes inserts, updates and deletes while still taking advantage of streaming checkpoints. We combined this flow with a materialized view and Enzyme incremental refresh for a low-code, efficient and robust end-to-end data flow. We process around 8 million sales transactions each day with 80 million items purchased. This flow not only handles new transactions but also handles updates to previous transactions. Join us to learn how 84.51 combined change data feed, data streaming and materialized views to deliver a “better together” solution. 84.51 is a retail insights, media & marketing company. We use first-party retail data from 60 million households sourced through a loyalty card program to drive Kroger’s customer-centric journey.
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

• 0:00 Introductions and overview
• 1:33 Streaming source CDC
• 4:56 Implementing CDC in DLT
• 13:38 Delta Lake change data feed
• 18:16 Processing Delta change feeds
• 26:34 Summary and conclusions

FAQs

What is change data capture (CDC)?

CDC is the process of recognizing when data has changed in a source system (inserts, updates, deletes) so that downstream systems can take action based on those changes.

How much data does 84.51 process?

84.51 processes around 8 million sales transactions each day with 80 million items purchased, while also handling updates to previous transactions using streaming data flows.

What is change data capture and why do teams use it?

CDC captures transactional changes from source systems in real time, enabling downstream systems to react immediately to data changes. Teams use CDC to power real-time analytics, replication, and event-driven architectures without the cost of full table scans.

What tools enable change data capture beyond Debezium?

While Debezium is a popular CDC tool, the Databricks Data and AI platform provides Delta Live Tables with native CDC support. This enables unified, SQL-based CDC workflows without external tools, simplifying architecture and improving data quality.

What is the best data engineering platform for streaming data flows?

Lakeflow on Databricks simplifies streaming data pipelines by unifying CDC, streaming ingestion, and real-time analytics in one platform. It handles stateful operations like joins and aggregations natively, with built-in observability for production reliability.

Full transcript

0:00 Introductions and overview

Hello everyone. Good afternoon. Appreciate you being here. I know it's late in the day and these days can be kind of mentally exhausting. So, we appreciate you sticking around for this. Good afternoon. I'm here with Matias Moser. We're from 8451. Uh, kind of laughed at me when I put this slide together because he noticed I pulled the photos off of our employee internet. These are our employee photos from like 10 years ago, I think.
So, he said they're going to think it, you know, the dads showed up to do the presentation instead of the kids. So, I guess if we get invited back next year, we'll probably need to update those. But anyway, so I'm a data engineer at 8451 TSU. I'm a data engineer and data architect at 8451. Okay, great. A little bit about our company. So 8451 is the market research, data science, and media marketing division of the nationwide grocery retailer, the Kroger Company.
We capture and analyze over 8 million instore and digital grocery sales transactions each day along with digital shopping behavior uh across all the outlets in Kroger's ecosystem. So that includes stores like Kroger,, Teter, Smiths, Fries, King Supers, Marianos, several other smaller banners as well. Uh we're gonna today we're going to be talking about change data feed in uh in a streaming data flow. We have two different examples to show you.

1:33 Streaming source CDC

First of all, what it looks like coming from a streaming source system and then secondly we'll show what it looks like when the change data is coming from a delta lake table. So there you go my perfect. So I will start talking about change data capture from a streaming source and um illustrate that with an example from our work. So what is change data capture? Here's a definition. Change data capture is the process of recognizing when data has changed in a source system so that a downstream system can take an action based on that change.
And in the most typical use case, our source system is a database relational database and the changes that can happen on the source systems are new records can be inserted, existing records can be updated or existing records can be deleted and um that's what I want to work with. Our specific example is Kroger online orders. If you go to kroger. com or any of the other banners, then if you place orders, these orders are stored in a database, when you create an order or when you start an order and an an order entry is created.
When you make changes like you add an item, then these orders are updated and in some cases an order is deleted. And um that's the example that I want to go through. will will greatly simplify that but um that's the example to work with. So one more time on the left hand side I have pro. com the website and um it updates orders in my orders database or orders table in the database and what we want to do is we want to analyze this data in data bricks.
So the question is how do I get the data from my operational relational database into data bricks and we want to do this in near real time. So the question is how do I do this best and this is where change data capture comes in. So on the orders database on the left hand side I have to enable change data capture works differently in different um in different systems but the general concept is is always the same. It captures every change on my table and you see five records here as as a simple example.
The first two records here are inserts. So a new a new order was created. that was inserted in the orders table and you see order ID, custom ID, status and items those are the simple that's a simple simplified repres representation of my order and then I have a time stamp and an operation that's not the actual data in the database but that's um data for the changes. So at a specific time um an insert happened or further down you see two updates.
You see highlighted what changed. In one case the status changed in another case an item was added. And then you see an example for a delete. So order with order ID 1620 was was deleted in this in this example. Good. So what do I do with this with this uh change data? In the typical use cases that some of you probably are familiar with, the change data is pushed onto a streaming system. In our case, we work in Azure.

4:56 Implementing CDC in DLT

We use event hub for that. Other systems that uh that you might be using are CFKA or GCP popsup or in AWS Kinesis. Um and on the right hand side I want to take that stream from the streaming system and persist it in in my analytical database. So in data bricks in our case all the way to the right you have the orders database in in data bricks that represents um a replica of what I have in the in the operational database.
And on the way there I have a bronze table um that captures all the changes. I'll go into this in more detail on the next slides. Before that here is just one example event as it comes from the streaming system. Typically we get this in in JSON and you see this is this was one of the inserts. You see the time stamp and the operation then the data for the order and then these streaming systems like event or cfka they send additional metadata.
This came on a specific topic. there's another time stamp that's now the time stamp when the data was pushed onto onto the um streaming system and there might be additional streaming metadata that you get with um with the data that you receive. Here's what this looks like specifically implemented in in DT. And I learned earlier that DT will no longer be called DT. I think it's now lakeflow declarative pipelines but I will continue to say DT until I get used to the new naming.
Good. So one more time on the left hand side that the orders database fed from kroger. com and then on the right hand side I have a bronze table to the left and then I have a silver table and that silver table should represent what I what I have in the operational system um in that backs our our website and in between I have a few um that I use for any transformations that I want to apply the store in the orders table the data as raw as possible.
So I have everything um I have all the data that I've ever received and transformations I can apply on this on this using this waters bronze view. Good. Here's the definition of the waters bronze table. I have to look here. This is a little small. Um so what happens here? I define the table. Then um this reads from a stream read stream format CFKA. You might ask why is this CFKA when I earlier said um we're using event hubs.
Events implements an interface implements the CFKA interface. So we can use that at some point. DT might support events natively and we could maybe do a readstream form at event hubs. I have to pass options in authentication the location or the address of the Azure event hub um servers and then I load that data and what I do here is my the value that we earlier saw in JSON actually comes in as a binary value. So I cast it as a string and then the last statement here converts this to um or converts it to a value column that has the actual schema applied.
Get order schema the function here imagine gets the schema that's that's defined um elsewhere. Good. Second second um step in my DT pipeline is the order bronze view. Very simple example here. In practice we will probably do more here but it just selects the val selects value star. So it breaks out that one value column into multiple columns the attributes in that JSON um in the JSON value that you saw earlier. And finally here's my silvers table.
My silver table. So um and here is where the DT magic happens. So now we need to take all these changes and we need to apply those changes in in our delta table. And um I could do this with a merge statement. If you were uh if you just heard the other change data capture talk um in room 2005 an hour ago, they talked about that it might feel simple to implement this with a merge statement, but there's a lot of um there's a lot of details that you need to that you need to take care of.
So, it's very convenient um to just be able to use apply changes and not have to worry about all these details. What does apply changes do? The source for my apply changes is the bronze field. The target is the silvers table. I need to provide a primary key. That's my order ID. And apply changes will then automatically use this order ID to see whether it needs to apply an insert or an update. So I don't need to tell um apply changes that.
If the order ID already exists, then it will apply an update. If it's a new order ID, then it'll do an insert. I need to tell apply changes how to sequence my records. records might on a streaming system not always arrive in order. So I use the timestamp that comes from the original database because I know that's how the changes happened on the original database. That should be a reliable value to sequence my my changes by.
I have I have deletes. Apply changes has this parameter apply deletes. And if the operation is a delete, then I tell it that it should apply deletes on that. And then finally I tell I I use the parameter stores sedd type one to say simply apply the changes as they come along in my in my target table. Apply changes also supports sedd type two. So you automatically get an sedd type two table if you want to that captures all the changes or preserves all the changes for you.
So you could look at what did your data look like yesterday, last week, a month ago. Good. With that, a few considerations for when you use apply changes. It does use merge under the under the hood. Again, here's a reference to Ray's presentation mastering change data capture with DLT. If you wanted to look at that video once it's um once it's provided, he talks about that. What's under the hood isn't a simple merge statement, but it's probably a couple thousand lines of code that deals with all sorts of edge cases that you could that you could run into.
What you need to make sure is you need to provide a well- definfined primary key. Really simple in our case with our orders. And for performance, you have to make sure that order ID can be looked up quickly. So either use petitions or zordering or use liquid clustering if if um if you're already using that. We in in our cases a lot of our data is still partitioned and C ordered good. You need a reliable sequence by column.
So time stamp or some numeric sequence value that comes from the original database. And then you can use expectations DT expectations um to manage any type of unexpected records or bad data. In our my very simple example, I'm just assuming the data looks the way it should and that's really all um a very simple CDC use case and will talk about what this looks like um if you work completely in data bricks. Okay, thank you.
So, uh, now we're going to take a look at what change data looks like when it comes through a Delta Lake table, uh, rather than through a kind of a a general, you know, change data capture interface. Um, the example, the flow of the data is going to be pretty similar to what just described, but there's there's some differences in the way uh, change data looks in a Delta Lake table. So, we'll get into those those details.

13:38 Delta Lake change data feed

So for continuity of this presentation, I'm going to build off of the silver table that populated with order update activity. Uh we're going to build a DT pipeline that creates an order status table by consuming update activity from order silver. And for this data flow, I'd like to use a streaming read because I don't want to read the entirety of the order silver table every time my pipeline runs. I I want to just process new activity with each each run of the pipeline.
And in addition to that, I know that the order silver table uh along with inserts is is going to likely have updates and potentially maybe even deletes as well. So I I need to handle that. So a little bit of context about reading a Delta Lake table using Spark. So when you read a Delta Lake table using a spark read, it's just like saying select star from the table. You get the entire contents of the table with all the rows in their your current image as as of the time of the read.
When you do a read stream on a Delta Lake table, it's like saying select star from the table where data is newer than my last read stream. So read stream automatically keeps a checkpoint upon successful read of the table. it it checkpoints where you were at in the delta log when you completed that read so that next time you call read stream it can go to that checkpoint and say okay what activity is new s since that read that checkpoint is it's part of spark structured streaming it's completely maintained for you it's invisible you don't code anything you don't see it DT will pick the location of the checkpoint and maintain it uh for you on your own So you think about using read stream on uh the original source like talked about using Kafka or event hubs you know a traditional streaming platform read reasoning what readstream is doing is pretty straightforward right it's just going to the topic and requesting new new events from the topic right so you get each event once and only once and in traditional streaming data flows they were referred to as as appendon data flows because that was the most common use case, you're always receiving new events in and you simply append those as records to your target table, always appending.
Um, well, when you think about reading a Delta Lake table, well, now you're reading, you know, a table, a static set of data. Well, that data could have gotten there via inserts or records could have been updated or records could have even been deleted. Well, your basic read stream, this appendon flow doesn't handle that. As a matter of fact, if you were to call uh read stream just as the code is here on a Delta Lake table that had updates on it, you're going to get an error.
Readam's just going to come back and say, "No, I see updates. I don't know what to do with those." So, there is an option you can throw in there to get around that. There's this skip change commits. So, if you add that option as you're reading through the table, it will read the inserts. when it sees an update or delete, it'll simply just skip that activity in the log and go on looking for more inserts. Okay, you've suppressed the error.
Great. But you you see the what's missing here. Right now, your downstream table is not up to date with the upstream because you chose to ignore those updates. So, technically it works, but probably not the output you're really looking for. So now better together we get to the title of this presentation and what inspired us to put this together in the first place by combining a read stream on a table with this option read change feed.
Now you have a mechanism an interface for which you can consume not only new inserts but you can consume successfully consume and process updates and deletes from that source table as well. Now, in order to be able to do that, to use the R read change feed option, the table you're reading from has to have uh change feed enabled on it. So, when built his uh silver table for me, he he turned on this feature, enable change feed, set that to true.

18:16 Processing Delta change feeds

What happens when you set that on a table is um if you picture a you know what a delta lake table looks like in storage it's a folder with a bunch of parquet files a subdirectory with the delta log and the delta log tells you which parquet files are are part of which version of the table well when you enable uh change feed you get another subdirectory and your change information is written there so it lives alongside the table data uh as part of that table but but separate from the actual table data itself.
Okay, so back to our our pipeline. Um no, I'm sorry. We're let's look at an example of some change data first. Um so so this is what it looks like when you call that read change feed. You see there's three metadata columns that come through in the data underscore change type underscore commit version underscore commit timestamp and then along with all of the other attributes for each row of data you'll see inserts updates deletes and uh specific to delta lake change feed format in the case of an update you'll see two rows you'll see the pre-image which is what that road looked like before it was updated and the post image which is what that row looked like uh after it was updated.
This is a little different format than some other uh change data feed patterns like what was showing. Some systems will just simply show you what changed. So they'll only show the attribute that had a change and not the rest of the row. That it gets a little complicated to deal with. Again, if you were in this session before ours, which if you weren't, I encourage you to go look at it online. Uh Ry talked about so how do you handle it when all you know is what the one column that changed and all the other columns are effectively null well you don't want to you know null things out right you just want to apply the change so that logic gets very complicated apply changes does handle that but when they crafted change data for delta lake I guess knowing how complicated that is they said hey let's just present the entire row right so I don't have to worry about um you know which attributes changed and which didn't.
I just know, oh, this is the current image of the row and that's what I'll apply. I have some more detailed examples I'll show you. Uh, so again, back to our our pipeline we're building. We're simply going to read from that silver table uh using the read change feed option and then we're going to apply those changes into uh an order status table. So what that code looks like when I'm when I'm reading uh I'm going to just declare a DT view.
It's going to do a read stream on the order silver table. And as I showed you previously, use this read change feed option. I'm also going to take a couple of the metadata columns and rename them. I want to actually capture the change type and the commit timestamp into my order status table so I can see what those look like when they came through the data. I don't want to persist them with the names underscore change type and underscore commit timestamp because those are those columns are reserved specifically for you know delta lake change data capture.
So simply rename them and I can capture the data into my target table. So here's what uh in this example we've been showing you. Here's what it looks like coming through change data feed. Again, I've renamed the change type and commit timestamp so I can keep them in my data and store them. So, we had order 2001 was inserted. Order 2002 was inserted as well. Order 1784 was updated. Uh again, you'll see the pre-image and the post image.
And we can see the status changed from open to ship. However, again, we do get the entire row here. I'm only showing you a few attributes, but but I get that entire row of data as it looked like after it was updated. Um 2000 order 2002 that was previously inserted got an update a few seconds later. Um you can see the items array was ch an item third item was added and then order 1620 was deleted. And again a key thing here is in that delete activity I still see the entire row.
So I see it what it looked like at the point where it was deleted. Okay. So now here's the right the right to order status. I'm going to use apply changes as already described to you. It basically works like what we commonly call an upsert, right? It's going to conditionally either or insert or update depending on if if uh if that key is in the table or not. So there's there's actually four things apply changes can do. when it sees an incoming row, it looks at the target table.
And there's four uh outcomes o of that. It can choose to insert that row into the target table. It can choose to update the corresponding row in the target table. It can optionally decide to delete the row and that would be if you're using this apply as deletes option and the row meets the delete condition, then it could delete it. There's actually a fourth option it can do. it can actually decide to ignore that row and just drop it and not apply an update.
When would it do that? Well, that's to handle that case of changes coming out of order. So, as soon as it sees that order ID is in the table, that becomes a candidate for an update. But then the next thing it's going to do is evaluate the sequence and it's going to say, is the sequence of the incoming row newer than what's already in the table? And if it is, then the update is applied. If it's not newer, that's the condition where it'll just reject that row.
Just drop it, ignore it. And that's just that guarantee to make sure that you're the output of apply changes is always the latest data, the most latest most accurate version of that record and it's never going to update uh newer data with an older value. Now in delta table uh change data feed, it won't come through out of order. you know that interface is written where it it will deliver to you the changes in in the proper chronological order but nevertheless the sequence by is there you know for use in in other cases where you know like if your upstream was a you know Kafka or a streaming platform since you're not the producer of the data you know you can't guarantee that producer doesn't sometimes accidentally put things into the stream out of order so the sequence by will protect you there.
Okay. So, this is what order status looks like after this first run of data comes through it. So, you see order 2001 was inserted. It captured the change type there. Uh 1784 had an update. So, our our results are that post image of the row where we see the status was is now shipped. Um order 2002, if you remember, it had an insert followed by an update. So again the out output of apply changes is we have that final image of the row that is uh uh the third item was added and that's where the benefit of the change data having that entire row is beneficial right because all I need to know is the final update post image and that tells me the current state of the record.
I don't need to know that oh items changed but status didn't. Right? All I need to know is what's the current latest version of this row and that's what gets applied. And then order which one we did we do is not in the table. That's because we told it to uh like I'll back up to that code a little bit. We had the yeah the apply as deletes. So is that showing? Yeah. apply as deletes uh was the action you know and it met that condition the source change type equal the word delete so that's what caused that one to get deleted so the deleting of the data is actually optional though that's uh you don't have to include that apply as deletes option so in this version of the code I just simply took out apply as deletes that that's all I took out so I said there's four outcomes of apply changes well if I remove delete as an option.

26:34 Summary and conclusions

Now, there's only three outcomes that can happen. An insert, an update, or ignore. So, what do you think it does with that delete that record that came through change data saying it was a delete? Is it going to ignore it? Who who thinks it'll ignore it? Good. All right. So, that it's not going to ignore it. So, that leaves insert or update, right? Well, following the rules, it's going to look and see, is that order ID already in the target table?
Yes, it is. Um, so it's a candidate for an update. Is the sequence of the delete newer than what's in my target table? Well, yes, it is because I have, you know, the version I have is whatever came through previously before it was deleted. So, it meets that condition for a delete. It is newer information than what I have in my table. So in that case that record's actually updated and because I'm capturing the change type that gets updated to the word delete.
I have the new time stamp showing when it was deleted and again I have that image of the row of exactly what it looked like at the time of delete. Now obviously whether or not you want to apply deletes or keep them is up to you, right? It's a business use case, not a a technical uh decision. But if you just decide to keep them, I don't Sometimes people refer to this as kind of a soft delete. I don't know if you agree with that or disagree, but regardless, I' I've retained the record in my table, but I still have a a change column that tells me, oh, this was deleted from the upstream, so I I can differentiate it.
Okay. So, in summary, using change data from a Delta Lake table, uh first of all, you have to enable change data feed on that table. uh that can be enabled when the table's created like if it's created in DT you can just add it as a table property in the pipeline it's created but it doesn't have to be you can add change data you can enable change data feed to a table at any time so it's not a feature you'd want to turn on in case you want to use it someday right because there is some overhead with tracking that change data with every update to the table so you can wait until one of your consumers says hey I'd like to consume changes from your table.
Okay, you you can enable it at that time and you can also disable it at any time. Then once you have change data enabled to read it, you'll read with uh this read change feed option. Now that is optional, right? The change data sits alongside the table data. So it's still a valid Delta Lake table. You can still read it, query it, select star from it. you don't have to consume the change data but once it's enabled then you have that option of reading the change data if you want.
Um and then generally you don't again you don't have to but generally you'll use apply changes on the target then to that's kind of the right side of the changes to apply the changes into the target table as you see fit. There's some references other presentations by 8451 this week which you already missed but you look them up online I guess.

Learn more about the Databricks Data and AI platform.