Deep Dive Into Streaming and Batch ETLs With Lakeflow Spark Declarative Pipelines

Summary

This video provides a deep technical exploration of Lakeflow Spark Declarative Pipelines (SDP) in Apache Spark, explaining how Python and SQL abstractions automatically resolve complex dependencies and build optimized DAGs for both batch and streaming workloads.
Internal architecture details covered include the Spark Pipelines CLI, gRPC message flow between the CLI and Spark Connect server, dp.table and dp.view decorators, and automatic streaming checkpoint management that requires no developer configuration.
New CDC support with SCD Type 1 and Type 2 is explained alongside the CDC merge writer, which replaces verbose for-each-batch patterns with declarative table definitions, delivering less code and less infrastructure maintenance.

Deep Dive Into Streaming and Batch ETLs With Lakeflow Spark Declarative Pipelines

Let's deep dive into the new declarative ETL pipeline framework in Apache Spark™, Lakeflow Spark Declarative Pipelines (SDP). Let's peel it off to learn how SDP's high-level Python and SQL abstractions translate into lower-level Spark SQL and Spark Structured Streaming queries. During the talk, you will learn how SDP automatically resolves complex dependencies and builds optimized Directed Acyclic Graphs (DAGs) for both batch and streaming workloads. We will walk through the internal state management and orchestration logic that allows SDP to handle retries and incremental processing out of the box, replacing thousands of lines of imperative "glue code." You will leave with a clear mental model of the engine's architecture, the pros and cons of SDP, and be ready to debug and optimize your pipelines. You'll understand where and how to use it (alongside your existing ETL pipelines).
Talk By: Jacek Laskowski, Freelance Data Engineer, books.japila.pl ;

Chapters

FAQs

What is Lakeflow Spark Declarative Pipelines (SDP) and what problem does it solve?

SDP is a declarative ETL framework released in Apache Spark 4.1 that lets engineers define transformations using Python or SQL without writing infrastructure code, retry logic, or dependency management. The framework automatically handles checkpoint management, DAG resolution, and incremental processing so teams can focus on transformation intent.

How does SDP automatically manage streaming checkpoints?

SDP handles streaming checkpoint creation and management automatically with no developer input required. The dp.table and dp.view decorators register table and view definitions with Spark Connect, and the runtime manages state persistence so streaming workloads resume from the correct position after interruptions.

What CDC capabilities does Spark Declarative Pipelines provide?

SDP includes native change data capture support with SCD Type 1 (upsert/overwrite) and SCD Type 2 (historical versioning) semantics. The CDC merge writer replaces complex for-each-batch patterns with a declarative merge definition, substantially reducing the code required to implement change tracking in pipelines.

What are the primary component types in a Spark Declarative Pipeline?

SDP supports three primary component types: streaming tables for continuously updated data, materialized views for precomputed transformations, and syncs for writing results to external targets. Code is written using Python or SQL only, reflecting the framework's focus on declarative transformation rather than imperative infrastructure management.

Full transcript

