Getting the Most Out of Spark Declarative Pipelines: Deep Dive on What’s New and Best Practices

Summary

• This video traces the evolution of Spark Declarative Pipelines (SDP) from Spark RDDs through SQL to fully managed declarative frameworks, showing what the system handles automatically so engineers focus on what to compute rather than how.
• Michael Armbrust covers the Enzyme incremental optimizer, which chooses the most efficient execution strategy—merge, partition-replace, or full-refresh—for materialized views at scales reaching 200 trillion events.
• Advanced topics include real-time mode with single-digit millisecond latency, AutoCDC for change data capture in one SQL command, SCD Type 2, and metaprogramming patterns for managing pipelines at 10,000-table scale.

Getting the Most Out of Spark Declarative Pipelines: Deep Dive on What’s New and Best Practices

Declarative pipelines are becoming the default way teams scale production data pipelines in Spark. Maximizing their value requires understanding the execution model, tradeoffs, and best practices. In this 90-minute-deep dive, distinguished engineer Michael Armbrust explores how to get the most out of Lakeflow Spark Declarative Pipelines (SDP), drawing on real-world usage and the latest platform advancements. You’ll learn how to: Apply proven design patterns for building reliable, maintainable pipelines Use SDP effectively across batch and streaming workloads Avoid common pitfalls as pipeline complexity and scale increase this video is the definitive technical deep dive for teams scaling through declarative frameworks in Spark.
Talk By: Michael Armbrust, Distinguished Software Engineer, Databricks ;

Chapters

• 00:00 Introduction and SDP history
• 01:11 Why declarative: from Spark RDDs to SQL to pipelines
• 05:37 What declarative pipelines manage automatically
• 08:08 SDP at scale: 200 trillion events, serverless
• 09:24 Materialized views and Enzyme incremental optimization
• 13:04 How Enzyme chooses merge vs partition-replace vs full-refresh
• 19:18 Getting started: SDP IDE and Genie code generation
• 22:39 Replace-where flows for windowed updates
• 26:04 Streaming tables: fundamentals and Auto Loader
• 36:32 Stream joins and watermarking
• 43:43 Real-time mode: single-digit millisecond latency
• 46:10 Data expectations and quality enforcement
• 49:32 Stream time travel: rewinding pipelines
• 55:41 AutoCDC: one SQL command for change data capture
• 62:39 SCD Type 2 and bitemporal CDC
• 67:40 Deployment: git integration and dev/prod environments
• 72:55 Metaprogramming for 10,000-table pipelines
• 75:43 Escape hatches: Lakeflow Jobs integration

FAQs

What is Lakeflow Spark Declarative Pipelines and how does it relate to Delta Live Tables?

Lakeflow Spark Declarative Pipelines (SDP) is the same product previously known as Delta Live Tables (DLT). This video explains that the name evolved from Delta Live Tables to Lakeflow Declarative Pipelines to Spark Declarative Pipelines, but the core technology is unchanged.

How does Enzyme incremental optimization work in Spark Declarative Pipelines?

Enzyme is the incremental query optimizer embedded in SDP that automatically decides whether to execute a materialized view update as a merge, a partition-replace, or a full refresh. This choice is made at runtime based on the incoming data and query shape, so engineers do not need to hand-tune execution strategies.

What is AutoCDC in Databricks and how does it simplify change data capture?

AutoCDC is an SDP feature that lets you implement change data capture with a single SQL command rather than writing custom merge logic. This video shows how it supports standard CDC, SCD Type 2 history tables, and bitemporal patterns, all declaratively.

How do Spark Declarative Pipelines handle streaming and real-time workloads?

SDP supports streaming tables built on Auto Loader for file-based ingestion, stream joins with watermarking for late data, and a real-time mode that delivers single-digit millisecond latency. This video also covers stream time travel, allowing pipelines to rewind and reprocess historical data.

Full transcript

