https://community.databricks.com/t5/technical-blog/declarative-pipelines-meets-foreachbatch-custom-streaming-for/ba-p/142033 <P><SPAN>If this is the first time you’re hearing about Sinks in Lakeflow Declarative Pipelines (LDP), we highly recommend reading </SPAN><A href="https://www.databricks.com/blog/introducing-dlt-sink-api-write-pipelines-kafka-and-external-delta-tables" target="_blank" rel="noopener"><SPAN>Introducing the SDP Sink API</SPAN></A><SPAN>, which explores the recently launched Sinks API, offering the ability to seamlessly sink data to external systems, such as event streams and Delta tables.</SPAN></P> <H1><SPAN>Introduction</SPAN></H1> <P><SPAN>LDP provides a declarative framework designed to simplify and streamline pipeline development, whether you're processing real-time data with Streaming Tables or aggregating data efficiently using Materialized Views. The introduction of the Sinks API further enhances functionality by enabling seamless integration with external systems.</SPAN></P> <P><SPAN>This blog delves into the support for </SPAN><A href="https://docs.databricks.com/aws/en/ldp/for-each-batch" target="_blank" rel="noopener"><SPAN>foreachBatch in LDP</SPAN></A><SPAN>, a feature from Apache Spark Structured Streaming that facilitates arbitrary batch operations on streaming data, enabling complex transformations and writes to various data sinks.&nbsp;</SPAN></P> <H1><SPAN>From Clicks to Insights: Clickstream Data in Action</SPAN></H1> <P><SPAN>For simplicity and reproducibility, this blog post uses the same data source as the </SPAN><A href="https://www.databricks.com/blog/introducing-dlt-sink-api-write-pipelines-kafka-and-external-delta-tables" target="_blank" rel="noopener"><SPAN>Sinks API blog</SPAN></A><SPAN>: the clickstream dataset from Databricks Datasets. You can think of this data as a map of the internet, illustrating where visitors navigate and how they move from one page to another. To keep things straightforward, we'll assume a one-to-one relationship between page IDs and titles, with titles remaining static. Using this dataset, we'll explore three key use cases:</SPAN></P> <OL> <LI style="font-weight: 400;" aria-level="1"><SPAN>The Comprehensive Visit Aggregator: Let's dive deeper into total page visits by aggregating them regardless of origin to uncover popular destinations.&nbsp;</SPAN></LI> <LI style="font-weight: 400;" aria-level="1"><SPAN>The Million-Click Club: This is where things get interesting. We'll implement a monitoring system that records data in SQL Server whenever page visits exceed the one-million mark within a single time unit.</SPAN></LI> <LI style="font-weight: 400;" aria-level="1"><SPAN>The New York City Tracker: In the final step, we'll actively monitor visits to the "New_York_City" page. Whenever the incoming stream captures such an event, we’ll archive the data as Parquet records for efficient storage and analysis.</SPAN></LI> </OL> <P><SPAN>By the end of this post, you’ll see how LDP can elevate your data processing workflow. We’ll walk through how to efficiently handle streaming data in real time, seamlessly merge it with existing Delta tables, and dynamically route it to multiple destinations!</SPAN></P> <H1><SPAN>Code Walkthrough</SPAN></H1> <H2><SPAN>Preparing Clickstream Data</SPAN></H2> <P><SPAN>We’ll start by importing the necessary modules and creating a Streaming Table to ingest Wikipedia clickstream data in JSON format. Using the </SPAN><STRONG><a href="https://community.databricks.com/t5/user/viewprofilepage/user-id/25059">@DP</a>.table</STRONG><SPAN> API, we’ll stream data directly from the source location. You can also configure the pipeline to stream data from any source supported by Autoloader, including event streams. It’s as simple as switching the format and providing the necessary authorization.</SPAN></P> <LI-CODE lang="python">from pyspark import pipelines as dp from pyspark.sql.functions import * from pyspark.sql.types import * from delta.tables import DeltaTable json_path = "/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/" @dp.table( comment="The raw wikipedia click stream dataset, ingested from /databricks-datasets.", table_properties={ "quality": "bronze" } ) def clickstream_raw(): return ( spark.readStream.format("cloudFiles").option("cloudFiles.format", "json").option("inferSchema", "true").load(json_path) )</LI-CODE> <P><SPAN>With our clickstream data now streaming into the raw table, the next step is to refine it by applying data quality constraints and transformations. This includes standardizing data types and filtering out invalid records. We achieve this by leveraging LDP’s native expectations to enforce data quality and maintain pipeline integrity.</SPAN></P> <LI-CODE lang="python">@DP.table( comment="Wikipedia clickstream dataset with cleaned-up datatypes / column names and quality expectations.", table_properties={ "quality": "silver" } ) @dp.expect_or_drop("valid_current_page", "current_page_id IS NOT NULL AND current_page_title IS NOT NULL") @dp.expect_or_fail("valid_count", "click_count &gt; 0") def clickstream_clean(): return ( dp.readStream("clickstream_raw") .withColumn("current_page_id", expr("CAST(curr_id AS INT)")) .withColumn("click_count", expr("CAST(n AS INT)")) .withColumn("previous_page_id", expr("CAST(prev_id AS INT)")) .withColumnRenamed("curr_title", "current_page_title") .withColumnRenamed("prev_title", "previous_page_title") .select("current_page_id", "current_page_title", "click_count", "previous_page_id", "previous_page_title") )</LI-CODE> <H2><SPAN>foreachBatch - Writing to multiple sinks</SPAN></H2> <P><SPAN>With our cleaned data in a Streaming Table, we can now implement our three use cases using </SPAN><STRONG><a href="https://community.databricks.com/t5/user/viewprofilepage/user-id/25059">@DP</a>.foreach_batch_sink</STRONG><SPAN>, LDP's new sink that brings foreachBatch-style processing into LDP. This allows streaming data to be handled as microbatches while integrating with regular DataFrame APIs.&nbsp;</SPAN></P> <P><SPAN>In Spark Structured Streaming, data is processed as a series of micro-batches. Each micro-batch contains a specific set of rows, represented as a DataFrame. During processing, this DataFrame, along with the corresponding </SPAN><STRONG>batchId</STRONG><SPAN>, is passed to the function registered with the sink. We will use the DataFrame </SPAN><STRONG>df</STRONG><SPAN>, which contains the rows from a micro-batch, to implement our use cases.</SPAN></P> <P><SPAN>In the first use case, we aggregate total page visits at the sink across all origins. Since each micro-batch may contain multiple rows for the same </SPAN><STRONG>current_page_id</STRONG><SPAN>, we first aggregate the data at the micro-batch level. If an existing value is present in the sink, we update it accordingly; otherwise, we insert a new record.</SPAN></P> <P><SPAN>The second use case filters the </SPAN><STRONG>click_count</STRONG><SPAN> to identify high-traffic pages within each time interval and appends this data to a table called </SPAN><STRONG>dbo.high_traffic_pages</STRONG><SPAN> in a SQL Server database for consumption by external systems. We are leveraging Databricks Secrets to store the connection string and utilize that for authentication with SQL Server.</SPAN></P> <P><SPAN>For the final use case, we’ll apply a simple filter on the DataFrame to capture page visits to New York City and securely archive this data as parquet files to a volume in Unity Catalog. Access to the archived data can be managed and controlled directly through the Unity Catalog.</SPAN></P> <LI-CODE lang="python">@DP.foreach_batch_sink(name = "all_the_sinks") def foreachBatchFunc(df, batchId): #Usecase 1: Aggregate page level click count at a Delta Sink agg_df = df.groupBy("current_page_id", "current_page_title").agg(sum("click_count").alias("click_count")) out = DeltaTable.forName(df.sparkSession, "harsha_pasala.default.clickstream_sink") out.alias("target") \ .merge(agg_df.alias("source"), "source.current_page_id = target.current_page_id") \ .whenMatchedUpdate( set = {"click_count": col("target.click_count") + col("source.click_count")} ) \ .whenNotMatchedInsert( values = { "target.current_page_id": "source.current_page_id", "target.current_page_title": "source.current_page_title", "target.click_count": "source.click_count" } ) \ .execute() #Usecase 2: High traffic pages to SQL Server. high_traffic_pages = df.filter(col("click_count") &gt; 1000000) high_traffic_pages.write \ .format("jdbc") \ .option("url", dbutils.secrets.get(scope="secret-lab", key="sql_connection_string")) \ .option("dbtable", "dbo.high_traffic_pages") \ .mode("append") \ .save() #Usecase 3: Monitor for a specific page and archive data. new_york_clicks = df.filter(col("current_page_title").like("New_York_City")) new_york_clicks.write.format("parquet").mode("append").save("/Volumes/path/")</LI-CODE> <H2><SPAN>Streaming data to sinks</SPAN></H2> <P><SPAN>Our data is ready, and the sink is in place - it is time to put it into action. Since our sinks modify data using the merge behaviour, we’ll use the </SPAN><STRONG>update_flow</STRONG><SPAN> API to direct data from the </SPAN><STRONG>clickstream_clean</STRONG><SPAN> Streaming Table to the </SPAN><STRONG>foreach_batch_sink</STRONG><SPAN> we created earlier, which is </SPAN><STRONG>all_the_sinks</STRONG><SPAN>. This ensures that data is efficiently processed and routed according to our defined use cases.</SPAN></P> <LI-CODE lang="python">@DP.update_flow( target="all_the_sinks", name="sink_flow" ) def read_data(): return ( dp.readStream("clickstream_clean") )</LI-CODE> <P><SPAN>Finally, by integrating clickstream data ingestion, processing with quality constraints, and multiple sinks within a LDP pipeline, we arrive at the following DAG from the pipeline run. While the DAG displays a single sink named </SPAN><SPAN>all_the_sinks</SPAN><SPAN>, it's important to note that this represents a </SPAN><STRONG>foreach_batch_sink</STRONG><SPAN>, that effectively dispatches data to all three configured downstream targets.</SPAN></P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="unnamed (1).png" style="width: 999px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/22351iBE2CB27863C24CD8/image-size/large?v=v2&amp;px=999" role="button" title="unnamed (1).png" alt="unnamed (1).png" /></span></P> <P><SPAN>Over the past year, LDP has introduced numerous enhancements, with </SPAN><STRONG><a href="https://community.databricks.com/t5/user/viewprofilepage/user-id/25059">@DP</a>.foreach_batch_sink</STRONG><SPAN> being one of the latest additions. This feature enables advanced streaming capabilities while building on existing sinks and append flow APIs, allowing for a cleaner and more streamlined implementation. For more information about the API and sample implementations, please refer to the <A href="https://docs.databricks.com/aws/en/ldp/for-each-batch" target="_self">docs</A>.</SPAN></P> <H1><SPAN>Related Databricks Blogs:</SPAN></H1> <UL> <LI><SPAN><A href="https://www.databricks.com/blog/introducing-dlt-sink-api-write-pipelines-kafka-and-external-delta-tables" target="_self">Introducing the DLT Sink API: Write Pipelines to Kafka and External Delta Tables</A><BR /></SPAN></LI> <LI><SPAN><A href="https://www.databricks.com/discover/pages/getting-started-with-delta-live-tables" target="_self">Getting started with DLT</A><BR /></SPAN></LI> <LI><SPAN><A href="https://www.databricks.com/blog/cost-effective-incremental-etl-serverless-compute-delta-live-tables-pipelines" target="_self">Cost-effective, incremental ETL with serverless compute for Delta Live Tables pipelines</A><BR /></SPAN></LI> <LI><SPAN><A href="https://www.databricks.com/blog/2022/08/09/low-latency-streaming-data-pipelines-with-delta-live-tables-and-apache-kafka.html" target="_self">Low-latency Streaming Data Pipelines with Delta Live Tables and Apache Kafka</A><BR /></SPAN></LI> </UL> Thu, 08 Jan 2026 14:22:41 GMT harshapasala 2026-01-08T14:22:41Z https://community.databricks.com/t5/technical-blog/declarative-pipelines-meets-foreachbatch-custom-streaming-for/ba-p/142033 <P data-start="365" data-end="811"><STRONG data-start="365" data-end="479">What if a single Declarative Pipeline could apply multiple downstream actions to each micro-batch?</STRONG><BR data-start="479" data-end="482" />With the new <STRONG data-start="495" data-end="523"><CODE data-start="497" data-end="521">@dp.foreach_batch_sink</CODE></STRONG> in Lakeflow Declarative Pipelines, you can bring the power of Spark’s <CODE data-start="594" data-end="608">foreachBatch</CODE> into a declarative workflow. This post walks through a real world clickstream example, showing how each micro-batch can be sequentially aggregated, monitored, and archived using standard DataFrame APIs.</P> Thu, 08 Jan 2026 14:22:41 GMT https://community.databricks.com/t5/technical-blog/declarative-pipelines-meets-foreachbatch-custom-streaming-for/ba-p/142033 harshapasala 2026-01-08T14:22:41Z https://community.databricks.com/t5/technical-blog/declarative-pipelines-meets-foreachbatch-custom-streaming-for/bc-p/143361#M874 <P>Great writeup&nbsp;<a href="https://community.databricks.com/t5/user/viewprofilepage/user-id/46901">@harshapasala</a>&nbsp;,&nbsp;</P> <P class="p1">The @dp.foreach_batch_sink decorator is a great example of what I love about modern Databricks pipelines: declarative on the surface, but with real imperative muscle underneath. One sink, three destinations, and suddenly a whole new design space opens up. You can merge into Delta, route high-traffic events to SQL Server, and archive targeted data into Unity Catalog—all without breaking the declarative flow.</P> <P class="p1">This is exactly the kind of flexibility streaming teams need when real-world requirements refuse to stay simple. For anyone juggling multi-destination streaming workflows, LDP just became a lot more compelling—and honestly, a lot more fun to work with.</P> <P class="p1">Keep sharing your thoughts.</P> <P class="p1">Regards, Louis.</P> Thu, 08 Jan 2026 14:28:57 GMT https://community.databricks.com/t5/technical-blog/declarative-pipelines-meets-foreachbatch-custom-streaming-for/bc-p/143361#M874 Louis_Frolio 2026-01-08T14:28:57Z