[00:09] Uh welcome welcome all to my talk uh about something that I started exploring quite some time ago. SDP it was released for the very first time in Spark 41. So now we've got 42 release candidate 3.
[00:22] It's supposed to be released around that time. So perhaps some things may change uh version wise, not necessarily feature- wise. As a matter of fact, they promise more than they delivered. So
[00:34] after all, we've got what we get. So um Spark 420 RC3 means that there is one feature that's not production ready yet, meaning that it's not released. So we'll
[00:47] touch base on this soon. So deep dive into Spark declarative pipelines. We are going to talk about uh how things changed uh uh for the better um with regards to how we are supposed to
[01:00] develop ETLs using spark. Um but then there was RDD API then we moved on to data frame API and now we've got spark declarative pipelines. Some people may
[01:13] have some objections around the direction that spark took but I like it. it's closer close to to my beliefs that ETLs should not be as hard as they are.
[01:25] So I'm not going to pitch um spark declarative pipelines. I simply assume that you know what spark declarative pipelines are even if it means that you just watched few uh videos some some you
[01:38] know pitch talks or or whatever. So just my assumption is that you know you heard about spark declarative pipelines but I'm going to do some quick intro
[01:51] and to spark declarative pipelines from person who is interested more into nitis greedies of how it works under the covers. Okay, because we are here to learn more from the sources. So rather than you
[02:04] know letting cloth index uh the source code and start asking questions well we are here to develop our understanding basic understanding about the source code and then our you know our uh
[02:17] prompts are going to be you know token free so less code and less maintenance that's the premise of SDP we are supposed to uh define our wishes without
[02:30] you know dealing with infrastructure code so The premise of uh SDP is to focus on transformations. So you are supposed to uh code as you probably have done already um tables, views and uh uh
[02:46] syncs. These are the main components with flows that are here and there but uh let's not talk about it. Um so the code code can be written using Python and SQL only. So no Java, no Scala.
[03:01] Well, if you could if you uh wish you could just do it. Uh but uh that's not the you know uh um feature proposal of SDP. So Python and SQL only. And that's exactly shows the direction where we are
[03:15] heading to with regards with uh Spark declarative pipelines. We are higher level um assuming that Scala and Java are are uh assembler like languages for Spark. Spark is written using Scala and
[03:29] Java. So for the time being for this for the purpose of this talk let's assume that Scala and Java are assembler alike.
[03:41] see in a in a sec or what you have seen already is basically uh SQL and Python code lots of devs uh without infrastructure code in between. So that's basically Python code and SQL
[03:55] code. So you can you know already how to test it. Piest for the rescue right so um the main premise no orchestration no compute management and error handling
[04:07] that's something that you offload to SDP good we start our journey with spark pipelines CLI by the way have you noticed that sometimes we refer to this
[04:19] product as spark pipeline there there will be some references to singular whereas sometimes we say that the this is spark pipeline means so yeah that could trigger some confusion even if it
[04:33] if it not uh in the past it will because I just mentioned it I was confused a few times already so spark pipeline CLI we've got shell script and what struck me when I first open you know the the
[04:47] the shell of uh this shell script was that this is just basically spark application so spark pipelines is yet another Spark application
[05:00] launched by Spark submit as we did in the past with uh any other you know Pi Python based application and this Spark pipeline shell script only launches uh
[05:14] Pispark Pipelines CLI PIY um um script Python script and that's the reason why you may see sometimes in error codes uh reference to CLI py even though you you
[05:27] might not explicitly call it. Okay, so just for your information. So it's just a pi spark application. Yet another pi spark application. Okay, there it was. Then ah uh by the way I'm a heavy user
[05:41] of UV. Uh if you haven't heard of UV, now you've got the chance. You should be using UV early and often. uh with this UVX you don't have to install anything to launch uh pispark pipelines even
[05:55] unreleased version well release because it's available out there but uh pre-release uh with this simple looking in quotes uh command so with this pre-release allo this is something that
[06:07] I learned just before this presentation because I needed to pull this um uh def five uh it doesn't really matches you know release candidate 3 but we are talking about scala and Python. Are you
[06:20] confused? Yeah, we probably will soon. So, let's keep it. So, the five is RC3. Yeah, because that's how things work. Okay. And for pipeline CLI, we've got
[06:32] three commands or subcomands. One is run. You probably guessed what it is for. Dry run. I'm sure that you know what it is for. And init. So, basic
[06:45] stuff. within it you set up your project and you dry run it followed by run. Okay. And the the the the the goal of this CLI is to execute these
[06:57] commands over and over again. Okay. You do something you want to test how it's going to work or how it should work and then so you pick between run and dry
[07:09] run. Okay. You start with init and once init is done you've got your scaffolded folder with all the files that are needed and then you can just modify them uh as you want and then uh you run all
[07:21] dry run uh interchangeably. Good. So what is what is CLA uh doing under the covers? It's as I mentioned it's a tiny little um CLI wrapper. So
[07:35] it's a Python pi spark application and what what was really clever what I found really interesting is that hey we are in
[07:47] python environment already because we just launched pi spark application. So if we happen to load any pyonbased
[07:59] files we can import them directly. If we import them as if it was code that you've written, everything that's in
[08:11] Python files that you wrote your transformation in are executed automatically. So that's the that's the goal of this CLI. It starts
[08:25] finds all the files that you defined as part of your application. finds all the SQL and Python scripts, transformation scripts and for Python scripts, it imports them automatically
[08:39] and since these defs are uh decorated with this Python decorator, for example, DP.t this code is going to be executed.
[08:53] And guess what? What happens under the covers while this death is being executed? This decorator is triggered and as part of this decorator
[09:06] there is a remote call to applica to spark connect to register what we just described by our code by our deaf. That was something earthshattering for me. It
[09:19] was like what a clever idea. So just by doing dp.table table you say to some remote spark connect server that you are about to register something that is
[09:33] known as ingestion st so you can guess it's a name of your future table and this table is described by good old dataf frame API so nothing changed
[09:47] except that you are not supposed to write write after load as you have done in the past, right? So, it's not that you just it's not that you
[10:00] haven't finished your work yet and you should write the second part, the right right. I'm not sure about this pronunciation is right. Right. Yeah, it's right. So, okay. So, after load
[10:13] there should be right side, right? Right. So,
[10:25] this code and sends it over the wire to spark connect server for future execution. We haven't told this code to be executed yet. But we are just
[10:37] registering all the objects. Okay. So we registered this and and what's what's very important is that we are using good old delta um delta data
[10:50] frame API. So you immediately notice that we are now creating spark read stream which is streaming data frame. If it's streaming means that there's there
[11:03] should be somewhere checkpoint location. Why do we need checkpoint location? to know exactly where we finished and where we should start our future, you know, uh, iterations. Right. Right. So, but
[11:17] look at this. There is nothing about checkpoint location. And by the way, why did I even mention it? Okay. Because I know that you are old enough codewise, not age wise, and you knew about this
[11:29] feature that now is gone. Well, not really, but that's something that SDP will take care of. Okay. And that's one of the uh uh features of and goodies of
[11:42] SDP. Moving on. So in this Python mod module and there are corresponding SQL statements as well. They are supported directly by uh Spark they have to because SDP is part of Apache Spark. So
[11:56] without Apache Spark support for SQL statements uh SDP couldn't do much about it. Okay? because after all SDP will ask uh Spark SQL resolver to resolve and
[12:09] plan all the statements. So now you can see that I created a table but this time I use delta format which got added quite recently with delta 41 if I'm not
[12:24] mistaken. There are lots of force uh with um with spark delta lake unity catalog and mlflow and and whatn not. Okay. So format delta then you specified
[12:36] name of your table and some properties. But look at this. You just set what you want from this table. How this table should look like. So you define it
[12:49] rather than saying apache spark how to do it. Okay. You basically wished all these annotations will take place and will take effect once this code is
[13:03] executed. Uh among this uh uh pispark pipelines bound methods or decorators, Python decorators, if they are decorators with prefixed with at uh for
[13:16] the purpose of this slide, uh they need to be uh around your death in Python. Okay, that should be obvious for people who are Pythonistas, not so much for for SQL devs. Uh so uh whenever you see at
[13:30] on the slides, it means that it's a decorator. So it needs a corresponding defaf and this death is wrapped and sent over the wire to spark connect. Okay, server. So we've got appendent flow,
[13:44] something that's brand new. So for the very first time we've got CDC support in SDP. Then this is uh slowly changing dimension type one. Uh type two uh was
[13:58] supposed to be part of this release but they didn't manage to to uh um uh make it happen. So we've got uh SCD type one for this auto CDC flow. We've got create
[14:10] sync. We've got create streaming table. We've got materialized view, table, and temporary view. And I'm pretty sure that you know what these uh annotations and nonotations uh uh uh pieces are all
[14:24] about, right? Except perhaps uh create sync might not necessarily be that obvious. It just, you know, creates something that's available somewhere for the purpose of your Dark uh directed
[14:38] cyclic graph of this uh project. it's going to be one of the many notes uh that that transformations will lead to. Okay, so um let's move on. Okay, so
[14:51] let's dig deeper if we weren't dig enough. So Spark declarative pipeline supports a Spark connect only. Okay, so it's remote client for Spark infrastructure.
[15:04] There is this idea I haven't tried but I'm sharing it whenever I'm talking about SDP that I truly believe but I haven't checked it yet yet that this code whatever you write using open
[15:16] source should also be somewhat executable by datab bricks infrastructure. So I believe that if you happen to know how to you know tie all these loose notes together just by
[15:29] pointing uh your code Spark CLI with remote datab brick server it should be possible to deploy what you just uh created written using SDP open source onto uh serverless infrastructure uh
[15:43] deployed somewhere at datab bricks. Okay. So that's this idea I haven't tried but uh from a theory perspective it should be possible. So spark connect only. Okay. Good. It since we are
[15:57] digging deeper it will spark CLI will check whether your spark API mode uh is connect and this is this can only be connect. So if it's not for for any
[16:09] reason it will blow up. Okay. assumes local by default as d- remote. This is a configuration for your spark spark connect right server. So if you haven't define it explicitly, it's going to be
[16:22] local. Okay, there there are some consequences of this uh uh local versus non-local and you can guess that uh certain uh file formats uh do not allow for overrides uh
[16:36] like parquet which is default in open source unlike delta lake which is default in datab bricks environments and SDP execution always happens somewhere outside your local machine except in
[16:50] this uh very generic very default setting. where local server is spin up automatically for demo purposes for example and it runs locally but in production environment it's going to be
[17:02] somewhere over there okay although your spark connect spark pipeline CLI may run on your machine good since we are talking about spark connect we are talking about something written
[17:15] in scala and or java and for spark declarative pipelines spark declarative pipelines yes uh we've got this pipelines handler that will intercept or
[17:27] will um allow for this communication between spark pipeline CLI and spark connect. So spark connect has this extra uh extra extension to process all the
[17:40] gRPC protobuff based uh messages so they can be executed uh on spark connect. So we've got gRPC and Protopath involved in this communication and remember we are
[17:54] we are starting from Python but we end up on Scala side. Okay. So just just for your information that we are dealing with two uh different um two different
[18:06] environments written in different languages but not a big deal. Okay. So pipelines handler is just a bridge between all the execution environments. Your local environment where you
[18:18] executed spark pipeline CLI where all files are and your spark connect. Good. So spark pipelines run when you run trigger one of the three available
[18:31] commands like run it will fire up series of gRPC messages. uh and this slide is supposed to show you what messages uh will be. So we are
[18:45] on your left and we are starting with Python. It could be also be SQL but as you will soon notice SQL is directly sent over the wire as is without any uh
[18:57] premputation or pre-transformation on client side. Okay, because well we can't do much about SQL. So we send them over the wire to spark connect and pipelines handler will know what to do with this
[19:10] particular message. Okay. Other than that uh python is executed on client side. Uh and the f the very first command is create data flow graph and
[19:22] this data flow graph is basically ID of your future DAC. Okay D that that's made up of tables, views, uh flows and syncs. Then after uh for SQL files for all SQL
[19:37] files there will be define SQL graph elements message sent over the wire to pipelines handler and at the end for run there will be yet another third command
[19:51] uh gRPC uh message called start run. Okay. So this this series of um messages gRPC messages is basically protocol for
[20:03] uh spark pipeline CLI to talk to spark connect and say hey this is my project and you are supposed to execute it. So you can intercept this communication and pretend that you are spark
[20:16] pipeline CLI and do basically the same. Okay. So you are not bound to this uh python environment. you can write your rust envir uh rust cli for spark pipelines whatn not okay and on the
[20:29] scala side you've got corresponding protobuff generated autogenerated uh messages like create dataf flow graph define SQL and start run okay and for each and every python definition that's
[20:41] executed on spark pipeline CLI side there will be corresponding message that hey I am about to create this or or there definition of this table or there
[20:54] is definition of this uh view or sync. So there'll be corresponding messages over the wire. Okay, so lots of communication going on just to start your Spark declarative pipelines uh
[21:06] project. And the same happened if you ever worked with DT. you had to wait for you know visualization to be rendered in databicks UI uh for DT and then the
[21:19] start also took some time uh DT it's now rebranded to be uh lakehouse uh spark declarative pipelines what you uh experience is open-source version of
[21:31] what you can uh work with uh using datab bricks good okay moving on dry run can you spot the difference between run and dry run.
[21:43] Well, I guess you're probably not because well the only difference is just a very tiny code inside start run uh start run uh processing
[21:58] how to say it pipeline not to overuse pipeline and uh it will choose between dry run and run that's basically it other than that from protocol perspective they are the same except one
[22:11] field being boolean and it's on off telling whether it's uh uh dry run or not. So again dry run and run will always run on spark connect. Okay,
[22:23] that's the takeaway from this uh message uh from this slides. Okay, so spark pipelines run on spark connect. So everything happens on spark connect. So we've got um pipeline execution start
[22:36] pipeline. It's written in Scala. So if you are interested in source code, you know that now you need to switch from Python to and or SQL to Scala and just review pipeline execution start pipeline
[22:48] and what it does is basically resolves all um uh links uh between uh tables and views that may or may not be created defined by your pipeline and or be
[23:04] already available let's say in unit catalog. Okay, they could be also external uh tables. We are not bound to views, tables and syncs that are only available within our spark declarative
[23:17] pipelines project. They could also already be available in our uh three level u unity catalog managed uh structure okay or catalog.
[23:29] We've got a fully refresh and also fully refresh all and uh refresh uh options for run command. And you probably
[23:41] already guessed what they for. Uh fully refresh just go over uh goes over all tables, all definitions and basically truncates whatever they uh was already available based whether it's a streaming
[23:53] or batch table. Uh there are some differences and uh well full refresh is a subset of uh tables and views to be touched by fully refresh. There is a fully refresh all variant for refreshing
[24:07] everything that's part of this project. And the last but not least just refresh uh so we can partially refresh uh our views, our tables, whatnot. Okay. So you can play around and what what's next?
[24:21] erh run will materialize all the data fra data sets and these data sets are tables and views and by saying that they are materialized I mean that they are registered in catalog that's part of
[24:34] your infrastructure it could be unity catalog within um well it could be within in data bricks and you could also set up your spark environment to be unity catalog ossbased environment too
[24:48] why not so materializing means that all the names are registered in this registry of all tables, all known names and so create metadata happens as well
[25:00] as truncate table, drop table and all the other SQL based commands statements that are part of this fully refresh and or fully refresh all and or refresh
[25:13] option. So depending on what we want from you know underlying tables data we may trigger truncate table drop table and others good and the last but not
[25:25] least it starts triggered graph execution which happen which happens on its own thread. So when you scan all available threads, you will see that there is topological execution thread.
[25:37] There is going to be this new name among the other threads. And it basically uh will execute up to 16 flows at once. And if you wonder what flows are, they are
[25:50] uh transformations between uh your tables because your table may depend on another table as part of your code. And within this definition without deaf in Python you may have some transformations
[26:03] and these transformations are these edges which mean that they will lead from one table to another. Okay. So there is connection and this connection is called in this graph this connection is called flow. Okay good. Okay. And
[26:17] we've got flow planner. We've got flows and flows executions which are basically a different representation of our code. Remember we started with SQL and Python. Now we are on Scala and for Spark
[26:31] connect with Spark SDP infrastructure want to know uh what exactly we need to be executed and by uh planning our um our uh declarative pipelines uh and um
[26:46] elements uh we may end up with uh batch table right for uh good old batch data frames but also streaming flow execution. which is trait for streaming table
[27:00] writer sync write and sedd1 merge streaming right okay in other words you are dealing with batch and or streaming
[27:12] tables that's basically it plus something that you might not necessarily um worked with in open source version of Apache Spark sedd1 merge summing right
[27:24] and now we've got 12 minutes steel and we are going to go over uh each. So botch table write writes writes is responsible for writing your code executing your code you know up to load.
[27:39] So we know what data frame uh this code is supposed to work with and basically we'll uh at the end you see data frame writer with all the options that you may or may not set up for your uh pipelines
[27:52] project. Uh there's mode append. You see that it's it cannot be overridden. It's basically append and it's only append. And then save as table. That's it. So
[28:05] this piece of code that you may have written by yourself uh in the past now is gone because that's part of SDP. Okay. Basically this right part is
[28:18] on right side and you are not responsible for it. Whereas whatever is left is yours and you are supposed to create series of defs that with the only um premise that they should return uh
[28:31] data frames either batch or streaming. That's basically it. Okay, good. So batch batch table write will be instantiated any code that starts with
[28:45] this DP table or is decorated by this decorator Python decorator um DP table. Next streaming table writer is one of the three available streaming uh uh
[28:57] pieces. So it's going to write streaming data frame to a table and it's created for the same uh Python decorator at
[29:10] table at DP table but the difference lies in your code. So if you happen to create data frame that was instantiated after you started with spark read it's
[29:24] going to be uh the other writer. Whereas if you happen to start your code with spark read streams, it's going to create streaming data frame which is exactly uh
[29:37] handled by streaming table writer. And next sync write this is for this piece of code that you may need for some external uh syncs like uh DP create
[29:50] sync. It's no long it's not a Python decor decorator. So create sync you give it a name you configure it with some other you know properties uh so it's something that's a logical reference to
[30:02] something that exist over there somewhere over there uh and so you are now crossing the boundary of your project and then you can uh create one of them or many appendent flows that
[30:16] will write to the sync and for this you will end up with sync write. So the code looks like data.right stream query name. And if you've ever created spark
[30:28] structure streaming code, you immediately recognizes that uh you immediately recognizes recognize that this is spark streaming code that you previously had to write. Now it's gone
[30:41] because well SDP will take care of it uh for you. So you don't have to worry about it. Output mode hardcoded always append. good uh sedd1 merge streaming right it's
[30:54] a brand new uh feature that's coming in uh spark uh for two so it's going to be available later this week or early next week or somewhere soon uh so apply CDC
[31:10] event uh stream to target table using sedd type one type two it's not available will not be available in 42 but it's coming and I know that streaming team at the um data bricks are
[31:23] working on them on it. So the target table has to support supports row level operators which is one of the many uh data um data source uh v2 API uh
[31:36] interface. Uh so if it's okay then um this write is going to happen and again there's be uh there's going to be write stream query name option trigger and
[31:48] this trigger will always be uh trig available now which means that there'll be a check whether data is available. If there is something to be processed, if there is nothing to be processed, this
[32:00] summing query will simply um disappear and will never be executed because there is no data to be executed, right? And you see that uh for this SEDD1 uh there is for each batch. Probably if you ever
[32:13] worked on SEDD1 or SEDD2 type u pipelines you had to use uh for each batch to handle you know uh late events and reconcile and problems and whatnot.
[32:27] So now it's handled automatically. uh streaming. You might be wondering where we talked about about streaming queries and you might be
[32:39] wondering okay uh we are going to execute uh run first time. Okay, this command finishes because there is uh all queries are executed with available now. So either there is something to be
[32:51] processed or not. We are done. So siming query will finish and now you are going to execute it this run again and again and again. And what about state of our
[33:03] streaming queries? And the answer is that one of the er least recognized properties of our pipelines um configuration file is
[33:15] storage and storage is exactly for the uh for storing um checkpoint locations of each of every streaming query available in our pipelines project. So
[33:28] that's where you can find checkpoint location and all the files that uh describe what offsets were already processed and and other streaming related uh metadata. Okay. And storage
[33:40] is required plus name. So there are two required properties in Spark Pipelines YAML file. One is name and another is storage. Exactly for the purpose of our
[33:52] streaming queries. Good. And that's basically it. Yes. If you want to know more about spark declarative pipelines, I'm publishing you know all my findings thoughts to LLM
[34:05] to you. Uh so you can just index it with your favorite LLM provider and or toy and basically uh have a look at human readable more readable uh version of
[34:20] what you can find in source code that's spread across Python, Python and Scala basically. Yeah. So uh enjoy and that's basically it folks. So, any questions?

Learn more about the Databricks Data and AI platform.

The information provided herein is for general informational purposes only and may not reflect the most current product capabilities or configurations.