Overview
This notebook demonstrates how to build an incremental data transformation pipeline using Spark Structured Streaming, Delta Lake, and the foreachBatch API. The pipeline reads streaming data from the Bronze Delta table, applies data quality transformations and standardization, and incrementally loads the cleansed data into the Silver layer using Delta MERGE (upsert) operations.
The notebook performs the following tasks:
Configures notebook parameters using Databricks widgets.
Reads incremental data from the brz_order_items Bronze Delta table as a streaming source.
Removes duplicate records based on the business key (order_id, item_seq).
Cleans and standardizes data by converting data types, normalizing text values, and formatting numeric fields.
Adds processing metadata, including the processing timestamp.
Performs incremental upsert (MERGE) operations into the slv_order_items Delta table using the foreachBatch API.
Automatically creates the Silver table if it does not exist and enables Delta Change Data Feed (CDF).
Uses checkpointing to provide fault tolerance and exactly-once processing.
Validates the successful completion of the streaming transformation and loading process.
The resulting Silver table contains cleansed, standardized, and deduplicated data, providing a reliable and analytics-ready foundation for downstream business reporting and Gold layer data models.
from pyspark.sql.types import StringType, IntegerType, DateType, BooleanType import pyspark.sql.functions as F from delta.tables import DeltaTable
Configure Notebook Parameters
This step creates Databricks notebook widgets to define configurable parameters used throughout the notebook. These parameters allow the notebook to be reused across different environments without modifying the source code.
Purpose:
Define the Unity Catalog name.
Specify the Azure Storage Account.
Specify the Azure Blob Storage container.
Improve notebook reusability by externalizing configuration values.
Enable easy execution across development, testing, and production environments.
dbutils.widgets.text("catalog_name", "ecommerce", "Catalog Name")
dbutils.widgets.text("storage_account_name", "stecommercedeveastus001", "Storage Account Name")
dbutils.widgets.text("container_name", "ecomm-raw-data", "Container Name")
Retrieve Notebook Parameters
This step retrieves the values entered through the Databricks notebook widgets and stores them in Python variables. These configuration values are then used throughout the notebook to access the appropriate Unity Catalog, Azure Storage Account, and Blob Storage container. The retrieved values are printed to verify that the notebook has been configured correctly.
Purpose:
Retrieve the configured notebook parameters.
Store configuration values in reusable Python variables.
Validate the retrieved parameters before executing the notebook.
Support dynamic and environment-independent notebook execution.
catalog_name = dbutils.widgets.get("catalog_name")
storage_account_name = dbutils.widgets.get("storage_account_name")
container_name = dbutils.widgets.get("container_name")
Read Streaming Data from the Bronze Layer
This step creates a Structured Streaming DataFrame by reading data from the Bronze Delta table. Instead of loading the entire table as a batch, the streaming DataFrame continuously monitors the Bronze table for newly ingested records and processes only the incremental changes.
The streaming DataFrame serves as the input for the Silver layer transformation pipeline, allowing new records to be processed as they become available while supporting scalable and incremental data processing.
Purpose
Read incremental data from the Bronze Delta table.
Create a streaming DataFrame for downstream transformations.
Process only newly available records instead of reprocessing the entire dataset.
Provide the source data for the Silver layer ETL pipeline.
Support efficient and scalable incremental data processing using Spark Structured Streaming.
df = spark.readStream \
.format(“delta”) \
.table(f”{catalog_name}.bronze.brz_order_items”)
Transform Data for the Silver Layer
This step applies a series of data cleansing and standardization transformations to the streaming DataFrame before it is loaded into the Silver Delta table. These transformations improve data quality, enforce consistent data types, remove duplicate records, standardize categorical values, and enrich the dataset with processing metadata.
The transformed data becomes suitable for downstream analytics, reporting, and further processing in the Gold layer.
Purpose
Remove duplicate order item records.
Standardize and convert data into appropriate data types.
Clean numeric fields by removing invalid characters.
Normalize text values for consistency.
Standardize channel names across the dataset.
Add processing timestamp for auditing and lineage.
Overall Transformation Flow
Raw Bronze Data
│
▼
Remove Duplicate Records
│
▼
Standardize Quantity Values
│
▼
Clean Unit Price
(Remove ‘$’ and Convert to Double)
│
▼
Clean Discount Percentage
(Remove ‘%’ and Convert to Double)
│
▼
Normalize Coupon Codes
(Lowercase and Trim Spaces)
│
▼
Standardize Channel Names
(web → Website, app → Mobile)
│
▼
Add Processing Timestamp
│
▼
Cleaned Silver Data
df = df.dropDuplicates([“order_id”, “item_seq”])
# Transformation: Convert ‘Two’ → 2 and cast to Integer
df = df.withColumn(
“quantity”,
F.when(F.col(“quantity”) == “Two”, 2).otherwise(F.col(“quantity”)).cast(“int”)
)
# Transformation : Remove any ‘$’ or other symbols from unit_price, keep only numeric
df = df.withColumn(
“unit_price”,
F.regexp_replace(“unit_price”, “[$]”, “”).cast(“double”)
)
# Transformation : Remove ‘%’ from discount_pct and cast to double
df = df.withColumn(
“discount_pct”,
F.regexp_replace(“discount_pct”, “%”, “”).cast(“double”)
)
# Transformation : coupon code processing (convert to lower)
df = df.withColumn(
“coupon_code”, F.lower(F.trim(F.col(“coupon_code”)))
)
# Transformation : channel processing
df = df.withColumn(
“channel”,
F.when(F.col(“channel”) == “web”, “Website”)
.when(F.col(“channel”) == “app”, “Mobile”)
.otherwise(F.col(“channel”)),
)
#Transformation : Add processed time
df = df.withColumn(
“processed_time”, F.current_timestamp()
)
silver_checkpoint_path = f”abfss://{container_name}@{storage_account_name}.dfs.core.windows.net/checkpoint/silver/fact_order_items/”
print(silver_checkpoint_path)
Upsert Data into the Silver Layer
This function is used by the Structured Streaming pipeline to process each incoming micro-batch from the Bronze layer and load it into the Silver Delta table. Instead of appending records, it performs a Delta MERGE (upsert) operation to ensure that existing records are updated and new records are inserted.
During the first execution, if the Silver table does not already exist, the function creates it from the incoming micro-batch and enables Delta Change Data Feed (CDF). For all subsequent micro-batches, the function uses the business key (order_id and item_seq) to identify matching records and performs an incremental merge into the Silver table.
Purpose
Create the Silver Delta table if it does not already exist.
Enable Delta Change Data Feed (CDF) for downstream incremental processing.
Perform incremental upsert (MERGE) operations instead of appending data.
Update existing records using the business key.
Insert new records that do not already exist in the Silver table.
Maintain a clean, deduplicated, and up-to-date Silver layer.
Function Explanation
Code Why it is Used
def upsert_to_silver(microBatchDF, batchId): Defines a function that Spark automatically calls for each streaming micro-batch. Spark passes the current batch as microBatchDF and assigns a unique batchId.
table_name = f”{catalog_name}.silver.slv_order_items” Defines the fully qualified name of the target Silver Delta table.
spark.catalog.tableExists(table_name) Checks whether the Silver table already exists in the Unity Catalog.
if not spark.catalog.tableExists(table_name): Executes the initialization logic only during the first run when the Silver table has not yet been created.
microBatchDF.write.format(“delta”).mode(“overwrite”).saveAsTable(table_name) Creates the Silver Delta table using the data from the first micro-batch.
ALTER TABLE … SET TBLPROPERTIES (delta.enableChangeDataFeed = true) Enables Delta Change Data Feed (CDF) to capture inserts, updates, and deletes for downstream incremental processing.
DeltaTable.forName(spark, table_name) Creates a DeltaTable object that provides programmatic access to Delta Lake operations such as MERGE, UPDATE, and DELETE.
.alias(“silver_table”) Assigns an alias to the target Silver table for easier reference within the MERGE statement.
microBatchDF.alias(“batch_table”) Assigns an alias to the incoming micro-batch DataFrame to simplify column references during the MERGE operation.
.merge(…) Performs a Delta MERGE operation that compares incoming records with existing records in the Silver table.
“silver_table.order_id = batch_table.order_id AND silver_table.item_seq = batch_table.item_seq” Defines the matching condition (business key) used to determine whether an incoming record already exists in the Silver table.
.whenMatchedUpdateAll() Updates all columns of an existing record when a matching business key is found.
.whenNotMatchedInsertAll() Inserts the incoming record when no matching business key exists in the Silver table.
.execute() Executes the MERGE operation and commits the changes to the Silver Delta table.
Overall Processing Flow
Bronze Streaming Data
│
▼
Current Micro-Batch (microBatchDF)
│
▼
Does Silver Table Exist?
│
┌─────┴─────┐
│ │
No Yes
│ │
Create Table MERGE Data
Enable CDF │
▼
Match on (order_id, item_seq)
│
┌───────┴────────┐
│ │
Match No Match
│ │
Update Record Insert Record
│
▼
Silver Delta Table