Declarative Pipelines in Databricks: Expectations, Streaming Tables, and When the Abstraction Fights You

I resisted Delta Live Tables for about a year, for a reason I now think was half right. It looked like a framework wrapping code I already knew how to write, adding a proprietary layer between me and Spark, with a price premium attached.

What changed my mind was counting what my hand-rolled pipelines actually contained. A medallion pipeline I maintained had roughly 400 lines of transformation logic and 900 lines of scaffolding: checkpoint path construction, idempotent merge helpers, a retry-with-backoff decorator, a table dependency order encoded in an Airflow DAG, schema evolution handling, and a homegrown data quality module that wrote failures to a side table. All of it was code my team wrote, tested, and got paged about.

DLT — now marketed as Lakeflow Declarative Pipelines — replaces that scaffolding. You declare tables as functions returning DataFrames, and the runtime works out the dependency graph, manages checkpoints, handles retries, enforces quality constraints, and records everything in a queryable event log. The transformation logic stays plain Spark.

It is not free and it is not universal. Here is how it behaves in production, and the four cases where I still write ordinary jobs.

What you actually declare

A DLT pipeline is a set of decorated functions. There is no writeStream, no checkpoint path, no execution order.

import dlt
from pyspark.sql import functions as F

# Bronze: streaming ingest from cloud files, append-only
@dlt.table(
    name="bronze_orders",
    comment="Raw order events from the ingest bucket",
    table_properties={"quality": "bronze", "delta.autoOptimize.optimizeWrite": "true"},
)
def bronze_orders():
    return (
        spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.schemaLocation", "/Volumes/prod/_schemas/orders")
            .option("cloudFiles.inferColumnTypes", "true")
            .load("/Volumes/prod/landing/orders/")
            .select("*", F.current_timestamp().alias("_ingested_at"),
                    F.col("_metadata.file_path").alias("_source_file"))
    )

# Silver: incremental, with quality gates that quarantine rather than fail
@dlt.table(name="silver_orders", comment="Validated, typed orders")
@dlt.expect_or_drop("order_id_present", "order_id IS NOT NULL")
@dlt.expect_or_drop("positive_amount", "net_amount > 0")
@dlt.expect("recent_event", "order_ts > '2020-01-01'")
def silver_orders():
    return (
        dlt.read_stream("bronze_orders")          # dependency inferred here
           .withColumn("order_ts", F.to_timestamp("order_ts"))
           .withColumn("net_amount", F.col("net_amount").cast("decimal(18,2)"))
           .dropDuplicates(["order_id", "order_ts"])
    )

# Gold: a materialized view, recomputed when upstream changes
@dlt.table(name="gold_revenue_daily")
def gold_revenue_daily():
    return (
        dlt.read("silver_orders")
           .groupBy(F.to_date("order_ts").alias("order_date"), "region")
           .agg(F.sum("net_amount").alias("revenue"),
                F.count("*").alias("orders"))
    )

The graph comes from dlt.read and dlt.read_stream calls. Nothing else declares order, which means renaming a table cannot desync a DAG definition from the code — a class of bug I used to hit every few months with Airflow-orchestrated Spark jobs.

flowchart TD
  land["Landing volume: JSON files"] --> br["bronze_orders: streaming table"]
  br --> si["silver_orders: streaming table"]
  si --> gd["gold_revenue_daily: materialized view"]
  si --> qr["Dropped rows: event log"]
  ev["Pipeline event log"] --> mon["Quality and lineage queries"]
  qr --> ev

The graph is inferred from reads. Expectations feed the event log, which is where quality monitoring actually happens.

Streaming tables versus materialized views

This is the decision that determines your compute bill, and the one I see made carelessly.

A streaming table processes each source record once. It keeps a checkpoint, reads only new data, and appends. Cheap and incremental, but the source must be append-only — an update or delete upstream either breaks it or requires skipChangeCommits, which ignores those changes rather than handling them.

A materialized view is defined by a query and recomputed to stay consistent with its inputs. Databricks applies incremental refresh when it can prove the computation is incrementalizable, and otherwise does a full recompute. Correct under any upstream change, potentially expensive.

My rules:

• Bronze ingest: always a streaming table.
• Silver cleaning and typing on append-only sources: streaming table.
• Silver with upserts or CDC: create_auto_cdc_flow (formerly apply_changes), which handles SCD logic properly.
• Gold aggregates: materialized view, unless the aggregate is over a strictly growing window and volume makes recompute painful.