[00:08] Thank you so much for coming. I am super excited today to talk for a surprisingly long time about Spark declarative pipelines. Uh which is, you know, the key idea here is how can we make all those data transformations you need to
[00:20] do in order to get value from all of the data in your businesses simpler. Uh, the lawyers want me to say that some of this is forward-looking statements, although most of the stuff I'm talking about today has already shipped and is already available in the product. Uh, and of
[00:34] course, when you're done, please go and uh, fill out the survey because it helps us decide what to talk about next year. I also want to clear up a little bit of confusion in case you've been following us for the past couple of years. This is
[00:46] the same thing as Delta live tables uh DLT Lakeflow declarative pipelines and I told them I'm going to quit if they change the name again. Uh it is Spark declarative pipelines
[00:59] uh or STP. So uh without further ado, let's get started. I want to start with a little bit of history. Why did we create this project? And really the whole idea here of making things declarative, making data transformation
[01:11] simple goes back almost 13 years now. And so when I started at data bicks back in 2013, Spark was taking off. It was replacing Hadoop. Uh, and this was kind of revolutionary at the time. And so
[01:24] this is some Spark code using RDDs or resilient distributed data sets. And it's doing something pretty cool. It is uh it's actually distributing this compute to uh to compute an average over
[01:37] thousands of machines with just a few lines of code. And coming from Hadoop where you had to write these big complicated map and reduce classes in order to do that this seemed pretty gamechanging at the time. But now looking at this we could say that looks
[01:50] extremely complicated. If I just want to compute an average why do I have to think about which operations are commutative and associative. Think about them in terms of reducing and mapping and then uh you know summing the value up. I have to break it down into the
[02:03] addition and the division. Uh I want to make this a little bit simpler. And so the first thing we added to Spark to abstract away these complexities of thinking how to get the answer uh was this thing called Spark SQL. And so
[02:16] Spark SQL was this layer that sat on top of Spark. And instead of coding with functional RDDDs, all you had to do was write in SQL, which looks a lot like English, and tell it what you wanted to compute, and the system would figure it out. So now we just say select date and
[02:29] region. I want to know the average revenue from the sales table. I don't have to worry about where the sales table is located or how it's stored, how to parse it. I don't know have to tell it how to compute an average or how to do a group by. It just takes care of all
[02:41] of that stuff for me automatically. And so this was pretty cool because it allows you to do these things, these these ad hoc queries much more efficiently. But we're not done yet because you never want to run a query
[02:54] just once. If it's an interesting result, you want to run it today and tomorrow and the day after that. And data is constantly arriving over time. So there's actually still a bunch of work need that I need to do if I'm going to be a data engineer. And the next step
[03:07] here is I want this query to not just run once but to efficiently update as new data arrives. And so the next thing we added to the Spark stack was a system called structured streaming. And this made incremental processing declarative
[03:20] as well. You could see I changed that SQL query by adding one more word into it. Instead of reading from sales, I now read from the stream of sales, the new data that's arriving over time. And this
[03:32] automatically figures out what data is old, what data is new, transactionally computes what changes, handles failures if it happens in the middle, and guarantees exactly what semantics. And so now this will automatically update
[03:44] efficiently as new sales are occurring. But even this is still not done because I need to store these results somewhere where people can read from them. And I want to make sure that that's efficient and transactional. Streaming often has
[03:57] this problem that it produces lots of tiny files because it's running with low latency. and I don't want to have to worry about compaction and all of these other things on top of it. So the next thing we added was Delta Lake and this is kind of the beginning of the lakehouse which you heard a lot about uh
[04:09] you know today and yesterday. Um and so just by changing this to now write to a delta table instead all of those complexities of managing files and their schema and doing compaction are taken
[04:21] care of by the system and all I have to do is again declaratively say what should happen and not how it's going to happen. And so this brings us to about four years ago when we said, "Okay, this is great, but still there's a whole
[04:33] system I have to build on top of this. I want to turn this stream not just into something that I'm running from my laptop, but a endto-end production pipeline." And so the this is the goal. This is the thing that we're trying to
[04:45] accomplish this SQL query. But people found that in order to make it work in all of the places you wanted to work, we had to worry about managing checkpoints and where we were going to store them. We have to build a system around it to restart the system when it fails. We
[04:58] have to worry about dependencies because it's never just one stream. It's hundreds of streams that depend on each other and need to run in the right order. We have to worry about making sure those tables exist and their schema is correct and the columns are added properly over time. If there's many
[05:11] things running, we have to kind of manage the parallel execution of these so that it's efficient. We have to worry about compiling it and making sure there aren't mistakes. So if I make a change, what's going to happen to all of the things that are downstream? And finally, we have to build CI/CD pipelines because
[05:24] we're proper engineers and we want to make sure that when we make changes, they get tested before they're deployed to production. And so, as you can see, the the center part is where we're getting value from. But you're spending all of your time on the periphery, all
[05:37] of these extra things that are around it. And that's why we created declarative pipelines. And so, here's a story from one of our early customers at Navy Federal Credit Union, and they're saying kind of exactly what the goal was is that it hides the complexity of
[05:49] modern data engineering. And it with this kind of simple declarative programming model where you just say these tables should exist and the system figures out how to make that true. So uh let's uh dig into declarative
[06:02] pipelines hopefully. And so now you can see we writing here looks still just a lot like SQL but this is now a full endtoend pipeline. This
[06:16] first statement here, create streaming table sales actually creates a background job that is constantly running, ingesting JSON data from that sales directory, writing it into a table. It manages retries and all the
[06:28] other things that have to happen to keep that running reliably. It makes sure the table exists. If new columns appear, it adds those columns and restarts them all with just that one line of SQL. And so that's our bronze table. That's how we ingest data into the lakehouse. And then
[06:41] when I want to compute reports for the business users to uh run queries over, that's as easy as running another SQL query. Just create materialized view. I write that thing and it actually does a lot of stuff under the covers there.
[06:53] It's incrementalizing it. And for the rest of this talk, I'm going to talk about what these pieces are, how they fit together, and some of the cool stuff that's happening under the covers in order to make it possible. Now, I want to put this in the context
[07:05] of the larger LakeFlow product because there's actually a bunch of pieces here. So you know the the different parts you need to ingest, you need to transform, you need to orchestrate all of that. Uh today I'm going to be focusing mostly on declarative pipelines and the
[07:18] transformation. But just so you can understand how this fits into the larger lakeflow picture. We also have LakeFlow connect which is no code ingestion. So if you want to get data in, you can just point and click in the UI. You don't have to write a single line of code
[07:30] secretly under the covers. It's creating an STP pipeline. We use the same technology but it's a much simpler experience if you want to get data from Google Drive, Salesforce, Workday, all of these other operational systems. And then similarly, jobs is the orchestrator
[07:43] for everything. So if you want to schedule a pipeline, if you want to schedule a notebook, if you want them to run together, use a job. A question I get very commonly is when should I use jobs and when should I use pipelines? You should always use a job. If you're
[07:56] scheduling, you're using a job and you should use pipelines for declarative transformation of data often inside of a job. And you know just a couple of stats you know I think uh declarative pipelines
[08:08] has been around for about four or five years now. We now process over 200 trillion rows per day. And I think another really cool stat is the SDP customers are some of the kind of cutting edge customers of data bricks.
[08:21] More than half of our workloads run on serverless today, which means pretty much everything, including the infrastructure is declarative, which I I think is pretty cool. So, let's get started. Let's dig into what it takes to
[08:33] create uh your first declarative pipeline. So, it's as easy as So, first of all, let's kind of start at the beginning. What what is a view? How should I think about this? Really, a view is the simplest declarative construct because all you say is I want
[08:46] to assign this name to this query and then anybody who queries that name will get that result. So, if I wanted to create this report for all of my business users to be able to read things about, you know, the revenue by region, I could create this view revenue by
[08:58] region and it will automatically put it in the catalog and anyone who queries it would get this information without having to write that query out over and over again. But this, of course, is inefficient because anyone who queries that view is computing it on demand
[09:11] every single time. And so, I'm rereading all of the sales data for all of history and computing it from scratch. And of course, that's very slow. So for many years, databases have had this construct called a materialized view where instead
[09:24] of just running that query on demand, we're going to run it ahead of time and we're going to store the result in the table and then when you query revenue by region, you read those already computed results. So it's extremely fast. This is a great way to speed up your dashboards
[09:37] and and other downstream queries by computing the answer ahead of time. But how does that actually work? Uh so let's dive into this really cool technique called enzyme which is the system that lives inside of uh STP and updates those
[09:51] materialized views. So you might say well that's really easy. I've been creating materialized views forever. I have a bunch of notebooks and they run create a replace table and they overwrite the table with the new result premputed good to go. Well, that's one
[10:05] way to do it, but it's an extremely expensive way to do it because when just one new row appears in the table, when you run a create or replace table, you recmp compute the entire thing. And as your data gets bigger and bigger, you know, anyone who knows big O notation,
[10:18] this quickly becomes an N squ problem. You don't want that in your data pipeline. Uh so create materialized view is much smarter. It's going to perform this technique called incrementalization. And so incrementalization means that
[10:31] when one new row gets added in the input, only one new row gets added to the output or the minimum amount of work that needs to be done is done to keep that already computed result up to date. So it's no longer about computing from scratch. It's about updating something
[10:44] that was already computed in the past. And so how does this work? Well, there's a couple of different techniques. So the first technique is for what we call monotonic queries. This is an idea from relational algebra. A monotonic query is
[10:58] one where a new piece of input will always produce a new piece of output, but we'll never retract an output that was already made. So it'll never update anything that was there before. Sounds a lot like streaming. It's extremely efficient. So if we detect that a query
[11:10] is monotonic, all we will do is look at the new rows that are available in the table since the last time we ran, run the transformation on them, append them extremely fast as a fast streaming. But of course, this only works for select,
[11:22] project, inner joins, things that can be monotonic. And it of course can't handle changes to the input because that violates the monotonicity guarantee. So this is one technique that it uses. But that's not enough. Another technique that enzyme will use is partition
[11:35] recomputation. This is probably something that many of the data engineers in the room have done by hand. But it will automatically detect, oh, these two tables have the same partitioning. And so when a row changes, I will only update the partition for the
[11:48] data that has changed and leave the rest of the data exactly the same as it was before. So this is great because now we can handle aggregations and updates. Uh it requires these two tables to have the
[12:00] same partitioning. Sometimes you're updating more data than you need to. That partition might be really big. So this isn't always the the best one, but when it works, it can be very efficient. And so then finally, we have the most complicated and coolest enzyme
[12:13] technique, which is to replace uh the specific rows that need to change. So here we use a bunch of techniques from the database literature to actually compute given an update. So we have one row that changed one there row that was
[12:26] added and we actually compute the delta to the table that was premputed before and apply that delta updating only the rows that need to change. So this is great because it can handle extremely complicated queries that involve joins
[12:40] and aggregations. Uh it can handle updates, inserts, and deletes. Merge is a little expensive and that's why this isn't always the right choice if you're over overwriting a huge fraction of the table. Sometimes it's actually cheaper
[12:52] to recomputee the whole thing rather try to update stuff in place. And of course it's complicated to reason about. But that's the best part about declarative. You don't have to reason about it. So how does it pick the right technique?
[13:04] It's got all of these different options. How is it deciding what it should do? Well, Enzyme has something called a cost model. So we take the query plans that we analyze. We take the change feed from Delta to figure out how much data has been updated in the inputs. We run it
[13:18] through all of these different strategies to estimate how much each of them will cost and then we pick the cheapest one and run it automatically. So this is pretty cool because it's not picking statically based on the query. It's actually dynamically at runtime
[13:30] looking at how big the change set is and deciding the most efficient way to do it. And this is where again I think declarative programming becomes pretty powerful. A great programmer can always sit down and build something that's
[13:42] super efficient, but rarely can they actually have it kind of reoptimize itself as the constraints change over time. And there's some other tricks that make it a little even more powerful. I want to just go through a couple of them and
[13:54] then and then I'll move on. But another idea is some queries are actually very difficult to incrementalize. So for example, how would you incrementalize a count distinct? Now, a change to the input. I actually don't know whether or
[14:06] not, you know, that's the only time I've seen this customer or whether they're a new customer I've never seen before. You need to do a little bit of extra tracking in order to make that possible. And so, we use this trick called decomposition where we take a query and
[14:20] we break it into smaller, easier to incrementalize fragments. So, we actually create multiple materialized views that back a single materialized view. So in this case, in order to compute a count distinct, what we would do is we would create another table that
[14:33] just counts how many clicks there were per customer. And then just looking at the number of rows in that table gives you the count distinct. And so what's actually going to happen under the the covers is we'll store multiple materializations in Unity catalog. We'll
[14:46] put a view on top of that. And so now when you as a consumer read from that table, you will not have to know that any of this complicated stuff is going on under the covers and you will just get the correct answer.
[14:58] So that's how Enzyme works. But is it fast? Well, it turns out it's pretty easy to show how fast it is because recmputation is extremely expensive. We ran TPCDS, this decision support benchmark, but instead of running the
[15:11] queries ad hoc, we converted it to materialized views. And it's dramatically faster because you're not reading all the data. You're not recomputing all the results. You're only touching the stuff that changes. And it's also dramatically cheaper to run as
[15:24] well. So pretty cool. Um, now let's talk a little bit about how you actually orchestrate these. How do I actually take this single SQL query that I typed into a SQL box and turn it into a
[15:36] production pipeline? So by default when you create a materialized view, it's what we call a snapshot materialized view. So it's up to date when you create it and if you want to update it again, you run this
[15:48] refresh command. So you see refresh revenue by region. that will spin up a job in the background that does all of this enzyme incremental work to keep that table up to date. But by default, it only does this when you want. However, we want something probably that
[16:01] you can just fire and forget that will automatically stay up to date. And that's where Lakeflow jobs comes in. It's as easy to use LakeFlow jobs as just adding one more line of code. So, if I just add this line schedule every
[16:14] one day, then it's going to automatically in the background create this job that will run the refresh command once a day. And this is pretty great. However, there are problems with using schedules for this. It's kind of naive because now, you know, at at best
[16:28] my stuff is going to be, you know, up to date within a few minutes, but most of the time it might be up to a day stale. And sometimes maybe the input isn't even changing. Just picking kind of a blind schedule tells you nothing about how often the input changes. And so there's
[16:42] another really cool feature that we rolled out in the last year uh called triggers. And so if I instead of using a schedule use these words trigger on update now it's actually going to subscribe inside of the job scheduler to
[16:55] the change feed from those input tables. It'll analyze this query. It will say we depend on the sales table. And now only when the sales table has a change will it spin up compute to update this materialized view. So now it can be very
[17:08] low latency but also very low cost because you're only paying for compute when you need it. And you know, you might say, well, wait a second. The sales table is updated all the time, and I want to bound my costs. I don't want
[17:21] to pay too much money to update it. I don't care about freshness, you know, within more than an hour or something. So, you can add this other clause to control cost where you say trigger an update at most and then any time interval. In this case, we chose three
[17:34] hours. And so, now it will run no more often than once every 3 hours. So now you get this super low latency anytime the input changes, but also cost controls to make sure that it doesn't update too often. And so here we have a
[17:47] full incremental automatic pipeline that is smart with its data dependencies running reliably on top of the jobuler. And all I had to do is type this one SQL command into DBSQL. And we just rolled
[18:00] out the ability to do this inside of serverless notebooks as well. So MDs simplify doing these gold table transformations because they premputee these results incrementally. They
[18:13] automatically update while controlling costs and all of the background infrastructure stuff, failures, retries, autoscaling, they're managed by by the infrastructure. So you don't have to worry about this.
[18:25] So that's pretty great. But many of you are saying I don't have just one single table. I have hundreds of tables, right? So creating a one-off MV inside of DBSQL is great when you're just doing some exploration. You just want to speed up
[18:37] this one thing that you're going to do over and over again, but more often your actual job looks more like this. There are tens to hundreds of tables that need to be orchestrated. And often there are dependencies between them. And that's
[18:50] why we created the ID for data engineering. It allows you to develop an entire pipeline containing many related data sets. And you get all of these nice benefits of an IDE. You can navigate using the graph of dependencies. You can
[19:03] see errors as you type. You can see results without having to switch pages. Uh it has support for Python, SQL, and someday Scola. Uh and you get version control and stuff in CI/CD all built into it. So um it's pretty easy to get
[19:18] started. All you do in the data bricks UI is go and click new ETL pipeline and it'll bring you to this page that can even get you started with some sample code. So it's great. And this is whether you're SQL or Python, anything you can
[19:30] use. You can uh you can code using this. So here's a a video of what it looks like. You can see we've got our SQL code here. We can see the graph of all of the tables that we're creating. We can jump around and see the results just by
[19:42] clicking on them. Uh you can switch and look look at uh you know each individual data set, the metrics for its computation. And then you can even click and rerun individual tables just by clicking on the action on the side. So
[19:55] as you're developing, you don't have to run the whole pipeline and pay the cost of all of that computation. You can work on just one small part and iterate on that section of it. So I think you know I've released many features in my years at data bricks. This is the only feature
[20:08] that I have only gotten positive feedback on. This is super cool. You should check it out. Uh and of course I think Ali would be unhappy with me if I didn't mention agents at least once. And it is of course plugged in with Genie code. You
[20:21] can see we asked it to build us a medallion architecture and Genie has all of the tools to actually control this ID for data engineering. So if you are intimidated by SQL, you don't even need to know SQL. All you have to do is write
[20:34] English and it will generate the SQL and the pipelines for you all here inside of the IDE. It'll show you the changes that it makes along the way. It'll look up assets in Unity catalog to make sure that it's getting the schema correct.
[20:47] And I think one of the coolest parts is the agent actually uses the same tools that we would use as humans. So I talked about the pipeline's ability to validate itself and to understand the dependencies. It will actually call and ask the pipeline to validate itself so
[21:01] it can catch its own mistakes before coming back to you and having you look at the pipelines code. So this is I think super powerful. And I think more than 50% of pipelines are now built using Genie, which is pretty crazy given
[21:13] how recently this was released. Okay, so now let's uh start digging into some of the more advanced concepts. Uh how do you control this incremental compute? Some people look at it and say,
[21:25] whoa, whoa, whoa, that enzyme thing seems a little bit too much, too magical. Uh I never want a full recompute. That's too expensive for me. So in the last year, we added this thing called enzyme policies. So this is an
[21:37] extra clause you can add that controls how enzyme works. So by default we run in auto mode. The system will use its cost model to decide what to do. If you are one of those people that wants to be extremely strict and you never want it
[21:50] to full refresh. You would rather get an error than spend the money that it would cost to recmp compute that table, you can use incremental strict. And so now it will only run incrementally or it will fail if something has happened that prevents that from occurring. if the
[22:02] query is too complicated, if the table upstream gets deleted, anything that would cause that to occur, it will uh it will let you know about that. And then there are these other two that just nudge the cost model in one direction or
[22:14] the other. If at all possible, incrementalize, otherwise full refresh. That's incremental. Or you could say, I don't trust incrementalization and full refresh. But really, what I'll say is if you ever use one of these bottom two and it's faster, file a bug because you
[22:27] should never be smarter than the cost model. Occasionally the cost model has bugs but I would be very curious to hear anytime you are smarter than the optimizer because we should fix it so you don't have to do that anymore.
[22:39] Okay, so uh let's talk about another thing. This was actually added in just the last 6 months. This one's pretty cool. Replace wearflows. So this is probably familiar to anyone who has used Delta before. But sometimes updating old
[22:52] data is expensive and materialized views have this kind of powerful but uh complicated construct where they always give the perfectly correct answer. But that can be very expensive sometimes. So
[23:05] let's take for example this query here. It's pretty simple. Select star from orders join products on product ID. But then one of your product managers comes in and does something really annoying like they rename the product. Um, now I
[23:18] have to recmp compute 13 years of history and I don't really care about that. I don't want to pay all of that money. I It's fine if the old data is wrong. That's too expensive. I don't want to delete it, but I also don't want to take the time to update it. This is
[23:31] where replace wearflows come in. So with a replace flow and I switch to Python here, but you could do everything I'm showing you can do in both Python and SQL. Um the key thing is this up here at the top where column date uh is greater
[23:45] than or equal to subtract seven from the current date. So what this says is only update the last seven days. So this is exactly like materialized views where it's just going to run that query and
[23:57] update uh you know that part exactly correctly and in fact it's going to use enzyme under the covers. So this is strictly more efficient than if you were using raw replace in a notebook from Delta. um but it's only going to update
[24:11] the last seven days and it will update the le rest of the table will be left uh you know exactly as it was before. So this is a great way if you have a little bit more knowledge if you're willing to deal with some incorrectness in the past you could save tons of money with
[24:24] replace workflows. This gives you a lot of control over what you get with materialized views while still retaining some of that power of incrementalization. And the really cool thing is you can have this default predicate where like
[24:37] by default every time the job runs it's going to update the last seven days. But let's say you make a mistake and all of a sudden there's like three months of bad data. You can dynamically change this predicate. You can start a run of
[24:49] the job with some different predicate to do a backfill to do a correction to fix some problem that's happened upstream. If you change the code you can decide to you know how far you want that change to propagate. So this is an extremely
[25:02] powerful tool that really marries kind of the traditional batch way that people think about doing data engineering with the power of declarative incremental computation.
[25:14] So to summarize, materialized views are great because they're super easy to reason about. It's always exactly the correct answer. It supports all of SQL. Uh but it can require more computation. Sometimes computing the correct answer
[25:26] is expensive. Replace where is great because you only update the uh the the specified data. You can control exactly what that is. It supports all of SQL. So it's, you know, very flexible. But you have to decide which parts to update.
[25:39] You have to decide if that's okay for your business and and reason about that data that you're not updating whether or not that's okay as well. Okay, so that's a deep dive into materialized views, but that's actually just one type of
[25:51] incrementalization inside of declarative pipelines. The other type of incrementalization is structured streaming. And so let's talk a little bit about streaming tables. So a streaming table is just a delta table that is written to by a stream. And we
[26:04] have this idea of a flow that reads some data in, transforms it, and writes it to that target. So uh in the here's an example of of creating a streaming table and a flow. So I'll say create streaming
[26:17] table raw data. This is just a table. And then the flow is the thing that actually is going to read from Kafka, perform some transformation on it, and then write it to that table. Now, the magic here of a flow is this is
[26:30] not like a query where it's something that just runs once. This is a persistent object that lives inside the catalog. And what it's doing is it's actually tracking what data it has read before. So, it remembers in the case of Kafka all of the different offsets that
[26:43] you're reading from. And that way when you run it again, it won't read read any of the same data twice. There's also a shorthand. When we first created STP, we realized that the very common case is to
[26:56] create a table and a flow and a table and a flow. Most tables have only one flow writing to them. So there's this other syntax create streaming table as select. These are exactly equivalent. This is just shorthand. It saves you some typing. You don't have to say raw
[27:09] data three different times uh with the the construct over there on the left. but they're exactly the same. You can go back and forth between them. And I'll talk a little bit about why you might use the uh longer form in a moment.
[27:22] Now, streaming is great for ingestion. Pretty much anytime you are doing ingestion, streaming is the right way to do it because fundamentally ingestion is this read the new data appendon
[27:34] workload. So a really common use case of streaming tables is with this uh this data source called cloud files where it kind of incrementally streams data in from any of anywhere in the cloud. So this fairly simple thing will
[27:47] automatically keep track of which files have been read. It does that in Rox DB so it can handle hundreds of millions of files. It will also automatically register for notifications from the cloud so that it doesn't need to relist
[28:00] that directory over and over again to discover what data is new. uh and it does automatic schema inference and evolution. So it'll actually parse that data, figure out what columns exist, add them to the table if they don't exist,
[28:13] and you can control that with these things called schema hints if you want to have a little bit more power over what's going on. But this fairly simple line of code is actually doing quite a bit. And some of the largest workloads on data bricks use this autoloader
[28:25] technology. And like I said, you can also do it from Kafka. I mostly wanted to show this because if you've used structured streaming in uh notebooks before, I want to make it clear this is structured
[28:37] streaming. STP is not some separate system with a separate engine. It's just a better way to run structured streaming. You can use all of the spark readstream options, data sources, transformations exactly the same way.
[28:51] But you'll notice what you don't see. You don't see the checkpoints and you don't see the start and you don't see the await termination. all of the orchestration stuff that you normally have to do around streaming streaming query listeners and stuff all taken care
[29:03] of by the system. Uh, but you can use any data source that's included in DVR and often you would use these message buses when you want the lowest latency. I'll talk about some exciting developments there in just a little bit.
[29:16] Uh, streaming tables, one thing you need to wrap your head around is they process only new data. So each input row is processed only once, even if the query changes. So I have this kind of, you know, madeup query here, but really all
[29:29] it's doing is taking the input and adding one. So one becomes two, two becomes three. But if I go in and I edit that query after it's run before, and I change it to be times two instead of plus one, it doesn't change any of the
[29:41] data that has already been processed, that's already been read. We're not going to go back in time and read it. This is one of the things that makes streaming so performant is it guarantees that it will never go back and revise results that were already processed. uh
[29:53] but when new data comes in it will use that new definition and so you can see three becomes six four becomes eight it's using the new query definition only for the new data but this is you can see this is a little bit more complicated than materialized views materialized
[30:06] views you get exactly the answer you want streaming you have to think about all of the different queries that have been there before and how they've evolved over time but much faster much more performant if you do want to re reprocess that data
[30:20] structured schema makes this easy I've seen so many notebooks where you have commented out like rm RF the checkpoint directory and truncate the table. It makes it extremely annoying to do uh debugging and testing. In STP this is
[30:34] built in. This is kind of a first class feature. Uh we call it a full refresh. And so you just click this button in the UI full refresh or you run a refresh name of the table full and it will automatically take care of all that. It
[30:46] clears out the table. It clears out the checkpoint. it reruns the steam from scratch processing all of the data that's available from the beginning of time to give you the updated answer. So it's kind of nice. This is another thing where declarative programming takes care
[30:58] of some of these common problems that you run into. Uh this is the reason why you would want to create flows by hand rather than using that shortened syntax that I showed. It's often the case that you
[31:10] have multiple things that are writing to a table and that they actually evolve over time. So in this example, I'm reading from three different Kafka instances. Maybe the business adds and removes Kafka instances over time as the business grows or shrinks or we change
[31:23] regions. Uh and so you can actually create multiple named flows that write to this table. You can add new ones. You can delete ones that exist and the system will automatically manage what's going on there. So this is why we have both of them. You can use the kind of
[31:36] expanded syntax when you want to do something complicated. You can use a shorthand syntax when you just have the simple case of one table, one flow. So everything is great in declarative
[31:48] land, but sometimes things go wrong, sometimes you make a mistake, sometimes bad data comes through. So I want to talk a little bit about these one-off operations. So streaming tables are tables. There's, you know, there's nothing special about them really. You
[32:00] can use standard DML to perform these ad hoc operations. You can insert, update, delete, and merge in them. Uh while they're managed by LakeFlow, all of these commands work on them. Uh, but I'm going to talk about in a little bit. You
[32:12] do need to think about structured streaming and any downstream things as you would with any delta table that has changes going into it. But this is something we've improved on quite a bit in the last year. If you've been using
[32:24] STP for three or four years, you might cringe a little bit when I say streaming tables because they were missing a lot of the cool features that made Unity Catalog powerful. We have almost completely fixed all of those. So DML
[32:36] and DDL alter commands, comments, rowle security, column level masking, uh getting a change data feed, using delta sharing, all of these things work on streaming tables. And by the end of the
[32:48] year, streaming tables and tables will be exactly the same thing, including the ability to take an existing table and use it inside of an STP pipeline without rewriting all of that data. So I'm super excited about that. Even though it's a
[33:00] little, you know, in the weeds, under the covers, I think this is going to dramatically simplify the process of working with STP. Um, one way to handle these one-off operations, like let's say something like a backfill, is you can actually
[33:12] make them declarative. You can put them into the pipeline. Some people don't want to go and run an insert in their production environment because they might fat finger it and make a mistake. And so, a common thing you might want to do is actually encode that into your
[33:24] pipeline. Well, that's possible with this thing called a once flow. Normally flows run every single time the pipeline runs. With a once flow, it runs once, it completes, and then it never runs again.
[33:36] It's perfect for something like backfill. So you can see in this example, I have some historical data in parquet. I'm going to read it once, write it to the table, and then never read it again. And then I have new data that is coming in from Kafka, and I will
[33:48] continually read that and append it to the table every single time the pipeline runs. This is great not only because it kind of follows these nice practices of having everything in code so that you're not going to make mistakes in production, but also if I ever choose to
[34:00] full refresh this table, well then that once flow is going to run again. So it it kind of takes care of all of those setup operations automatically so that there's no manual steps in your data pipeline.
[34:12] If you want to uh reset a table without doing a full refresh, that's actually also possible. So sometimes you might take a change like you know I have this this uh this streaming query right here. It's reading from table one. If I change
[34:26] that to read from table two instead normally that will break structured streaming because we'll look at it and we'll say this checkpoint is for table one. It makes no sense to use this checkpoint with table two. Uh what do
[34:39] you want to do? So the system will will break. It'll tell you that there's this error that the table has changed. This allows you to reset the checkpoint without clearing the data out from the table. So I would run this operation
[34:51] reset checkpoint selected which I believe is available in the UI now. It'll create a new checkpoint. And you can see I've changed the code here. I added this option starting version latest. So when it's reading from table 2, it's not going to read all the
[35:04] historical data from table 2. It's only going to read the new data that moves forward. So this is a great example of evolution where I'm moving from one table to another. I don't want to lose the data that was processed before, but I want to pick up right at this, you
[35:17] know, in the with this table number two, starting at the point where I first run this stream. Okay, now let's talk about joins because joins are kind of one of the most mindbendy parts of streaming. Uh,
[35:30] streaming joins uh operating against they they basically operate against a snapshot is the right way to think about it. So that snapshot is updated automatically each microbatch. So if I'm doing a a fact dimension join, I have
[35:42] some other table. So you know I'm reading from cloud files which is a stream. I'm joining with this dimension table cities on the city ID. Uh that snapshot is updated every single microbatch. So as the stream is running,
[35:54] it's continually picking up new data. But as I said before, streams do not go back and reprocess data that's already been processed. So let's say I come in here and I correct this spelling mistake. So now Berkeley is spelled correctly. that data that was already
[36:07] processed will continue to have the old uh spelling there. Only the new data that comes in will have the correct spelling. Again, you can use a full refresh to fix this. So, this is the default way that joins work and
[36:19] structured streaming. Again, it's great because it's super efficient. It's like a little bit harder to reason about the materialized views because a materialized view would make sure that it was always up to date no matter what happens.
[36:32] There's this other construct called stream joins. This is honestly one of the most confusing features in all of Spark. You probably shouldn't be using stream joins. They they have an extremely niche when when you have the
[36:45] right use case, they're a great tool, but the right use case is extremely narrow. Streamst stream joins are for fact joins where there is temporal locality and arrival. So, for example,
[36:58] the the case why you would use this is if you're doing ad tech and you have clicks where people are clicking on ads and impressions where ads are being shown. Those always happen nearly the same time, but then they're emitted into
[37:10] separate systems separate from each other and I want to join them up. That's where you would use a stream join. But it's extremely important that you have that timebound on their arrival. Uh otherwise it is extremely inefficient.
[37:23] So if I you could see here again this code is pretty complicated. uh but I'm you know joining on ad id equals impression ID. I have this time bound where I say they arrive within five minutes of each other. I have watermarks
[37:35] on both stream to say how how late data could be arriving. And if I do not have all of these things, the watermarks on both streams, the equality in the join and the time bound, this will buffer all
[37:47] data forever. And that's probably not what you want to do. This is not for updating dimensions. If you want to do updating dimensions, you should use materialized views replace these other constructs that do handle updating dimensions. Uh so I just wanted to cover
[38:00] that probably don't use it. Okay, so I've been hinting around this a little bit, but I want to talk a little bit about changes in upstream delta tables. So first of all, why is it hard?
[38:12] Why do I care about changes? Uh, it seems like a pretty easy thing, but you have to think about this efficiency guarantee that we make with streaming that we don't need to store all of the input forever. That each new row requires a constant amount of
[38:25] processing. And let me give you an example that seems simple, but it's actually pretty difficult. If I'm running a streaming query, select the max number from numbers, and I do a bunch of operations. So, I insert a 100, I insert one, I insert two, but then I
[38:39] delete a 100. Well, what's the new answer? The answer is from this. It's impossible to tell what the answer is because it's not a hundred, but it I don't know if it's one or two because I
[38:51] don't know all of the data that arrived in all of time. And so the the problem here is these non-monotonic queries in the face of retractions can require arbitrary amounts of storage. And so
[39:03] this is a very complicated way to say that streaming makes this simple by just disallowing it. But what that means is this is there's this very common pitfall when you're using streaming pipelines
[39:16] where you're reading from a delta table and you perform an update and then the stream doesn't know what to do because it has not been keeping the state that it needs in order to compute this result. So if you do an update, delete or merge on any input table, doesn't
[39:29] have to be a streaming table, it's just any delta table that you're reading from. If you uh stream from an aggregation that doesn't have a watermark, so it's actually updating those values instead of outputting new ones. If you're using auto CDC where it's changing that table, you cannot
[39:42] stream from any of those tables because they have updates instead of just depends. There are a couple of workarounds. Uh option number one is the easiest. Just don't have changes. Just make it append only. So this is honestly
[39:55] where most people should be. Uh streaming is great for bronze ingestion where you're doing simple transformations. Like I said, you should always use it for ingestion because ingestion is always this appendon game.
[40:07] Uh and you can enforce this by setting this delta table property on all of your input tables. Say append only equals true. It will prevent these streams from ever breaking because it will not let you delete data in those tables.
[40:20] Option number two is to handle those changes manually. So we have this option called skip change commits which tells the stream don't worry about it. I'm going to handle all of the changes myself out of band. So you set this on
[40:33] the data stream reader. Now, new data will flow through automatically with streaming, but when a delete comes in, it just gets ignored and then you have to manually go to the downstream silver
[40:45] table and issue that same delete. So, this is great if you're doing like manual GDPR. You can kind of build processes that take care of this. Uh, and then option three is reading a
[40:57] change feed. So, this is the most complicated but also the most powerful. In this case, instead of saying give me all of the new data from this table, you're saying give me an appendon stream of the changes. And now you have to
[41:11] think about how to process changes. So in this example, you can see I'm using this case when statement to actually say so I'm here I'm computing a sum and so uh you know when there's a new value
[41:24] when it's inserted you take the value when something is deleted you subtract the value. And similarly, when something has changed, you delete the old value and insert the new value. And so this gives you kind of the computation that you would do to process these
[41:37] incremental changes. So you can do kind of arbitrary custom logic that thinks about how to handle inserts, deletes, and updates. Like I said, this is the most complicated way, but is also the most powerful if you have specific business logic that you want to
[41:49] implement for those changes. And then finally, option four, this is my favorite, just use MVs. MVs take care of this automatically. MVs or replace where automatically handle changes,
[42:01] automatically handling handle changing dimensions. You don't have to think about any of this stuff with watermarks or state or uh you know uh retraction all handled automatically.
[42:13] So to summarize what we've been through so far, uh we've got these three different incremental strategies and I like to think about these as a spectrum. At one far end, we have streaming, which maximizes performance, uh, minimizes
[42:25] latency, super low latency and linear compute, but it has limited expressibility and it doesn't support these updates and there's just a lot of cognitive overhead to dealing with some of these limitations. In the middle, you have replace where where it's just plain
[42:38] SQL. It supports everything you want to do. Uh, but you have to decide which parts of the table to update. And at the far other end, you have the absolute simplest materialized views. super easy to reason about, as easy as writing SQL,
[42:51] supports all of the SQL language, uh, but can occasionally it's just fundamentally expensive to execute certain queries. So they can be a little bit more expensive, but that expense is somewhat fundamental if you want to get
[43:03] the correct answer. Uh, so another really cool trick with streaming is you can write to other systems. This is another thing that we've rolled out in the last year. So it's not just delta tables, but it's any system that DBR can write to. So you can
[43:16] now stream to any sync with STP. So in this example, I'm creating a Kafka sync and then I have an append flow that writes to that sync. So I'm basically replicating this table as new data arrives into Kafka. It's as easy as
[43:29] these few lines of code. And again, you'll see this looks a lot like structured streaming, but we're taking care of the checkpoints and the dependencies and the retries, the the cluster and all that stuff for you automatically. And this is where it gets really cool.
[43:43] This is an announcement that we made last year at Summit that is now fully supported inside of STP. Um, real-time mode. So, for the longest time, uh, structured streaming was great for
[43:55] streaming in ETL where you cared about seconds to minutes to hours to days. And in fact, it's really days and stuff where we are uniquely capable because no other streaming engine can turn off and turn on as efficiently as we can. So
[44:07] when you have stuff where you really don't care about latency of, you know, less than an hour or a day, you can turn that cluster off in the middle. Um, but some use cases like fraud detection or uh sensor monitoring require like actual
[44:21] millisecond latency. That's where real-time mode kicks in. The greatest part is there's nothing new to learn. It's all of the APIs you already know with structured streaming and STP. You just flip on real time mode and it gets
[44:33] dramatically faster. And it is very fast. Here is a comparison that we did with Flink. Uh, for really simple stateless transformation stuff, we're on par, 3 milliseconds, you know, pretty pretty fast. For more complicated
[44:46] things, it can actually even be dramatically faster. So, this is now, don't listen to the FUD that Spark isn't a true streaming system. It is true streaming running all the time, extremely low latency, you know,
[44:58] singledigit milliseconds for those business critical applications where latency is of utmost importance. Okay, now I want to uh bring up Jose who's going to do a demo. This is
[45:10] actually I think one of the coolest features that we've rolled out in the last year. Is Jose here? Okay,
[45:22] is for the first couple of years of working on STP, it was all about catching up with batch jobs. You can come up and start getting set up. Uh it was all about catching up with batch jobs and just trying to make the STP
[45:34] language expressive enough so that you could uh do all of the things that you could do with Turing complete code inside of a notebook which as you can imagine was a a pretty big challenge. Is it not showing? >> Yeah,
[45:46] >> we need to uh my slack. >> There we go. >> Uh so for the longest time we were playing catchup. This is, I think, one of the first features where we truly
[45:58] leaprog and do something that just wouldn't have been possible inside of a notebook. So, without further ado, yeah. All right. So, I I got the I got an alert just before this. It told me I'm
[46:10] getting a lot of expectation failures in this pipeline that I have. So, I pop into the nice little nice little graph here and I can see clicking on the warnings. Okay, so something's something's up with my reliability
[46:22] weights. I have a validity check is just triggering on every single one of them that I'm generating. Based on my based on my goal, it looks like that's been happening for a while. So, let me get let me go to the editor and see what's
[46:34] up with that. All right. So, yeah, we have our nice editor that Michael showed off earlier here. Uh there's there's kind of a lot
[46:46] of code here. I don't really want to go through it line by line. So, let me pull up pull up our version history and see if there's any changes that seem like they might be responsible. So, if I go here, ah, okay, I see that I
[47:01] see the problem now. I have in I've entered the I've I was tweaking the reliability weight calculation and I've made a pretty clear error here. Obviously, I should be subtracting the battery percentage from 100 and not from
[47:13] 10. That that doesn't quite that doesn't quite make sense. So, what what am I going to do now? Historic historically the answer is okay. I have to fix it. I have to restart the pipeline and I have
[47:25] to figure out how to go through all of my gold and silver. I have to manually get the bad data out and figure out how to back fill it from the whatever the appropriate bronze range is. And even calculating that could be really hard.
[47:38] What would be really convenient is if I could just go back a few hours and not have made this error. That would have been very convenient. So I'm I'm going to I'm going to demonstrate the process of doing that. First uh we go we have
[47:51] the version history here and we should be able to go back to the previous version. Restore this version.
[48:03] Right. Let me double it. Double check that worked. The the calculation seems to be correct. All right. So now we're going to go to this rewind pipeline menu option. I should say here that this is one of the forward-looking guidance things. Don't go looking for this button
[48:16] in the UI today. But yeah, so so we see here a list of rewind points. Basically, every hour in the every hour in the background, we were taking a metadata snapshot of where
[48:28] are all of the streams and all of the tables in my the in my pipeline app. So I can open up the n the 940 one here. This is GMT. Uh check my silvermetrics
[48:40] table. loading the Huh. Okay. The the demo laptop does not have permissions to load the to load the preview
[48:54] this in this rewind point, this was after I made the ch the change. So all of the reliability weights will not be in the in the proper zero zero to one to one range. If we go back to this one at
[49:07] 7 to at 720 that was the that was when the bad the bad code change was made. So if I can get back before that every the everything should work. So I'm going to
[49:19] go through the go through this here. I'm going to deselect the this heartbeats table. This has nothing to do with the exotation failure. So I don't need to bother paying the cost of re rewinding that. Right. Confirm that everything
[49:32] looks good. And then while that while that rewind is running, let's talk a bit more about what I just did there. All right. So error the error recover the error recovery is something where in
[49:44] the in batch land it's kind of a lot easier to how do I presume? >> If you want there we go. All right. So the error recovery is
[49:56] something that's kind of a lot a lot easier in batch land land often because if your process just looks like every day I'm doing this thing with this day's data and then the next day I'm doing the same stuff with the next day's data you
[50:10] can kind of just rerun the day you right and there there you know you know that does make some things simpler but obviously there are a lot of streaming advantages that you that you can't get with that you know you can't get you
[50:23] can't get even moderately low latencies Certainly you're never going to get anything like real-time mode in in batch. You know the process the processing is guaranteed to be to be incremental but because of the because of this you
[50:36] know as a consequence of doing this kind of streaming workload there's a lot of there's a lot of internal state that's tracked as part of it as part of that. Uh yeah, so you you have the data in the
[50:49] you have the data in the data in the data sets which of course you know if you've generated bad data you have to go that you have to go take out that take out that bad data you have the streaming offsets which are how far in the source
[51:01] data have it have you read you know you're keeping track of that and you have to be able to go back and say you know I I need to reingest this and you have it and you have this the streaming
[51:14] operator date which is an entire separate out of the out ofband buffer that tracks like you know you know what rows have I seen duplicates of you know what is the what is the sum of my current of my current window and all of
[51:27] the all of these are somewhat hard to man to manage some of them are really not reasonably possible to today and in practice what what undoing these kind of
[51:39] issues in streaming looks like can be a lot more a lot more of you know let me do some manual ual corrections and you know I'll do as good I'll do as well as I can there will probably be some things that go wrong and I'm just going to
[51:51] accept that as part of the state of the world so need to go back sorry still figuring out how to work this laptop all right so looks like the
[52:04] rewind has completed here so let me run the run the next batch while I go through the last slide here all Right. So what we what we have here
[52:16] is what we're what we're calling stream the day streaming time travel. What this system does does is periodically every the every hour like I mentioned earlier we generate that we generate state snapshots that capture what is the
[52:29] current progress of all of the streams and everything in the everything in the streaming engine everything in everything in your pipeline. uh when the when something goes wrong and you say, "Oh, you know, I really wish that I
[52:42] really wish that I hadn't made that change at 4 at 4 PM," you can go that you can go back to that to that snapshot. You know, you you select the rewind point that you want that you want to go to. You know, you did you do your
[52:55] data previews and so and such to make sure, you know, everything looks good. This is as far back as I need to go. And then you just perform the perform the rewind under the under the hood. We do
[53:07] all of the we do all of the metadata tracking. And this is something where pipelines really really helps us here because having the having your entire data flow graph be a core concept really
[53:19] lets us do the full traversal and understand what are all of the pieces we need to rewind and coordinate with it with each other. And once we've once we've done that, as far as the pipeline is concerned, you are back at where the
[53:32] rewind point is. You can just push it push run again as we as we did and then and it will automat automatically reload it reload everything that uh
[53:45] yeah everything that everything that it that you wanted to ingest. So you can see in the table metrics here. Yeah. So you can see in the table in the table metrics here that this so that this run
[53:58] was obviously more a bit more more volume than the previous ones because the because we were doing this replay. You know we we loaded not just the in the new incoming batch but all of the
[54:11] ones between when we reround and when and the the current time and yeah now now everything is perfect. Everything is perfectly recovered. The expectation failures are all gone. There's nothing
[54:23] wrong with my calculations. You know, I can just restart restart my schedule and get back to work. Cool.
[54:39] I will say I'm extremely excited about this because, you know, of course there's nothing here that you couldn't do by hand, but what's going on behind the covers is extremely complicated. And so every once in a while, an extremely large data bricks customer would get
[54:51] into one of these problems. And of course, then it would become our problem to go in and hand edit their checkpoints and do other things. And I'll be honest, I'm afraid to do it. There's a kid, Ryan Zoo, who I call to do it, who's like literally the best Spark debugger in the
[55:04] world. And I am glad that we no longer have to wake him up at 3:00 a.m. It's now just a feature built into the product, the undo button to fix these mistakes that happened in the past, either because you broke the code or
[55:17] because someone upstream from you sent you garbage data. I think it's pretty cool. Okay, so now let's talk about some of the other kind of uh you know, I'd say higher level flows that are built into STP. So far we've been talking
[55:29] about things that look like queries and things that uh look like you know just appending data into a table. CDC is actually kind of a whole application bundled into a declarative construct.
[55:54] getting it wrong. So what is CDC change data capture? It's actually a really simple problem to describe. I have a stream of updates and I want to take the latest update for every key. So in this
[56:06] case, you know, I see two keys, key one and key 2, and I want to take the latest records for each of those and produce this new table that is a snapshot of this other system at some point in time. So seems pretty simple, but it becomes
[56:20] extremely complicated when there is out of order data in a streaming system. And so let's walk through why that is. So any of you who have used structured streaming in 4 batch, you've probably even handcoded this before. And I
[56:32] regularly see customers with code that looks exactly like this. So it looks pretty simple. I'm going to read a stream of CDC data. I'm going to use merge. I'm going to match on the ID and then I'm going to either update or
[56:44] insert based on whether or not that record is already in the table. 14 lines of code. Okay, Michael, why should we be excited about this? This is really not that hard. This code is extremely incorrect. uh there are several things that are
[56:57] wrong with it. The first thing that's almost certainly going to happen to you is there will be multiple updates to the same key and a single microbatch. So with streaming the microbatch boundaries are arbitrary. The system picks them based on how much data there is and how
[57:09] big the cluster is. And so this is bound to happen at some point and merge doesn't know your business logic of which one is most important, which one should win. And so it's just going to throw an exception and say that there are multiple matches and refuse to
[57:22] operate. So you'll go in and you'll fix that uh and you'll add this extra aggregation. So you'll do a group by key. You'll take max by some timestamp. You'll take all of the values for that row and you'll explode them back out. So
[57:35] now instead of two rows in that microbatch, you'll have the correct single row, the one that occurred latest in time. We can add that back into our code. And now we're at 18 lines of code. But guess what? This is still wrong.
[57:48] There are several mistakes with it still. Uh, another one is those outof order updates aren't actually always in the same microbatch. They can actually get split across micro batches. And so now even though we picked the latest,
[58:01] you're still going to get the wrong answer because these have arrived out of order. And so now your table will actually go back in time, all of your downstream consumers will be confused why the data is stale. And it may never update again if you don't get another
[58:14] update for this key. So this is subtle because it runs but it runs and it returns the wrong answer. So how do we fix this? Uh we fix this by storing a version inside of the table so that we
[58:26] can remember the last logical time that this row was updated at. So we'll store a timestamp uh for each row when it was last updated. We'll write out these expressions for every single column that
[58:38] say only update this column if the row that is coming in is newer than the one that's already stored there. We can go ahead and add that into our codebase. It's a little complicated because we have to spell it out for every column. We're up to 22 plus lines of code
[58:51] depending on how long your schema is. You might be doing some meta programming to construct this. Now, uh, but it's actually still not correct because we need to handle deletes as well. Fortunately, that's not too hard. We'll
[59:03] add in the when matched and delete. So now when records come in with a special delete flag, we will recognize that flag and remove them from the table instead of just setting that flag in the table. Uh, so now we're at 23 lines of code,
[59:16] but there is still another bug here. And this one is subtle. This is the one that I'm not sure I've ever actually seen customer code get correct. Uh, the problem is there's another type of reordering that can occur. You can have
[59:28] a delete and an update that are reordered with each other. And deletes are special because they actually delete that row from the table. So you forget that version that you stored. And this means we end up with a zombie. A row
[59:41] that lives in the table even though it should be dead and it is very difficult to kill. So in order to solve the zombie problem, we actually have to create tombstones. These like memories that
[59:53] this row has died. Uh and so you know basically instead of deleting the row right away, we're going to change it to be soft deletes. Uh so you know, we'll just mark that row as deleted with a time stamp from when it when it was
[60:05] deleted. We don't want our users to see these tombstones. That's almost as bad as seeing the zombies. They'll get scared. And so we need to redirect them to a view that filters out all of the tombstones. And of course, we don't want these tombstones to accumulate forever
[60:18] because we're paying a bunch of money for them. So we have to build a cleanup process that removes those tombstones over time. And so now to summarize the correct code, which you're welcome to copy and code if you think that's the right way to work, uh it you now have to
[60:31] redirect these users to a view. You have to run this cleanup periodically. Uh, and we have 24 plus lines of code from something that seemed pretty simple. And along the way, we made a bunch of mistakes that made all of our downstream consumers angry. Uh, so you know, it's
[60:45] pretty complicated. It's a lot of infrastructure to manage for such a simple workflow. So, there has to be an easier way. And I bet you can guess what it is. It's declarative pipelines. So, we have this
[60:57] really cool command called autocdc. And it's one SQL command, five lines of code, and it takes care of all of those problems that you talked about. And as you can see, all we're talking about is
[61:09] what should happen rather than how it's going to happen. So, uh, you know, we say auto CDC into this table from this source data. You tell it what the key is. You tell it what your time stamp is. The time stamp can be multiple columns
[61:22] if you want. If you have like transactions and you need some sort of a tiebreaking record inside of it, you could just create a strct and have multiple columns there. Uh you can tell us if you want to delete and what flag indicates a deletion and it takes care
[61:36] of all of that stuff for you automatically under the covers. And so uh you know now uh another problem that you might have on top of this you're because every single database is a little bit different and
[61:48] some databases do this kind of annoying thing where when they send you change records they only send you the columns that have changed. They don't send you all the other columns. So in this case you know I've got the the name changed
[62:00] and the city changed those came in two different records. The correct thing to do is to ignore those nulls and combine them into a single record. Uh, handling that with AutoC is super simple. All you do is tell it declaratively what to do.
[62:13] Ignore null updates. I don't have time to go into it, but of course implementing this is actually quite hard. That version that we talked about before can no longer be a single version. You now actually have to implement what's called a vector clock
[62:26] where you have versions for every individual column at some point in time and you have to compare with the correct part of the vector clock for each column. So, it ends up getting quite complicated. And then finally, there's this other variant of CDC that many
[62:39] people care about called SEDD2. Uh so SEDD1 means update in place. SEDD2 means keep a history of each value and how it has changed over time. SECD2 is too complicated to talk about even in a
[62:52] two-hour talk. Uh, but the important part here is the implementation of AutoCDC is over 2,000 lines of code and even more importantly, it has 10,000
[63:04] lines of tests to make sure that it's correct. And you get all of that power with those five lines of SQL. And this is just another illustration of why declarative programming is so powerful. When it has the application that you
[63:16] want, you can tell it what to do. You don't have to worry about all of those complex distributed systems details. And we just released I think this is in private preview right now. I think this is very popular in the finance industry.
[63:28] There's another version of CDC called by temporal CDC where you actually remember not only the actual time of the record but the time at which it came into your system. So we have this business time and the system time. All you have to do
[63:41] is say store it as by temporal and it will automatically implement that process for you as well. So if you're in a regulated industry, I think this is often very powerful because it lets you go back in time and not just say what was the data supposed to be at this
[63:54] time, but what was the data actually at this time if I created any reports or anything based on it? Cool. Okay, so a couple more things. Uh expectations. We've talked a lot about
[64:06] data processing, but how do I know if my data processing is correct? Jose hinted at this a little bit, but we have this thing called expectations. Expectations are like constraints, but they're constraints with a policy. So you can tell the system what to do when they're
[64:18] violated. It turns out in a big data system, if you abort every time something looks weird, you probably won't do any processing. And so expectations give you a little bit more power here. So they're true false expressions that are used to validate
[64:31] each individual row. And we have these flexible policies. You can say just track how many are wrong. That's what Jose was using where it was just showing him, hey, thousands of records look bad, but the processing continued. Uh, you
[64:44] can choose to drop those records, but this is much better than a filter. You could always filter out bad data, but how do you protect yourself from the case where all of a sudden you're filtering all the data or most of the data? It's extremely confusing to those
[64:56] downstream consumers. Everything is just missing. This gives you a big red warning bell that says, "I'm dropping a lot of data and maybe you need to look into that and maybe you need to use streaming time travel to go back and undrop it and fix it." Uh, and then
[65:09] finally, if you do have something where it's really a business problem for anything to go wrong, you can use an abort expectation, which is basically a constraint where it says, "If this is ever violated, abort processing. Stop
[65:23] the job. Let me come in and figure it out." But even this is a little bit better than a constraint because it'll actually rewrite the query plan to include providence so that you can understand what was the input row that
[65:35] caused the bad output row. It's often very confusing if you're doing something like parsing JSON and then you have an expectation that is something like you know this parsed result is not null. Well, getting null doesn't tell you
[65:47] anything. You want to see that broken string from the input that failed to parse and expectations can automatically take care of that for you. Now expectations I think often people don't understand just how powerful they
[66:00] are and when you start layering them with materialized views they you get the whole power of SQL. So for example let's say you wanted to express something like a primary key constraint is unique. You can say well I can't do that with just a
[66:13] single row but by creating a materialized view with an aggregation you can. So I'm going to create this materialized view whose only job is to count how many times I see each key in the input. This will automatically be incrementally maintained. So it's fairly
[66:26] efficient and if any value appears more than once, this will fire an expectation and it can even tell you which record has been duplicated. So this is pretty powerful. Uh and another example that I
[66:38] often see people get confused by is you know you're doing a join and what happens when rows are missing from the dimension? You just end up with missing data or nulls in your data and you don't
[66:50] really want either of those. So with expectations, you can kind of get the best of both worlds. The key trick here is we're going to change the join to a left outer join so that I get all the
[67:02] rows even if the corresponding row is missing. For those of you who uh you know need a quick refresher on relational algebra, an outer join is one where you keep rows even when they don't match. But rather than just confusing all of my downstream consumers with
[67:15] this, I can put this expectation here that says to, you know, either alert or drop or do something with those rows. So I just know how many are missing rather than them all flowing downstream with all of these extra nles in it. So I
[67:27] think when you combine outer joins with expectations, you end up with this very powerful data tool for enforcing foreign key constraints as well. Uh so let's get into a few more advanced things. uh because you know even with
[67:40] declarative pipelines there's still layers on top. You need to think about deployment you think need to think about staging and production. We have this really cool system called dabs. Uh dabs automate the creation of these resources
[67:52] so you can automatically sync. And so here's an example where I version control everything including my deployment. Uh I can automate the creation of these multiple environments. So I have one set of code in git that
[68:04] does this you know raw clean and scored. Uh, but then I do checkouts into each individual environment. I have my own version in Michael Dev where I work on it by myself. I have a version in staging that runs on a larger amount of
[68:17] data. And then I have the production pipeline that's actually generating reports that people care about, but they're all sharing the same codebase. And I'm running in these lower environments before I run in the higher environments in order to get that validation. DABS automates the creation
[68:30] of all of this. And so there's this pretty cool trick you can do where, you know, I create these different environments inside of the DAB. They point to different targets so that I'm not overwriting the tables in the other environments, but the rest of the
[68:42] configuration is automatically shared by the system. So I end up with these isolated copies for each environment. Uh, and the the CI/CD pattern that is unique with oh, and I forgot to update this DLT STP is uh separate you use
[68:57] separate directories for your sources and for your transformations. The sources are the things that read from external input. They're the things that I want to mock out in lower environments because I may not want to process the full data set every single time in
[69:09] development. I might want to work with a smaller version of that data set. So you can create this sources directory with a dev version and a and a prod version that are different. And then the transformations are the same. They do the real work. This is the actual code
[69:22] that you're trying to do the testing on. And then use expectations in both of them for validation. So you're checking in dev whether your code is correct. And then in production, you're making sure there are no unexpected things that you you uh that might break that you weren't
[69:34] expecting in dev. And so what you end up with is you end up with like so here in this example, my input is sales. And I have a prod version of sales that actually just reads from the sales table. And this is
[69:46] a view. So it's it doesn't cost me anything to create this extra node in the dataf flow graph. But in dev, I don't want to read the entire sales table. That's very expensive. I'm just going to read the last day of data. So
[69:58] now in dev it runs really fast on one day of data. In prod it runs over the entire data set. And in both cases I only have to write this query out once. And the magic of this is of course something you could do with notebooks.
[70:10] But the magic of declarative pipelines is it stitches all of this together automatically for you and gives you that nice data flow graph for navigation and error checking along the way. You can validate the production pipeline before you deploy it. Uh and so I think really
[70:23] kind of all these technologies work together uh very nicely. So there's a bunch of different ways you can create this test data. You can use a subset of the production data like I showed in the previous slide. You can hardcode in
[70:35] specific pathological values that might break your pipeline. You can read from some test file that has data again that might you might choose or you can even randomly generate data in dev so that you can test different scales and see
[70:48] how long it takes to compute things like that. So I think these are all great tricks and you'll notice in all these cases I'm just creating this view. So these are free nodes in the graph. They cost nothing to create views. And so all it's doing is making your pipeline more
[71:00] readable and giving you this place where you can interpose and substitute in different data based on the environment. And so you can see here's the the DAB version of this where the
[71:12] transformations are shared by all of them, but for the different production and development environments, I actually include a couple of extra sources. There's another really cool feature that we added in the last year, which is this ability to include entire directories.
[71:24] So you don't, you see that star? You don't have to go in and add each file individually as you add new sources. The star just automatically includes all of them. Just some nice developer ergonomic stuff.
[71:36] Now, here's a preview. You might say, "Okay, that's great, but that sounds like integration testing. I want unit testing." And this has been a huge gap in Spark since the beginning of time. We are previewing now unit tests for
[71:48] pipelines. And so I think this is pretty cool. You can have your pipeline code and then you can actually go and mock individual data sets run subsections of the graph and assert on the results. So
[72:01] this test, you know, puts in mock data into the system and then asserts that I get an expected number of rows out as a result. So I think this is pretty cool. Again, like this is I said this is in preview and we will of course build this
[72:15] into the ID for data engineering. So you'll be able to when you have these tests run individual ones. You'll be able to run the whole thing.
[72:29] No, I'm extremely excited about this. I think you know over time we're going to expand it quite a bit. So you can even do advanced things like include mult multiple micro batches so that you can test streaming and how state changes over time. Uh so stay tuned to this area. talk to your data bricks account representative to get added to the
[72:42] preview. Uh and I do believe we'll also be open sourcing all of this in Apache Spark. Okay, a few more advanced things before we close out. Metar programming is one of the most powerful ways to build big
[72:55] complicated pipelines. And so metaroming is when you actually write a program that generates the graph instead of spelling out each individual data set by hand. This is extremely popular in ingestion where it's like I have 10,000
[73:09] Kafka topics and I need to create 10,000 tables from each topic. Anytime you have a big list of of things you want to do and they're kind of similar transformations, this is a great pattern. I've seen this used oftent
[73:21] times where there's a small data engineering team that needs to enable business teams. So the important thing to realize here is the input to this for loop can be arbitrary data. So I've seen places where they create a spreadsheet
[73:34] and this for loop reads from that spreadsheet and then business users who want to ingest a new data set just go and add a line to the spreadsheet or there's a table in lakebase or something like that. So this is an extremely powerful primitive and so uh let me walk
[73:48] through what's going on here. So I've got this list of regions. I say for each region create a table and then I just have standard STP code and you can see that that it's being called inside this function. You can integrate with schema
[74:01] registries or spreadsheets or other things outside to power what this is. And there is one gotcha here. It's a Python thing, but it confuses almost everyone. So I will talk about it a little bit. If you are using this pattern, you will notice I do not run
[74:13] the DT table code inside of the for loop. I create a function and then call that function inside of the for loop. There's some weird rules about how variable capture works for functions in
[74:26] Python. If this was inlined inside of the for loop, every single table would get the last iteration of the for loop. So they would all have the same region in them. It's because you actually you
[74:38] don't capture the value of that variable. You capture the variable itself and then it changes over time. So easy way to get around that, just create a function. If you didn't understand what I just said, just create a function. Um, but it's if you do it this
[74:53] way, it will work and it's it's actually super powerful. Okay, the final section, declarative escape hatches. So, I'm a huge fan of declarative programming, in case you hadn't figured that out. Uh, but I
[75:06] really think declarative programming is an 8020 game. I my goal is to get 80% of the ETL that happens on data bricks to happen inside of STP. I don't think we're going to replace Spark. I don't
[75:18] think we're going to replace notebooks or CC code or any of these other things. And I think there will always be that 20 to 10% of stuff that is extremely business-specific, that is complicated, that doesn't fit into the framework. And
[75:30] that's why we have a bunch of declarative escape hatches. You can do the things that fit into STP inside of STP and not worry about all of the extra complexities, but because data bricks is a unified analytics platform, you can
[75:43] also do all of the other stuff there as well. So, uh, the biggest escape hatch is LakeFlow jobs. So again I often get this question of should I use jobs or should I use pipelines. I hate that
[75:55] question. It is the wrong question. You should use both. And jobs is so powerful because you can run anything on data bricks and as all talked about yesterday even things outside of data bricks. And so it is great when you do a little bit
[76:07] of work and then do lots of data transformations in pipeline and then do a little bit of work with a notebook or you know with a wheel or with a dashboard or something that doesn't fit into the pipeline. jobs is the right way to orchestrate all of this stuff
[76:19] together. So, this is a great escape hatch if some pieces don't fit into your pipeline. Uh, you know, you can you can use jobs to orchestrate them all together. I will say and this is a a a sneak preview. I am pushing very hard
[76:31] for us to also add a feature where you can put tasks directly into your pipeline. So, if there's just a little bit of work, some DDL you need to run or some file you need to create, you can just do that at any point inside the
[76:43] graph. It'll just be like a little notebook cell that you can run at any point. Uh, cross your fingers coming soon. Um, but there's a couple of other escape hatches that I think are pretty cool. Oftentimes, we don't have the right data source. And another thing
[76:55] that people often don't think about, uh, you know, as a I like to think of these as wrappers for procedural code is these Python sources and syncs. So, you can use any existing pip library to read from any data source that has some
[77:09] Python interface to it. And this allows you to plug it into structured streaming. It's secure and isolated. It works great on serverless compute. It supports incremental ingestion so you can
[77:21] remember what data was ingested before. Uh and the streaming checkpoints manage this state and your code will automatically work with things like full refresh or reset checkpoint because it's all being handled by the streaming
[77:33] engine. So this is a great way to take for example the like GitHub pull request reader and plug it into declarative pipelines without giving up all the nice benefits of streaming and the graph and all the other things you saw.
[77:46] Another one is for each batch. So we recently added this into STP. You can now anything you can do in structured streaming in a notebook you can now do inside of STP. For each batch was one of the last things that we added. Um but
[77:59] the 4 batch is basically this streaming construct where instead of doing any processing with each microbatch, we just hand you a data frame and you can do whatever you want. You can do aggregations and joins. You can write it to external systems. You can merge it
[78:11] into another delta table. And so this is super powerful um because you know you can you can do all of these operations that we might not support. Now there are times when you shouldn't use this. So like for example one of the the main
[78:24] reasons that people used to use for each batch was for things like CDC. If you're doing CDC you should use autocd. But if you're doing something complicated that doesn't fit into autocd this is a uh a great thing. One thing I will mention,
[78:36] another feature, a preview that I didn't put in the slides, but that's coming soon is merge flows where you can specify the clauses of a merge but get all of the cool streaming graph stuff
[78:48] that comes with STP. So if you want to do things like aggregations or if you have some special requirement like these columns change but this column never changes, stuff like that, you'll be able to express that declaratively and simply soon. And then the last one is transform
[79:02] with state. So you know this is again I think one of the more powerful constructs structured streaming. Uh this is the the the right model here is you group by some key and then the streaming engine will give you an iterator with
[79:14] all of the uh new records for that key and then you have full access to a key value store. You can store values, you can compute aggregations, you can do whatever you want and then when you're ready you can emit results. It has
[79:27] timeouts and TTLs and all the things you would want. This is very powerful for arbitrary uh complex event processing use cases. Um and so now also fully supported inside of STP.
[79:41] Okay, that was a marathon. Uh thank you so much for sticking through.

Learn more about the Databricks Data and AI platform.