SQL-First ETL: Building Easy, Efficient Data Pipelines With DLT

Key Takeaways

Hello everyone. Thank you for joining. We'll let a few a few more folks trickle in. My name is Paul Lapis. I'm a product manager at Data Bricks. And today I'm going to talk about SQL first ETL. How you can build easy efficient data pipelines with DT and data bricks. First, any u anything that I say here is not a commitment. I'm required to show this slide by our legal folks. I wanted to start by sharing some of the challenges that we see teams facing when they're building ETL with SQL. First, development is slow and fragile. imperative code, especially in SQL, can get really verbose really quickly and adding things like a table, adding a column often requires refactoring of your pipeline. Debugging incremental logic is really, really complex. So is managing state, managing checkpoints. It really, really slows down development time. Second, there's a language divide. Your analysts are writing primarily in SQL whereas data engineers and platform engineers are using Python or Scala or other languages. This makes it really hard to hand off projects and to and it leads to duplicate work duplicate pipelines inside of your organization ultimately really really slowing down iteration time. And finally streaming is really hard. Oftentimes with most frameworks, you're forced to choose upfront whether you'll do streaming or batch and changing it later. If your requirements change, say that you have to run, you know, start running your pipelines more quickly because your business wants data more quickly, faster, you end up having to rewrite the entire thing. Very few platforms allow you to do this with a flip of a switch. So that's the world that we're in today. slow, fractured, and no real path to real time. Let me actually actually share uh what is declarative programming. I don't know about you, but when I first joined data bricks, I really had to look up the term. But this is one of the core shifts that we're embracing, moving from imperative code to declarative programming. Imperative programming assumes that the system is dumb. Think of it like cooking a recipe with 50 steps. You're manually iterating through rows, maintaining state, updating variables. It's tedious. It's fragile, and it's errorprone. Declarative programming flips the script. It allows you just to define what you want the end result to look like, and it ends up requiring a lot less code, much fewer mistakes if you're able to trust the platform to do the right thing. Now let's talk about why we see declarative SQL as becoming the default for ETL across teams that we speak with including analysts, data engineers, platform engineers. First, it's all about speed. SQL is built purpose-built for transformations. It was designed to help you define logic on top of tabular data. So it it allows you to develop much much more quickly. It also requires less debugging and uh it allows your teams to you know create something that you would otherwise do in a Python script in just a few lines of code. Second performance because it's declarative and you're just defining what the end result will look like the engine can optimize your SQL under the hood on data bricks SQL is is u uh compiled onto Spark anyway. So if you're writing SQL, you're making the bet that data bricks engineering can optimize your SQL code for Spark better than you can, which in my book is a pretty safe bet. So that means that you can get the best possible performance without any manual tuning. And third, I think what's most important is collaboration. Your analyst teams are already using SQL, but data engineers know SQL. business users know enough SQL to understand what's happening and to understand the logic and so you're able to self-s serve and collaborate with those downstream teams and oftentimes your organization is already using SQL for BI and dashboarding and reporting so using SQL for pipelines just makes everything consistent so what I want you to take away is that we see that SQL is the fastest path to production grade ETL simple powerful and scalable So I think so let's talk about where we're headed on the data bricks platform to provide you with a unified declarative approach to building a a like ETL pipelines using SQL. On the left hand side of the slide, you're seeing how easy it is to go from ingestion to transformation all with simple SQL with no orchestration tools required, no state management required. You're seeing us creating a streaming table for ingestion from from an S3 bucket and then a materialized view for a transformation. This this maps perfectly to the bronze or to the medallion architecture where you're landing your data in bronze using a streaming table. You're transforming it, cleaning it in silver and materialized views for complex transformations. And under the hood, all of this is powered by and made fast and and optimized for cost using the photon engine, the incre um incremental refresh technology, vectorzed s uh vectorzed indexing and cost aware optimizations. Now um what are these two objects? I wanted to give a little primer about streaming tables and materialized views which are the primary declarative SQL objects that u that we're talking about here. A streaming table is incredibly simple. You ingest data using it using only SQL, no spark, no orchestration. Take a look at the example here where you're just typing create streaming table. you're scheduling it to run every hour in this case and you're reading from an S3 bucket or even from a Kafka topic. It supports any data source that data bricks um supports. Under the hood, it's it's powered by Spark structured streaming. So, it's leveraging technologies that are tried and tested. But this opens the door for your entire organization to build real-time pipelines without really having to understand the internals of how streaming works. Streaming tables are incremental by design. So they will ingest one event at a time and they'll guarantee that they will deliver each row in your data set exactly once. They're efficient. They're scalable. They scale vertically and horizontally and they're a core part of the real-time architecture. So if streaming tables are great for ingestion when you're doing complex modeling and transformations, that's where materialized views come in. They let you premputee data. So they're ideal for query acceleration or for ETL use cases. So here's what a join in a transformation looks like using materialized views. And materialized views have a secret magic trick that we'll actually show you a demo of later where they will incrementally refresh when possible. And I'll talk more a little bit about how that engine works in the future slides. But ultimately they help you scale analytics with lower latency, simpler architecture, and it's all declarative. So let's make it real. There's two ways of using materialized views and streaming tables on the data bricks platform. The first is DT pipelines that I think I'm sure many of you in this room are familiar with and I'll give you a rundown of what that is in more detail later. But DT pipelines are ideal for production grade ETL pipelines backed by Git. If you want full CI/CD support, they support SQL and they support Python with automatic quality checks, lineage, scalability, failure retries, etc. And they're great for data engineers that want to build governed ETL. But if you're in the data warehouse and you don't need the full power of a full production grade pipeline, you just want to create some simple transformations for a project. You can also launch materialized views and streaming tables directly in DBSQL over JDBC. This is for quick SQL only workflows and it's ideal for analysts who want to be able to get their project done without spinning up a a whole pipeline. So the takeaway is there's flexible options. you can start simple and scale up and it's the same engine under the hood. Um, DT pipelines as I mentioned before are production grade uh a production grade engine for ETL. When you write code in DT like in this example we're creating a streaming table in a and a materialized view. You get uh like like uh out of the box observability, data quality checks, efficient incremental processing all with a simple declarative syntax in SQL or Python. And it's tightly integrated with data bricks. So it supports Delta iceberg. It's powered by Photon and all the efficiencies. And so we see teams using DT pipelines for all their use cases across ingestion, operational use cases, training ML models or just plain analytics. We're also excited this year to announce a simplified data engineering IDE. It's purpose-built for declarative ETL with MVs and STS in mind. And it's tailored for declarative coding. You can organize and execute your SQL or Python logic with built-in primitives. And it's modular and structured, so you can focus on one data set at a time as you're developing and iterating. You get previews, you get error feedback, you get automatic linting, a visual DAG entirely in one surface. It's guided, but it's flexible. So, it's great for both new users a and and if you're an existing developer of DT, you can upgrade directly to this IDE without impacting your pipelines at all. And it's all in one place. You can author, you can validate really quickly and deploy your pipelines without switching tools. So now let's talk about the lightweight SQL only path where you define you create and you manage your materialized views and streaming tables with no pipeline setup at all. These are built for um uh you know u performance and scale. They're using the same engine like under the hood. It's all powered by Spark at the end of the day. And it's perfect for analysts who are doing transformations and aggregations and connecting their BI tools with no orchestration or heavy lifting required. So if you're currently using create table statements to create tables or you're defining views on top of views in order to materialize results for a project, consider using materialized views instead. They will always guarantee that that the contents of the of the view will always reflect the query that the view is defined with. And as I said earlier, they're really efficient and fast. With these, you can get endto-end incremental without really doing any hand coding or manual logic in data brick SQL. So earlier I talked about the incremental refresh engine for materialized views. Um it's a cost model that analyzes two things. It looks at the change set of the source tables, what changed since the last time the materialized view refreshed and the syntax of the the query that you've defined your materialized view with. And then it picks the best possible strategy. So if if the change set includes new rows only, it'll just insert those new rows into the MV. If there's changes or deletes or updates in the change set, it'll craft the merge statements in order to update the rows that only need to be updated to materialize the result. So, it just picks the most efficient option for you. So, you get fresher data at lower cost without having to do any extra work. Over the past year, the team's been working really hard to expand query support. So we now handle more complex logic, all types of joins, window functions, aggregations all incrementally. We're also working on support for mutable um you know uh current date windows so that you can actually keep a running window of of data as the MV u as um time goes on. So the takeaway is that MVs are smart, not just simple. They keep your data fresh with zero tuning or manual intervention. So, don't take it from me. Now that you've seen how they work, let's switch gears and show you what it looks like in practice. I'm pleased to introduce Rit Rick Yatav, who's going to walk through a real world example. Thank you. [Applause] Thank you for the introduction, Paul. Uh, good afternoon to everyone present out here. My name is Rit Yadav and I lead the team which build the incremental view maintenance engine here at data bricks. Um today we are going to demonstrate some of the functionality that we built as part of this incremental engine. Uh before we dive into the demo, let me first walk over um walk through some of the setup that we have here. So we are going to create a single MV that is based on a variant of a TPCDS decision support benchmark uh query. It aggregates sales and return data across items in a varied uh product catalog. It performs two inner joins, one left outer join, uh followed by an aggregation and each incremental refresh it processes 3 to 5% uh changes to the base data. To give you an idea of the scale that we are going to operate at here uh in this demo uh we have 6 and a half terabytes of uh sales data uh spread out across half a million items in the product catalog and roughly half a terabyte of uh returns data and uh we have two billion uh sales that are processed incrementally on each incremental refresh. We ended up uh modifying the catalog uh for 100 items and uh also added 73 million uh returns um to the returns table. So that's what the setup looks like. Let me go and kick off the demo here. So this is the editor that most of you would be familiar with. This is the notebooks editor where you could go write SQL for your MV. You can see all of the joins which are defined here. These are all equins uh followed by the aggregation uh which would get written out uh to the MV eventually. You could have also used the DBSQL uh tool to create an equivalent MV and like Paul pointed out you would have got the same incremental performance um either way. So the incremental engine underneath the covers remains uh the same. All right. So let's go ahead and launch the full recomputee first. It's it has already been validated. So what I'm going to do is I'm going to use the full refresh all uh functionality. It is going to give me this warning that all of my data would be truncated. And um we are going to compute the entirety of this um MV from scratch and then write out uh all of the data. We have already got a cluster here from the serverless warm poolool and right now it is in the planning phase. Um it didn't really have to do anything in the planning phase this time around because we have already told uh it to do a full refresh. But in the case where it does have the option, it tries to compare the computational complexity of the incremental plan against the full refresh plan and determines the cheaper option uh and triggers the most efficient way to refresh your MVEs. And all of this is actually done uh behind the scenes based on our cost model which we are constantly improving. um and it's going to um try and figure out which which option is really uh computationally more efficient. So we have kicked this run uh off and um it's going to take a while to uh to process like almost 7 terabytes of data that we have here. uh you can see some of the metrics um uh in real time and um there is this performance tab that is added here. If you click on the performance tab, what you see is the query plan that is executing uh behind the covers. You also get to see some of the statistics which are involved. So it has already processed roughly 60 gigs of data. There is no pruning that has happened here because this is a full refresh. So you don't uh prune any of the data. you like the currently it is performing all of the joins the aggregation hasn't started yet so it let's go back let's uh wait for it to complete I I guess that's the only option that you've got with uh with a full refresh now one other thing that I would like to point out here is uh we also have uh behind the scenes autoscaling in place so if we have uh a query or the refresh uh if it is computationally really uh intensive. Uh the serverless autoscaler is going to constantly keep on adding uh executors behind the scenes so that u your refresh uh could finish uh quicker and all of this you don't have to configure anything. It happens behind the scenes out of the box. And we didn't want to wait for the full refresh to complete. So let's go ahead launch the incremental refresh as well. Here we are just going to go click on the start button and um we would get a cluster again from our warm poolool. It's planning um this time around it would have the option of doing incremental refresh and you'll see that hopefully the incremental refresh uh finishes much quicker than um the full refresh uh here. So the cost model here uh what it would try to do is at it would try to figure out uh all of the changes that had been made to the source tables that uh the MV was reading from. It would try to determine the changes at every operator that was used in defining the MV. So it would try to figure out the changes at the end of the first inner join then the second inner join and then take the inputs from the second inner join try to figure out the the change set that would uh that we would get at the end of the outer join and then finally the aggregation before making any changes to the um MV here. So it tries to estimate everything without really executing uh all of this and it tries to figure out the best plan uh to refresh the MV here. If you look at the performance tab for the incremental refresh, it already finished in under a minute. Uh and we um we processed uh roughly 3 to 5% changes. The query plan I would uh call it a work of art and I'm slightly biased here because uh my team built it and I would like to invite some of you if you are interested uh you can come along uh have a pizza with us and we can draw this incremental query plan at the back of the pizza box uh and see if it fits. Um all right so let's go back check back on our full re uh fresh run. still chugging along four minutes in. And what I'm going to use is I'm I'm going to use this time warp functionality. Uh I don't want to waste all of your time. It's but even if you manage to create a time machine, it's it might it might actually give you better latency, but it would not save all of your executors from all the compute that is getting expended over here. So let's see how long it takes. Um hopefully it finishes sooner. Well, it took us 17 minutes 56 seconds if I'm not wrong. All right. So, that's what you get when you are trying to compare incremental refresh with uh uh full recomputee. All right. Now, let's go look at some of the statistics u around u the comparison. So, we are 19 times faster with incremental refresh in this specific case. We are also almost 28 uh times more computationally efficient with the incremental refresh. With the full recomputee run, we scaled almost up to 62 executors at its peak. And with the incremental refresh, we only ended up using four executors uh at peak. So that's the difference that we get uh with incrementalization of your MVs. Now another point that I wanted to drive across was the fact that as you process uh more and more changes to your base tables what the computational utilization uh somewhat remains stable and it is uh proportional to the amount of changes that are getting processed rather than the full size of your data. Um so this is really important for all of the customers which operate at the pabyte scale and uh they want uh to only uh use as much comput resources as the changes that are being processed. Now before I end this uh demo I wanted to call out some of the best practices. So we have all of the supported operators in our uh documentation. uh there are certain operators which require row tracking uh to be enabled on the source tables uh for you to get u incremental performance. There's another well-known feature that we have uh it's called deletion vectors. If you have it enabled on your source tables, it minimizes the amount of uh changes the change set size uh gets minimized. The reason is that we don't have to write the the changed files um again and we only have to figure out the rows that have changed and write out the changes to those rows rather than rewriting the entire files uh in those um in those source tables. And again I cannot stress enough how much proper clustering on the grouping keys as well as the join keys is going to help with refresh performance. Um, in the example that I had used, all of the source tables, they were already clustered on the grouping keys as well as uh the join keys. And with that, I'm going to hand it back to Paul. And thank you so much for listening. And if you have any questions, I would be uh out there after the session. I would be happy to answer any other questions that you might have. Thank you. [Applause] Wow, pretty compelling, right? Now imagine bringing that same incremental performance and intelligent optimization to DBT. No rearchitecture, no complex migration, just some minor code changes and you can update your existing DBT models to be fully incremental, faster, more efficient and um endto-end streaming. So with that, I will introduce my colleague Messian who's going to give us a demo of just that. Thanks Messian. So hi everyone. Uh today I'm going to walk you through how can we dramatically simplify data pipelines by moving away from the traditional DBT setup to using streaming table and maturized views in data bricks. So pipelines today are very complex. Team normally spend a lots of times to do writing things up including managing the ingestion uh doing the incrementalization computation as well as coordinated system and to end. But with streaming table and maturized wheels, everything could just be expressed in simple SQL and you will get automatic incremental incrementality as well as lower latency. So uh let's dive into the demo. Uh let's start with by looking at uh how a traditional DBT pipeline normally looks like. Here I want to ingest some raw bookings data into a browse table. What I have to do is to create the table first and then run copy into to ingest the data to the browse table I just created. And then for this specific case, I also run the other transformation to just cast the data type into the correct ones. And this works, but let's be honest, for simple injust operation, it takes quite a bit of steps to get to that point. Now, let's kick off the run here. It's probably going to take a while for the execution to complete. So let's take a look at the other two sources I prepared. Here is another source I repeated the same setup including the payments and there's another one is the reveal source. All these sources I'm going to use is to for the later on golden layer transformation. As we can see the the previous run is still continue running. So what happens here is each time when we trigger DPT run from scratch DPT has to connect to data brick uh data bricks SQL warehouse and that connection time takes a quite a bit of latency for into the end to end latency of this execution. Now as you can see uh it takes about one minute to just start the query and the end to end latency is about 1 minute and 30 seconds. Now we have all the sources get prepared. Uh let's move on to the goal layer transformation over here. Here I want to compute some metrics including total bookings total bookings with payments as well as average rating by joining all the tables I just created. And to make sure this model could run incrementally, what I have to do is to specify materialized equal to incremental having a incremental strategy with unique key and specify a ginger mark to let the model know how to load the data incrementally. This works but obviously it requires a quite a bit handwritten logic and if I want compute some matrix more complex it's very hard to get the incrementality incrementality part correct now I think let's kick off the run here okay it's probably a little bit longer Cool. Uh so when I kick off the run here, uh pretty much the same story is takes quite a bit time for DBT to connect to the data bricks warehouse and that adds into the end to end latency and I probably fast forward this one. Cool. Uh let's switch to the jer to see how much simpler the pipeline could become by using streaming table and the matur views instead. Let's get back to the previous ingest chase but this time I'm using streaming table. You could see the whole ingestion and transformation could get replaced by a single simple select select star with read files and that's it. As it's a streaming table, data will get processed incrementally by default. And with a schedule specified, we will create a background job to process the new data periodically without having to go through the step between DBT to data bricks warehouse. So all those incremental refreshes will get much lower latency through that. And uh I think we could kick off the run here again. So at this time I already run multiple queries. So SQL warehouse probably already warmed up and is probably running a little bit faster for this specific case. While we are still running list, I think we could also take a look at only other sources. I repeated the same setup for Cool. Here's the payment one. Just a single select as well as the review one is also done with a single select. And uh now let's take a look at the goer transformation again but this time. So I think I probably won't be able to show you how much simpler the streaming table becomes after this one. the materialized view part but the story is pretty much the same for the materialized view case all I have to do is just writing up the query that express that compute the matrix itself and I don't have to do do anything additional the system will automatically perform the incremental refresh for me again and uh I also did like a ingestion to the source table of the matur views to see if the model actually get incrementalized Here I'm going to the unifi list to figure out to find out all the DBT pipelines I just created. Uh here we could see all the DBT pipelines I created are showing up in the unified list. And then I'm going to the catalog explorer page for checking up the table I just created. Here for the matured views, you could see the full SE definition displayed on page and on the left hand side right hand side we could see the refresh schedule as well as the current update status shows up. Now click into the update details page. We land on the DT page where you could check more detailed information about your latest update. And here uh I sample some data just to get a rough idea of what the mature zero looks like. And then I go to the event log to verify that the latest update I performed extra runs in the generic aggregate algorithm which is indeed a incremental algorithm uh we have for matur. So uh let's eight uh so let's recap uh with streaming table and maturized views we can dramatically simplify and speed up data pipelines everything could just get expressed in declarative SQL with no more ginger no more config required for building up that pipeline and we handle the incremental compute automatically with more scalable service background compute. Uh so let's Oh, thank you everyone.