https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/ba-p/151728 <H1>How to: Master Streaming Tables and Materialized Views</H1> <P>Make sure to check out the previous post: <A href="https://community.databricks.com/t5/technical-blog/spark-declarative-pipelines-how-to-series-part-1-how-to-save/ba-p/149180" target="_blank" rel="noopener">https://community.databricks.com/t5/technical-blog/spark-declarative-pipelines-how-to-series-part-1-how-to-save/ba-p/149180</A></P> <P>Welcome to the second post of the <STRONG>Lakeflow Spark Declarative Pipelines (SDP) “How-To” Series</STRONG>! In the previous post, we saw how SDP lets us focus entirely on business logic by automating persistence, checkpoints, and state management.</P> <P>I can hear you asking:</P> <P><EM>"Okay, I get it now. I don’t have to call an explicit </EM><CODE><EM>.save()</EM></CODE><EM>. My code is cleaner, my logic is decoupled from the plumbing, and I’m finally thinking declaratively. But now I’m looking at these decorators (</EM><CODE><EM>@​​​dp.table</EM></CODE><EM> and </EM><CODE><EM>@​​​dp.materialized_view</EM></CODE>),&nbsp;<EM>and they both seem to create tables. Wait, there’s another one, </EM><CODE><EM>@​​​dp.temporary_view</EM></CODE><EM>? Which one do I use for my Bronze layer? <I><SPAN>Which one is for the Gold layer</SPAN></I>?"</EM></P> <P>In this post, we’ll dive into these building blocks and show how to choose the right one for a specific use case.</P> <H3><STRONG>The Basis: It’s all about the Flow</STRONG></H3> <P>In the declarative world, we don't just create tables; we define the flow that hydrates them. Think of a flow as a "data contract" between your source and destination.</P> <P>A flow determines two things:</P> <UL> <LI><STRONG>Which</STRONG> data is moving (defined by a query or a DataFrame).</LI> <LI><STRONG>How</STRONG> it lands (Append, Upsert, or Replace).</LI> </UL> <P><SPAN>A flow is not just a query - it is a stateful, incremental computation with zero plumbing. It tracks what has already been processed, so that on each run only new or changed data is handled. This is what allows incremental semantics, whether you run it in streaming or batch mode.&nbsp;</SPAN></P> <P>In SDP, flows are the core of your pipeline. Sometimes they are defined implicitly through a decorator, and sometimes you define them explicitly to handle more complex logic.</P> <P>Finally, each flow has its own destination - a physical table in Unity Catalog (UC), or a sink to an external system.</P> <H3><STRONG>Virtual Logic: The </STRONG><CODE><STRONG>@​​​dp.temporary_view</STRONG></CODE></H3> <P>Before we persist anything, let's talk about <STRONG>Views</STRONG>. Not every step in your pipeline needs to be a physical table.</P> <P>The <CODE>@​​​dp.temporary_view</CODE> decorator allows you to define temporary views. These are perfect for structuring your code and making it more modular and readable. They don't cost anything in storage and are ephemeral. You can easily reference them within your pipeline. A view lives within its pipeline only. Temporary views are useful when you want to break a pipeline into smaller transformation steps without creating extra physical tables. The results are then persisted downstream into Streaming Tables or Materialized Views.</P> <P>Here's how you'd typically create a reusable transformation. Notice how the SDP version wraps the same logic in a decorator, turning it into a named, referenceable pipeline object:</P> <TABLE border="1" width="100%" cellspacing="0" cellpadding="6"> <TBODY> <TR> <TH style="font-size: 1.2em;" width="50%"><FONT size="3"><STRONG>&nbsp; &nbsp; &nbsp; &nbsp;&nbsp; &nbsp; &nbsp; &nbsp;Spark Imperative Code</STRONG></FONT></TH> <TH style="font-size: 1.2em;" width="50%"><FONT size="3"><STRONG>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;Spark Declarative Pipelines</STRONG></FONT></TH> </TR> <TR> <TD width="50%" style="vertical-align: top;"><LI-CODE lang="python">cleaned_sales_logs = ( spark.readStream.table("raw_logs") .filter("amount &gt; 0") .withColumnRenamed("ts", "event_timestamp") ) </LI-CODE></TD> <TD width="50%"><LI-CODE lang="python">@​dp.temporary_view def cleaned_sales_logs(): return ( spark.readStream.table("raw_logs") .filter("amount &gt; 0") .withColumnRenamed("ts", "event_timestamp") ) </LI-CODE></TD> </TR> </TBODY> </TABLE> <H3><STRONG>The Append Flow</STRONG></H3> <P>Imagine you have a Kafka topic or a directory of raw JSON files. Your goal is to land them into your Bronze layer as quickly as possible, perhaps with light, row-by-row transformations, like column formatting, adding a <CODE>CASE WHEN</CODE> statement, or joining with a static reference table. This is the classic <STRONG>Append Flow</STRONG>. Because the Append Flow is stateful, it automatically tracks which data has already been processed. Only new rows are ingested on each run, and no manual bookkeeping required.</P> <P>A <STRONG>Streaming Table</STRONG> (<CODE>@​​​dp.table</CODE>) is the primary target for an Append flow.</P> <P>This is the direct equivalent of the Spark Structured Streaming <STRONG>writeStream</STRONG><STRONG> API</STRONG> with <CODE>.outputMode("append")</CODE>.</P> <P>When you use the <CODE>@​​​dp.table</CODE> decorator, SDP implicitly creates an Append Flow for you:</P> <LI-CODE lang="python">@​​dp.table def bronze_orders(): return spark.readStream.table("raw_kafka_orders")</LI-CODE> <P>However, you can also define the flow <STRONG>explicitly</STRONG> if you want more control over the source-to-target relationship.&nbsp;<SPAN>This is helpful when you have multiple flows targeting the same Streaming Table:</SPAN></P> <LI-CODE lang="python"># First, declare the target destination dp.create_streaming_table("bronze_orders") # Then, explicitly define the flow 1 that hydrates it @dp.append_flow(target="bronze_orders") def flow_cleaned_sales_logs(): # This refers to our .temporary_view above! return dp.read_stream("cleaned_sales_logs") # Define the schema for Kafka JSON messages # order_schema = StructType([StructField("order_id", StringType()), ...]) # Define a second flow that hydrates from a Kafka topic @dp.append_flow(target="bronze_orders") def append_kafka_orders(): return ( spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "&lt;broker-host&gt;:9092") .option("subscribe", "orders_topic") .option("startingOffsets", "latest") .load() .select(F.from_json(F.col("value").cast("string"), order_schema).alias("data")) .select("data.*") ) </LI-CODE> <P><SPAN>What if your Append Flow needs to write to an external database or Kafka topic instead of a table?&nbsp; For arbitrary sinks, there are the <CODE>dp.create_sink()</CODE> function and the <CODE>@​dp.foreach_batch_sink()</CODE> decorator. We will deep dive into custom sinks in a future post.</SPAN></P> <H3><STRONG>The Append Once Flow</STRONG></H3> <P>I know what you're thinking: <EM>"I really like this streaming approach, but I can't move to it. All my old data is sitting in a legacy table, and my Kafka topic retention is only 7 days. If I start a stream now, I lose my history. How do I bridge the gap?"</EM></P> <P>This is the perfect use case for the <STRONG>Append Once Flow</STRONG>. You can use a one-time batch flow to hydrate your table with legacy data, and then let the Append flow take over for the fresh data.</P> <LI-CODE lang="python"># 1. Declare the target table dp.create_streaming_table("bronze_orders") # 2. THE BACKFILL: Fill the table once from your legacy Parquet data @​dp.append_flow(target="bronze_orders", once=True) def backfill_historical(): return ( spark.read.format("parquet") .load("/Volumes/catalog/schema/historical_data") ) # 3. THE STREAM: Let the real-time flow handle everything moving forward @​dp.append_flow(target="bronze_orders") def stream_cleaned_sales_logs(): return spark.readStream.table("cleaned_sales_logs")</LI-CODE> <H4><STRONG>The AUTO CDC Flow</STRONG></H4> <P>What if your source isn't just a list of events, but a stream of database changes? You have updates and deletes, and you need your target table to reflect the <STRONG>current state</STRONG>. This is the classic Slowly Changing Dimension (SCD) Type 1 or Type 2 scenario.</P> <P>In the traditional approach, this meant writing a massive <CODE>MERGE INTO</CODE> block inside a <CODE>foreachBatch</CODE>. Worse, you had to repeat that same boilerplate for every single pipeline you built.</P> <P>In SDP, we replace that with the <STRONG>AUTO CDC Flow</STRONG>. And what is the physical target of an AUTO CDC Flow? It's the Streaming Table.</P> <P>To use it, you first declare your target as a streaming table, and then use the function <CODE>dp.create_auto_cdc_flow()</CODE>.</P> <TABLE border="1" width="100%" cellspacing="0" cellpadding="6"> <TBODY> <TR> <TH style="font-size: 1.2em;" width="50%" height="33px"><FONT size="3">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Spark Imperative code</FONT></TH> <TH style="font-size: 1.2em;" width="50%" height="33px"><FONT size="3">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Spark Declarative Pipelines</FONT></TH> </TR> <TR> <TD width="50%" height="765px"><LI-CODE lang="python">from pyspark.sql import DataFrame from pyspark.sql.functions import col, struct, max_by def upsert_to_delta(micro_batch_df, batch_id): # Deduplicate within the micro-batch to get the latest record per key (micro_batch_df .groupBy("key") .agg(max_by(struct("*"), col("ts")).alias("row")) .select("row.*") .createOrReplaceTempView("updates")) # Execute the MERGE logic # Note: Replace the ellipses with your actual column mapping spark.sql(""" MERGE INTO cdc_data_raw t USING updates s ON s.key = t.key WHEN MATCHED AND s.is_deleted = true THEN UPDATE SET DELETED_AT = now() WHEN MATCHED THEN UPDATE SET A = CASE WHEN s.ts &gt; t.ts THEN s.a ELSE t.a END, B = CASE WHEN s.ts &gt; t.ts THEN s.b ELSE t.b END, -- ... for every column ... ts = CASE WHEN s.ts &gt; t.ts THEN s.ts ELSE t.ts END WHEN NOT MATCHED THEN INSERT * """) # Start the stream (cdcData.writeStream .foreachBatch(upsert_to_delta) .outputMode("append") .start()) </LI-CODE></TD> <TD width="50%" height="765px" style="vertical-align: top;"><LI-CODE lang="python">from pyspark import pipelines as dp from pyspark.sql.functions import col, expr @​dp.view def users(): return ( spark.readStream .table("cdc_data.users") ) dp.create_streaming_table("target") dp.create_auto_cdc_flow( target = "target", source = "users", keys = ["key"], sequence_by = col("ts"), apply_as_deletes = expr("is_deleted = True") ) </LI-CODE></TD> </TR> </TBODY> </TABLE> <H3><STRONG>Materialized Views</STRONG></H3> <P>Once your data is in Silver, you usually want to aggregate it for Gold in order to calculate daily revenue or active users. This is where <STRONG>Materialized Views</STRONG> (<CODE>@​dp.materialized_view</CODE>)&nbsp;<SPAN>come into the picture.</SPAN></P> <P><SPAN>A Materialized View is a special pipeline object which stores results of a query.</SPAN></P> <P>Think of a Materialized View as a query acceleration mechanism. It persists the result of your query and <STRONG>automatically updates</STRONG> it incrementally whenever the source changes. T<SPAN>he framework </SPAN>appends, upserts, and deletes in order to reflect the current truth.</P> <P>To solve this in traditional ETL, you usually have to choose between two difficult paths:</P> <UL> <LI><STRONG>The Recompute Path:</STRONG> Whenever an update happens, you re-run the aggregation for that entire chunk of data (like a whole day or month). This gets more complex when handling <STRONG>late-arriving data</STRONG>.</LI> <LI><STRONG>The Watermark Path:</STRONG> You use watermarks to define a time window. The tradeoff? Any data arriving after the watermark is dropped and never considered.</LI> </UL> <P>Materialized Views solve this by acting as a <STRONG>"State" Mirror</STRONG>. They persist the query result and automatically update it incrementally. If data arrives late, SDP knows how to revisit that state and update the summary without you writing a single line of merge logic:</P> <TABLE border="1" width="100%" cellspacing="0" cellpadding="6"> <TBODY> <TR> <TH style="font-size: 1.2em;" width="50%"><FONT size="3">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Spark Imperative Code</FONT></TH> <TH style="font-size: 1.2em;" width="50%"><FONT size="3">&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;Spark Declarative Pipelines</FONT></TH> </TR> <TR> <TD width="50%"><LI-CODE lang="python">def update_aggs(microBatchDF, batchId): # 1. Identify affected dates distinct_dates = microBatchDF .select("order_date").distinct().collect() date_list = [row.order_date for row in distinct_dates] if len(date_list) &gt; 0: # 2. Recalculate chunks from source new_aggs = (spark.read .table("silver_orders") .filter(F.col("order_date") .isin(date_list)) .groupBy("order_date") .agg(F.sum("amount") .alias("total_revenue"))) # 3. Sink: Merge into target new_aggs.createOrReplaceTempView("updates") spark.sql(""" MERGE INTO daily_sales t USING updates s ON s.order_date = t.order_date WHEN MATCHED THEN UPDATE SET t.total_revenue = s.total_revenue WHEN NOT MATCHED THEN INSERT * """) (spark.readStream.table("silver_orders").writeStream.foreachBatch(update_aggregates).start()) </LI-CODE></TD> <TD width="50%" style="vertical-align: top;"><LI-CODE lang="python">@​dp.materialized_view def daily_sales(): return (spark.read .table("silver_orders") .groupBy("order_date") .agg(F.sum("amount").alias("total_revenue"))) </LI-CODE></TD> </TR> </TBODY> </TABLE> <P>Let's clarify the difference. You might be wondering: <EM>"Why can't I just use an Append Flow and a Streaming Table for my aggregations?"</EM></P> <P>An <STRONG>Append Flow</STRONG> never looks back on what has already been processed. It only moves forward, appending new rows as they arrive. Because it never revisits past data, running a standard <CODE>SUM</CODE> or <CODE>GROUP BY</CODE> over an <STRONG>Append Flow</STRONG> would require the engine to maintain an ever-growing state of all previously seen rows, with no natural point at which to finalize the calculation.</P> <P>To do aggregations, we need a <STRONG>finite table -&nbsp;</STRONG>a specific "state" or "snapshot" that represents the current truth. <STRONG>This is exactly why we use </STRONG><CODE><STRONG>spark.read</STRONG></CODE><STRONG> instead of </STRONG><CODE><STRONG>spark.readStream</STRONG></CODE><STRONG> inside a Materialized View.</STRONG> By using <CODE>read</CODE>, we tell SDP to treat the source as a full dataset to be queried, which the framework then intelligently optimizes to update incrementally behind the scenes.</P> <H3><STRONG>Final Decision Tree</STRONG></H3> <TABLE style="border-collapse: collapse; font-size: 1em;" border="1" width="100%" cellspacing="0"> <TBODY> <TR> <TH style="padding: 12px 16px; font-size: 1.15em; text-align: left;" width="45%"><FONT size="3">If you need...</FONT></TH> <TH style="padding: 12px 16px; font-size: 1.15em; text-align: left;" width="30%"><FONT size="3">Use this</FONT></TH> <TH style="padding: 12px 16px; font-size: 1.15em; text-align: left;" width="25%"><FONT size="3">Flow Type</FONT></TH> </TR> <TR> <TD style="padding: 10px 16px; vertical-align: middle;">To land raw data from a streaming append-only source</TD> <TD style="padding: 10px 16px; vertical-align: middle;"><CODE><STRONG>@​dp.table</STRONG></CODE></TD> <TD style="padding: 10px 16px; vertical-align: middle;">Append Flow</TD> </TR> <TR> <TD style="padding: 10px 16px; vertical-align: middle;">To process CDC data</TD> <TD style="padding: 10px 16px; vertical-align: middle;"><CODE><STRONG>dp.create_auto_cdc_flow</STRONG></CODE></TD> <TD style="padding: 10px 16px; vertical-align: middle;">AUTO CDC Flow</TD> </TR> <TR> <TD style="padding: 10px 16px; vertical-align: middle;">A one-time backfill</TD> <TD style="padding: 10px 16px; vertical-align: middle;"><CODE><STRONG>@​​dp.append_flow(once=True)</STRONG></CODE></TD> <TD style="padding: 10px 16px; vertical-align: middle;">Append Once Flow</TD> </TR> <TR> <TD style="padding: 10px 16px; vertical-align: middle;">Aggregations / Current Truth</TD> <TD style="padding: 10px 16px; vertical-align: middle;"><CODE><STRONG>@​​​dp.materialized_view</STRONG></CODE></TD> <TD style="padding: 10px 16px; vertical-align: middle;">N/A*</TD> </TR> <TR> <TD style="padding: 10px 16px; vertical-align: middle;">Code refactoring</TD> <TD style="padding: 10px 16px; vertical-align: middle;"><CODE><STRONG>@​​​dp.temporary_view</STRONG></CODE></TD> <TD style="padding: 10px 16px; vertical-align: middle;">N/A*</TD> </TR> </TBODY> </TABLE> <P><EM>* Remember, </EM><EM>Materialized Views and temporary views are pipeline objects rather than flows. Flows define how data moves into a destination, while these objects define how data is stored or structured within the pipeline.</EM></P> <H3><STRONG>Key Takeaway</STRONG></H3> <P>Flows are the core primitive of SDP: stateful, incremental computations that track what has been processed, so you only need to decide how data lands, and the framework handles the rest. Streaming Tables and Materialized Views are targets of the flows. Altogether, they intelligently handle technical complexity, allowing you to focus entirely on the business logic.</P> <H3><STRONG>Conclusion</STRONG></H3> <P>Today, we covered how Flows land data into Streaming Tables and how Materialized Views help accelerate queries. In our next post, we’ll look at the <STRONG>Consumer</STRONG> side: how to read from these objects, the nuances between <CODE>spark.read</CODE> and <CODE>spark.readStream</CODE>, and how to choose the right one for your specific scenario. Stay tuned!</P> Fri, 17 Apr 2026 13:15:43 GMT aleksandra_ch 2026-04-17T13:15:43Z https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/ba-p/151728 <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="Gemini_Generated_Image_9ejwob9ejwob9ejw.png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/24275i72A055CFA248C22D/image-size/large?v=v2&amp;px=999" role="button" title="Gemini_Generated_Image_9ejwob9ejwob9ejw.png" alt="Gemini_Generated_Image_9ejwob9ejwob9ejw.png" /></span></P> Fri, 17 Apr 2026 13:15:43 GMT https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/ba-p/151728 aleksandra_ch 2026-04-17T13:15:43Z https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/bc-p/152223#M966 <P>Nice summary!</P> Thu, 26 Mar 2026 21:06:13 GMT https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/bc-p/152223#M966 Hubert-Dudek 2026-03-26T21:06:13Z https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/bc-p/153862#M988 <P>Really enjoyed this breakdown especially how you positioned streaming tables for ingestion and materialized views for transformation. It makes designing real-time pipelines much more intuitive for SQL users.</P><P>One question: in large-scale environments with thousands of tables, how do you decide when to rely purely on streaming tables vs introducing materialized views, considering both cost and latency trade offs?</P> Thu, 09 Apr 2026 10:03:24 GMT https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/bc-p/153862#M988 antoalphi 2026-04-09T10:03:24Z https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/bc-p/153908#M990 <P>Hi&nbsp;<a href="https://community.databricks.com/t5/user/viewprofilepage/user-id/181633">@antoalphi</a>&nbsp;,</P> <P>Thanks for the feedback!</P> <P>The choice between <STRONG data-index-in-node="44" data-path-to-node="9">Streaming Tables (ST)</STRONG> and <STRONG data-index-in-node="70" data-path-to-node="9">Materialized Views (MV)</STRONG> really comes down to your required processing semantics:</P> <UL> <LI>If a row from the source needs to be processed only once / the logic never looks back at historical data -&gt; go for a Streaming Table</LI> <LI>If your logic requires complex aggregations, joins, or updates to existing records -&gt;&nbsp; you have to go for a Materialized View</LI> </UL> <P>For example, Streaming Table is typically not suitable for aggregations and joins, just because the semantics need to look back to existing records in the table in order to perform those. There are exceptions which I plan to discuss in my next post.</P> <P>If you need to ingest thousands of sources into UC tables in an incremental way with lightweight row-by-row processing - Streaming Tables is a good choice.</P> <P>If you have to create aggregations / joins on those tables - you have to go with Materialized Views on top.&nbsp;</P> <P>All in all, wherever you have thousands of tables or just few of them - the choice really depends on the business logic.</P> <P>A note regarding large-scale environments:</P> <UL> <LI>One pipeline can contain up to 1000 datasets (including Materialized Views, Streaming Tables, and Temporary Views). At a large scale, pipeline organization is critical since all datasets within a single pipeline share the same compute resources and lifecycle. If one table hangs, it can impact the entire group.</LI> </UL> <P>If you have a specific pipeline diagram or a list of requirements, feel free to share it. I’d be happy to provide a more tailored vision for the setup!</P> Thu, 09 Apr 2026 12:35:05 GMT https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/bc-p/153908#M990 aleksandra_ch 2026-04-09T12:35:05Z https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/bc-p/156982#M1065 <P>I just want to double down that these articles are super helpful so the time put into them is well worth it! Especially for data engineers and architects supporting databricks.&nbsp;</P> Fri, 15 May 2026 12:17:53 GMT https://community.databricks.com/t5/technical-blog/sdp-how-to-series-part-2-how-to-master-streaming-tables-and/bc-p/156982#M1065 Randy_meacham 2026-05-15T12:17:53Z