Declarative Pipelines and Data Expectations

The previous lesson built three layers by hand: three CREATE TABLE AS statements, run in the right order, with quality checks left as an exercise. A declarative pipeline takes the same queries and manages the ordering, the incremental processing and the quality reporting for you.

The same medallion, declared

import dlt
from pyspark.sql.functions import col, lower, trim, current_timestamp

@dlt.table(
    name="bronze_orders",
    comment="Raw orders exactly as they arrived.",
    table_properties={"quality": "bronze"},
)
def bronze_orders():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("cloudFiles.schemaHints", "order_id bigint, amount decimal(10,2)")
        .option("header", "true")
        .load("/Volumes/bookshop/raw/landing/")
        .withColumn("_ingested_at", current_timestamp())
    )


@dlt.table(name="silver_orders", comment="Cleaned, typed, deduplicated orders.")
@dlt.expect("valid_amount", "amount > 0")
@dlt.expect_or_drop("valid_order_id", "order_id is not null")
@dlt.expect_or_fail("known_status", "status in ('completed','returned','refunded','pending')")
def silver_orders():
    return (
        dlt.read_stream("bronze_orders")
        .filter(col("_rescued_data").isNull())
        .select(
            "order_id",
            "customer_id",
            "ordered_at",
            lower(trim(col("status"))).alias("status"),
            "amount",
            "_ingested_at",
        )
    )


@dlt.table(name="gold_daily_revenue")
def gold_daily_revenue():
    return (
        dlt.read("silver_orders")
        .filter("status = 'completed'")
        .groupBy("ordered_at")
        .agg({"amount": "sum", "*": "count"})
        .withColumnRenamed("sum(amount)", "revenue")
        .withColumnRenamed("count(1)", "orders")
    )

Starting the pipeline:

Updating pipeline bookshop-medallion (update 8f2c1a44)

Setting up tables
  Creating BRONZE_ORDERS
  Creating SILVER_ORDERS
  Creating GOLD_DAILY_REVENUE

Running
  bronze_orders        STREAMING TABLE       COMPLETED   13932 rows written
  silver_orders        STREAMING TABLE       COMPLETED    9788 rows written
  gold_daily_revenue   MATERIALIZED VIEW     COMPLETED     184 rows written

Update 8f2c1a44 COMPLETED in 4m 12s

No ordering was declared. dlt.read_stream("bronze_orders") created the edge, exactly as ref does in dbt, and the pipeline built the graph from it.

The SQL form is equivalent:

create or refresh streaming table bronze_orders
as select *, current_timestamp() as _ingested_at
   from stream read_files('/Volumes/bookshop/raw/landing/', format => 'csv');

create or refresh streaming table silver_orders (
    constraint valid_amount   expect (amount > 0),
    constraint valid_order_id expect (order_id is not null) on violation drop row,
    constraint known_status   expect (status in ('completed','returned','refunded','pending'))
        on violation fail update
)
as select order_id, customer_id, ordered_at, lower(trim(status)) as status, amount, _ingested_at
   from stream(bronze_orders)
   where _rescued_data is null;

Streaming table or materialized view

Choosing a streaming table over a source whose rows get updated is the mistake to avoid: updates are not re-read, so the table silently drifts from its source. Bronze is nearly always streaming; gold is nearly always a materialized view.

Expectations

Three behaviours, chosen per rule:

Running
  silver_orders  STREAMING TABLE  COMPLETED  9788 rows written

  Data quality
  ┌──────────────────┬─────────┬──────────┬──────────────────┐
  │ expectation      │ passed  │ failed   │ action           │
  ├──────────────────┼─────────┼──────────┼──────────────────┤
  │ valid_amount     │    9694 │       94 │ warn             │
  │ valid_order_id   │    9788 │      412 │ drop             │
  │ known_status     │    9788 │        0 │ fail             │
  └──────────────────┴─────────┴──────────┴──────────────────┘

94 rows with a non-positive amount were kept and counted; 412 rows without an id were dropped. Nobody had to write a test — the rule sits on the table definition and reports every run.

An expect_or_fail violation stops everything:

  silver_orders  STREAMING TABLE  FAILED

  org.apache.spark.sql.streaming.StreamingQueryException:
  Flow 'silver_orders' failed to meet the expectation 'known_status'.
  1 row violated the expectation, and the pipeline is configured to fail on violation.

  gold_daily_revenue  MATERIALIZED VIEW  SKIPPED

Update 3d9c4b21 FAILED in 1m 08s