The trap is a gold materialized view over a 4 TB silver table with a non-incrementalizable aggregation — a count(distinct ...) or a window function — recomputing every 15 minutes. That is a full scan per refresh, and it will dominate your Databricks bill without anyone noticing, because the pipeline looks healthy.

Check the event log for planning_information to see whether refreshes are incremental or full. If they are full and the table is large, either widen the refresh interval or restructure the aggregation into a streaming table with two-phase aggregation.

CDC without writing MERGE

The MERGE logic for slowly changing dimensions is where hand-rolled pipelines accumulate bugs: out-of-order records, deletes represented as flags, tie-breaking on identical timestamps. DLT handles it declaratively.

import dlt
from pyspark.sql import functions as F

dlt.create_streaming_table("silver_customers")

dlt.create_auto_cdc_flow(
    target="silver_customers",
    source="bronze_customers_cdc",
    keys=["customer_id"],
    sequence_by=F.struct("commit_ts", "commit_seq"),  # tie-break with a struct
    apply_as_deletes=F.expr("op = 'DELETE'"),
    except_column_list=["op", "commit_ts", "commit_seq", "_ingested_at"],
    stored_as_scd_type=1,
)

# SCD Type 2 into a separate table: full history with validity windows
dlt.create_streaming_table("silver_customers_history")

dlt.create_auto_cdc_flow(
    target="silver_customers_history",
    source="bronze_customers_cdc",
    keys=["customer_id"],
    sequence_by=F.struct("commit_ts", "commit_seq"),
    apply_as_deletes=F.expr("op = 'DELETE'"),
    except_column_list=["op", "commit_ts", "commit_seq", "_ingested_at"],
    stored_as_scd_type=2,
)

Out-of-order handling is the part worth paying for. If a batch contains version 3 of a row followed by version 2, sequence_by ensures version 3 wins. My hand-written merges got this wrong twice, and both times it surfaced as a customer attribute silently reverting.

Using a struct for sequence_by is a small trick that matters: CDC feeds frequently have multiple changes within the same millisecond, and a timestamp alone gives you nondeterministic ordering.

Expectations, and treating quality as a time series

Three expectation flavors, and the middle one is where almost all my constraints live:

• @dlt.expect — record the violation, keep the row. Monitoring.
• @dlt.expect_or_drop — record it, drop the row. Quarantine.
• @dlt.expect_or_fail — abort the pipeline. Reserved for invariants that mean something is fundamentally broken.

I use expect_or_fail for exactly one class of check: a primary key that must be unique, or a required column that has gone entirely NULL, meaning the upstream schema changed underneath us. Everything else drops or warns, because a pipeline that halts on 12 bad rows out of 40 million creates an incident larger than the data problem.

The event log is what makes this operational. Every expectation evaluation is recorded with pass and fail counts per batch:

-- Expectation pass rates per table per day, from the pipeline event log
SELECT
  timestamp::date                                        AS d,
  details:flow_progress.output_dataset                   AS dataset,
  expectation.name                                       AS expectation,
  sum(expectation.passed_records)                        AS passed,
  sum(expectation.failed_records)                        AS failed,
  round(100.0 * sum(expectation.failed_records)
        / nullif(sum(expectation.passed_records + expectation.failed_records), 0), 3) AS fail_pct
FROM event_log(TABLE(prod.silver.silver_orders)) e
LATERAL VIEW explode(
  from_json(e.details:flow_progress.data_quality.expectations,
            'array<struct<name:string,dataset:string,passed_records:bigint,failed_records:bigint>>')
) t AS expectation
WHERE e.event_type = 'flow_progress'
  AND timestamp >= current_date() - INTERVAL 14 DAYS
GROUP BY 1, 2, 3
ORDER BY 1 DESC, failed DESC;

I materialize that into a table and alert on fail percentage changing by more than a factor of two week over week. Absolute thresholds produce noise; rate-of-change catches real upstream regressions. This is the same philosophy as data quality checks in Python, except the plumbing already exists and the results are queryable without you building a results table.

One gap worth knowing: expect_or_drop drops rows without writing them anywhere. If you need the bad rows for investigation, invert the expectation into a separate quarantine table reading the same source. It is a few extra lines and it converts “we dropped 4,000 rows” into “here are the 4,000 rows.”

When the abstraction fights you

Four situations where I still write plain Spark jobs.

