Transformation: chained batch notebooks to Lakeflow Declarative Pipelines

#What you get

• Dependencies inferred from the code — no hand-ordered tasks.
• Incremental processing for streaming tables and materialized views.
• Data quality rules with metrics (expectations).
• Retries, lineage and an event log out of the box.

#1. Map notebooks to tables

#2. Write the pipeline (Python)

import dlt
from pyspark.sql import functions as F

@dlt.table(comment="Raw orders")
def orders_bronze():
    return (spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("/Volumes/main/ingest/raw/orders/"))

@dlt.table
@dlt.expect_or_drop("valid_amount", "amount >= 0")
@dlt.expect("has_customer", "customer_id IS NOT NULL")
def orders_clean():
    return dlt.read_stream("orders_bronze").withColumn("order_date", F.to_date("order_ts"))

dlt.create_streaming_table("orders_silver")
dlt.apply_changes(target="orders_silver", source="orders_clean",
                  keys=["order_id"], sequence_by="order_ts", stored_as_scd_type=1)

@dlt.table
def daily_sales_gold():
    return (dlt.read("orders_silver").groupBy("order_date")
            .agg(F.sum("amount").alias("revenue"), F.count("*").alias("orders")))

Module and function names are being renamed in newer releases (pyspark.pipelines, create_auto_cdc_flow); check your release's docs — the concepts are the same.

#3. Configure

Target catalog/schema, serverless or classic compute, triggered (scheduled) or continuous mode, and the pipeline's run-as identity. Put it in a bundle so it deploys like everything else.

#4. Migrate existing tables

Pipeline-managed tables are created by the pipeline. Options: let the pipeline backfill from bronze/raw (simplest), or for large silver tables do a one-time load and then let the pipeline continue incrementally. Point downstream consumers at the new table names, or keep old names with views.

#5. Monitor

Expectation results and flow progress are in the event log:

SELECT timestamp, details:flow_progress.data_quality.expectations
FROM event_log(TABLE(main.sales.orders_silver))
WHERE event_type = 'flow_progress' ORDER BY timestamp DESC LIMIT 20;

Tip

Start with one notebook chain that is painful today (frequent ordering bugs or reprocessing). Wins there make the case for the rest.

#Checklist

• Notebook chain mapped to tables/views
• Checks converted to expectations
• CDC via AUTO CDC / apply_changes
• Pipeline in a bundle with run-as identity
• Consumers repointed; notebook jobs disabled