SKIPPED downstream, exactly like dbt build and like a task DAG — the gold table keeps yesterday’s correct data rather than being rebuilt from a broken silver.

Quality as a time series

Every expectation result lands in the event log:

select
    timestamp,
    details:flow_progress.data_quality.expectations[0].name::string   as expectation,
    details:flow_progress.data_quality.expectations[0].passed_records::int as passed,
    details:flow_progress.data_quality.expectations[0].failed_records::int as failed
from event_log(table(bookshop.pipelines.medallion))
where event_type = 'flow_progress'
  and details:flow_progress.data_quality is not null
order by timestamp desc
limit 4;

timestamp            expectation    passed  failed
-------------------  -------------  ------  ------
2026-09-09 06:00:12  valid_amount     9694      94
2026-09-08 06:00:09  valid_amount     9702      88
2026-09-07 06:00:14  valid_amount     9688      91
2026-09-06 06:00:11  valid_amount     4204     882

Four days of history in one query, and the fourth row stands out — 882 failures against a normal ~90. Charting this makes a data quality regression visible the morning it happens rather than at quarter end.

CDC without writing a MERGE

create or refresh streaming table silver_customers;

apply changes into live.silver_customers
from stream(bronze_customers_cdc)
keys (customer_id)
apply as delete when operation = 'DELETE'
sequence by sequence_num
columns * except (operation, sequence_num);

  silver_customers  STREAMING TABLE  COMPLETED  4120 rows written
    inserted: 3988   updated: 118   deleted: 14

Out-of-order changes are handled by sequence by — a late-arriving older record does not overwrite a newer one, which is the bug most hand-written MERGE statements have.

SCD type 2 is one clause more:

apply changes into live.silver_customers_history
from stream(bronze_customers_cdc)
keys (customer_id)
sequence by sequence_num
stored as scd type 2;

select customer_id, country_code, __START_AT, __END_AT
from bookshop.silver.silver_customers_history
where customer_id = 1 order by __START_AT;

customer_id  country_code  __START_AT  __END_AT
-----------  ------------  ----------  --------
          1  GB                    12        48
          1  NL                    48      NULL

The same slowly-changing-dimension history a dbt snapshot produces, maintained by the pipeline. AUTO CDC is the current name for this; APPLY CHANGES INTO remains valid.

Running it

Development mode is the one that saves time while iterating — the 4-minute cluster start is paid once rather than per attempt. Switching to production before scheduling is a step people forget, and it is what enables retries.

select timestamp, event_type, message
from event_log(table(bookshop.pipelines.medallion))
where event_type in ('update_progress', 'flow_progress')
order by timestamp desc limit 4;

timestamp            event_type       message
-------------------  ---------------  --------------------------------------------------
2026-09-09 06:04:24  update_progress  Update 8f2c1a44 is COMPLETED.
2026-09-09 06:04:20  flow_progress    Flow 'gold_daily_revenue' has COMPLETED.
2026-09-09 06:03:02  flow_progress    Flow 'silver_orders' has COMPLETED.
2026-09-09 06:00:12  flow_progress    Flow 'bronze_orders' has COMPLETED.

When to use it

Declarative pipelines are worth it when you want expectations, automatic dependency management and incremental processing without writing the plumbing. They are not the answer for arbitrary logic — training a model, calling an API, unloading a file — which belongs in a job with notebook tasks, the next lesson.

If your transformations already live in dbt, keep them there: dbt on Databricks supports streaming tables and materialized views as materializations, so you get most of this without maintaining two frameworks.

Practice

┌────────────────┬────────┬────────┬────────┐
│ expectation    │ passed │ failed │ action │
├────────────────┼────────┼────────┼────────┤
│ valid_order_id │   9788 │    412 │ drop   │
└────────────────┴────────┴────────┴────────┘

  silver_orders       STREAMING TABLE    FAILED
  gold_daily_revenue  MATERIALIZED VIEW  SKIPPED

select date(timestamp) as day,
       sum(details:flow_progress.data_quality.expectations[0].failed_records::int) as failed
from event_log(table(bookshop.pipelines.medallion))
where details:flow_progress.data_quality is not null
group by 1 order by 1 desc limit 5;

day         failed
----------  ------
2026-09-09      94
2026-09-08      88
2026-09-06     882

customer_id  country_code  __START_AT  __END_AT
-----------  ------------  ----------  --------
          1  GB                    12        48
          1  NL                    48      NULL

Next: jobs and workflows — scheduling, parameters, retries, and multi-task DAGs.