Complex Python orchestration. DLT functions must return a DataFrame. If a step needs to call an API, branch on a result, write to two places conditionally, or run a loop over accounts with different credentials, you are fighting the model. foreachBatch in a plain stream gives you arbitrary Python; DLT does not.

External side effects. Publishing to Kafka, calling a notification service, writing to an operational database. DLT owns the write path and it writes to tables it manages. A plain job with its own sink logic is the right tool.

Precise control over write timing and layout. If you need partition-scoped overwrites, specific file sizes, replaceWhere semantics, or coordinated writes across tables in one transaction, DLT’s managed writes are a constraint rather than a convenience.

Tables other systems must write. DLT owns its output tables exclusively. If another engine also writes that table, DLT is out. That is the same ownership boundary that decides managed versus external tables in Unity Catalog.

I also weigh the cost premium honestly. DLT compute carries a DBU surcharge over plain job compute. For a pipeline where the scaffolding is 60 percent of the code, the engineering saving dwarfs it. For a single-table job that reads and writes once, it does not, and I write the job.

Pitfalls

Full refresh on a streaming table. It truncates the target and reprocesses from the source’s earliest available data. If the source is a Kafka topic with 7-day retention or a landing bucket with a lifecycle policy, you get a permanent gap. Set pipelines.reset.allowed to false on bronze tables you cannot rebuild.

Materialized views over huge tables with non-incrementalizable aggregations. Every refresh is a full scan. Check the event log for the refresh type before you trust the schedule.

Mixing streaming and batch reads carelessly. dlt.read on a table that a streaming flow is appending to gives you a snapshot at plan time, not a stream. It is correct but it makes downstream tables recompute rather than increment.

Expectations as documentation. A @dlt.expect nobody queries is a comment with a runtime cost. Wire the event log into alerting or use expect_or_drop.

Development mode in production. Development mode reuses the cluster and does not retry failed flows, which is what you want when iterating and dangerous when scheduled. The pipeline mode is part of the deployment config, so put it in the asset bundle.

Assuming DLT removes the need to understand Spark. Skew, shuffle volume, and partition sizing behave exactly as they always do. A declarative wrapper does not change the physics, and you will still be reading task distributions when a flow gets slow.

FAQ

Is DLT worth the DBU premium?

When the pipeline has real scaffolding — multiple tables, checkpoints, quality checks, CDC — yes, comfortably. For a single-hop job, no. I decide by asking what fraction of the code would be infrastructure rather than transformation.

Can I run DLT pipelines from CI?

Yes, through Databricks Asset Bundles. The pipeline definition and its notebooks live in git, the bundle deploys per environment, and validation runs on PR. Treat pipeline configuration as code, the same as job definitions.

Does DLT work with SQL as well as Python?

Yes, and the SQL syntax is clean for straightforward medallion transforms. I use Python when I need shared helper functions, loops that generate many similar tables, or complex schema handling, and SQL for everything an analytics engineer will maintain.

How do I debug a slow flow?

The event log gives per-flow durations and row counts, and each flow has a Spark UI you can open from the pipeline graph. From there the diagnosis is ordinary Spark work: exchange nodes, task duration distribution, and spill.

Should DLT replace dbt on Databricks?

They overlap on transformation but differ on strength. dbt is better at modeling conventions, testing ergonomics, and multi-engine portability; DLT is better at streaming ingest, CDC, and managed incremental state. Teams running both usually put ingest and CDC in DLT and analytics modeling in dbt, which is close to what I recommend after reading how each handles incremental models.

What this means for your pipelines

Count your scaffolding before you decide. Open your largest Spark pipeline and mark every line that is checkpoint management, retry logic, dependency ordering, merge boilerplate, or data quality plumbing. If that is more than half the file, DLT will pay for itself in maintenance alone, and the DBU premium is noise against an engineer’s time.

Then make two decisions deliberately rather than by default. Every table is either a streaming table or a materialized view, and the wrong choice on a large gold aggregate is the most expensive mistake available in this system. Every expectation is either a monitor, a quarantine, or a hard stop, and only genuine invariants get to stop the pipeline.

Keep the escape hatch in mind. DLT is very good at the medallion shape — ingest, clean, aggregate, with quality gates and CDC — and it is a poor fit for orchestration with side effects or writes you need to control precisely. Mixing both in one platform is fine. Forcing a pipeline that wants arbitrary Python into a declarative model is how a good abstraction earns a bad reputation.