Delta Live Tables Lakeflow Declarative Pipelines: Streaming + Batch as SQL/Python

lakeflow declarative pipelines is the framework you now reach for when you want a Databricks pipeline to define what the data should look like instead of scripting every step to produce it — and it is the same product that shipped for years as Delta Live Tables, renamed and folded into the broader Lakeflow family at the 2025 Data + AI Summit. The distinction matters because the entire mental model is inverted from a hand-wired job: you do not write a driver that reads a source, transforms it, writes a table, checkpoints, retries, and schedules the next task. Instead you declare a set of target datasets — each one a streaming table or a materialized view backed by a SQL or Python query — and the engine reads those definitions, works out the dependency graph between them, decides what has to be recomputed, manages the checkpoints and retries, enforces your data-quality rules, and publishes observability for every run.

This guide is the walkthrough you wished existed the first time an interviewer said "Delta Live Tables got renamed — walk me through what Lakeflow Declarative Pipelines actually is and why it beats a hand-wired Spark plus orchestrator job," or "when do you pick a streaming table over a materialized view," or "what replaced APPLY CHANGES INTO and how do you build an SCD Type 2 dimension with it." It moves through five things every data engineer must be fluent in: why the declarative model wins over imperative jobs and how the dlt → databricks lakeflow rebrand is organised; the core dataset model — streaming tables versus materialized views, the decorators, and the dependency DAG that stitches a medallion architecture together; data quality expectations (EXPECT / DROP / FAIL, quarantine, the event log); streaming and batch unified through auto loader and AUTO CDC INTO for change data capture; and the operational surface — development versus production mode, serverless unit economics, and the migration path. Each section pairs a teaching block with a Solution-Tail interview answer: code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library →, rehearse on the streaming practice library →, and sharpen the modelling axis with the data-transformation practice library →.

---

On this page

• Why declarative pipelines beat hand-wired jobs
• Core model — streaming tables vs materialized views
• Data quality and expectations — EXPECT, DROP, FAIL
• Streaming and batch unified — Auto Loader and AUTO CDC
• Ops, migration, and interview signals
• Cheat sheet — Lakeflow Declarative Pipelines recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why declarative pipelines beat hand-wired jobs

The DLT → Lakeflow rebrand and the imperative-to-declarative shift — you declare datasets, the engine builds the pipeline

The one-sentence invariant: a lakeflow declarative pipelines job is a set of dataset definitions — each target table paired with the query that produces it — from which the engine derives the dependency graph, the execution order, the incremental-vs-full recompute decision, the checkpoint and retry behaviour, and the data-quality enforcement, so that the engineer never writes an orchestration driver, a foreachBatch checkpoint dance, or a per-task Airflow dependency by hand. The rename from Delta Live Tables changed the marketing surface, not the shape of the abstraction: you still write @dlt.table or CREATE OR REFRESH STREAMING TABLE, you still get streaming tables, materialized views, and expectations, and the older @dlt code keeps running. What changed is that the product is now one pillar of a three-part family and the syntax has a forward-looking successor.

The Lakeflow family — three pillars, one lineage each.

• Lakeflow Connect — managed ingestion connectors (databases, SaaS apps, files) that land raw data into the lakehouse. This is the "get the bytes in" pillar; it feeds the bronze layer.
• Lakeflow Declarative Pipelines — the transformation pillar, formerly Delta Live Tables. This is where streaming tables, materialized views, and data quality expectations live. Everything in this guide is this pillar.
• Lakeflow Jobs — the orchestration pillar, formerly Databricks Jobs / Workflows. It schedules and chains tasks (including a pipeline update as a task).
• The relationship. Connect ingests, Declarative Pipelines transforms, Jobs orchestrates. A pipeline update is itself a task you can trigger from Jobs — but inside a pipeline you never schedule tasks, because the pipeline engine derives its own internal ordering.

Imperative vs declarative — the difference in one contrast.

• Imperative (hand-wired). You write a driver: read source A, transform, write silver_a, checkpoint; read silver_a plus source B, transform, write gold; register the write-order in an orchestrator; add retry logic; add a data-quality check as a separate job; wire alerting. You own the DAG, the state, the retries, the quality gates.
• Declarative (Lakeflow). You write: "silver_a is SELECT ... FROM STREAM(source_a) with these expectations; gold is SELECT ... FROM silver_a JOIN source_b." The engine parses the dlt.read/STREAM() references, discovers that gold depends on silver_a, orders them, runs them, checkpoints the streaming ones, retries on transient failure, and records quality metrics — without a line of orchestration code.
• The payoff. The definition is the documentation, the lineage, and the contract. Add a new table and the graph re-derives itself; nobody edits a DAG file. Delete a table and its downstream dependents are flagged.

What the engine manages so you don't.

• Dependency graph. Derived from the dlt.read/dlt.read_stream/STREAM()/FROM live.table references. No manual task_a >> task_b.
• Incremental state. Streaming tables keep Structured Streaming checkpoints automatically; materialized views track what changed and recompute incrementally when possible.
• Retries and recovery. Transient failures retry; a failed run leaves the last-good table version in place (Delta time travel underneath).
• Data quality. Expectations are declared on the table and enforced on every row, with pass/drop/fail metrics emitted to the event log.
• Observability. Every run writes a structured event log (row counts, expectation metrics, lineage, durations) you can query as a table.

What interviewers listen for.

• Do you say "declare the target dataset and its query; the engine builds the DAG" rather than "it's a fancy scheduler"? — required answer.
• Do you name the DLT → Lakeflow Declarative Pipelines rename and place it in the Lakeflow Connect / Pipelines / Jobs family without prompting? — senior signal.
• Do you distinguish streaming tables from materialized views by source semantics (append-only vs full recompute), not by "one is faster"? — required answer.
• Do you mention that expectations and lineage are declared on the table, so quality and provenance travel with the data? — senior signal.

Worked example — the same medallion step, imperative vs declarative

Detailed explanation. The clearest way to feel the shift is to build the same silver step twice: once as a hand-wired Structured Streaming job with an explicit checkpoint and merge, and once as a declarative streaming table. The declarative version deletes the orchestration, the checkpoint path, and the write mechanics — they become the engine's job.

• Task. Read an append-only bronze events stream, filter out nulls, standardise a timestamp, write to a silver table.
• Imperative burden. Checkpoint location, output mode, trigger, foreachBatch or .start(), restart-on-failure, schema handling.
• Declarative burden. One function; one @dlt.table; one dlt.read_stream("bronze_events").

Question. Rewrite an imperative Structured Streaming silver step as a lakeflow declarative pipelines streaming table and list what the engine now owns.

Input.

Concern	Imperative job	Declarative pipeline
Read source	spark.readStream.table(...)	dlt.read_stream("bronze_events")
Checkpoint	you set checkpointLocation	engine-managed
Write	.writeStream...start()	return the DataFrame from @dlt.table
Ordering	orchestrator dependency	derived from dlt.read_stream
Retry	your try/except + restart	engine-managed

Code.

# ---------- IMPERATIVE: hand-wired Structured Streaming ----------
from pyspark.sql import functions as F

def run_silver_imperative(spark):
    src = spark.readStream.table("catalog.bronze.events")
    silver = (
        src.where(F.col("user_id").isNotNull())
           .withColumn("event_ts", F.to_timestamp("event_ts_raw"))
           .drop("event_ts_raw")
    )
    (silver.writeStream
        .format("delta")
        .option("checkpointLocation", "/mnt/chk/silver_events")  # you own this
        .outputMode("append")
        .trigger(availableNow=True)
        .toTable("catalog.silver.events"))                       # you own restart/retry


# ---------- DECLARATIVE: Lakeflow Declarative Pipelines ----------
import dlt
from pyspark.sql import functions as F

@dlt.table(
    name="silver_events",
    comment="Cleaned events, one row per source row (streaming/incremental).",
    table_properties={"quality": "silver"},
)
def silver_events():
    return (
        dlt.read_stream("bronze_events")          # dependency edge derived from here
           .where(F.col("user_id").isNotNull())
           .withColumn("event_ts", F.to_timestamp("event_ts_raw"))
           .drop("event_ts_raw")
    )

Step-by-step explanation.

1. The read. In the imperative job you call spark.readStream.table(...) against a fully-qualified name. In the declarative version you call dlt.read_stream("bronze_events") against a pipeline dataset name. That call is what the engine parses to draw the edge bronze_events → silver_events in the DAG — the dependency is discovered from the code, not declared in a separate file.
2. The checkpoint. The imperative job hard-codes a checkpointLocation; if you forget it, or two jobs collide on the same path, you get silent corruption. The declarative table has no checkpoint in the code because the engine assigns and manages one per streaming table, keyed to the pipeline's storage.
3. The write. Imperative code ends in .writeStream...toTable(...). Declarative code just returns a DataFrame from the decorated function; the engine performs the write, sets the table properties, and owns the output mode (append for streaming tables).
4. Ordering and retry. The orchestrator dependency (bronze first, then silver) and the restart-on-failure logic vanish from your code. The engine runs bronze_events, then silver_events, retries transient errors, and leaves the prior good version if a run fails.
5. The net. Roughly a dozen lines of ceremony — checkpoint path, output mode, trigger, restart handling, orchestration edge — collapse into a decorator plus a dlt.read_stream reference. That deleted ceremony is exactly the surface where hand-wired jobs accumulate bugs.

Output.

Responsibility	Imperative owner	Declarative owner
Dependency ordering	you (orchestrator)	engine (from dlt.read_stream)
Checkpoint / state	you (path)	engine
Output mode	you (append)	engine (streaming table)
Retry / recovery	you (try/except)	engine
Last-good on failure	you (manual)	engine (Delta version)

Rule of thumb. If your pipeline code contains a checkpointLocation, a hand-managed output mode, or an orchestrator dependency edge, you are still writing imperative ETL. The declarative version deletes all three; what remains is the query and the dataset name.

Worked example — reading the Lakeflow family map correctly

Detailed explanation. Interviewers frequently test whether you understand that "Lakeflow" is not one product but three pillars, and that Declarative Pipelines is specifically the transformation pillar (the old DLT). Getting this map right prevents the common error of trying to schedule tasks inside a pipeline or ingest with a pipeline when a Connect connector is the right tool.

• Ingest. Lakeflow Connect — managed source connectors.
• Transform. Lakeflow Declarative Pipelines — this guide's subject (was DLT).
• Orchestrate. Lakeflow Jobs — schedule and chain tasks (was Workflows).

Question. Given a raw-to-gold requirement, place each responsibility on the correct Lakeflow pillar and explain why a pipeline does not schedule its own internal tasks.

Input.

Requirement	Wrong tool	Right pillar
Pull a Salesforce object hourly	a DLT @dlt.table scraping the API	Lakeflow Connect connector
Clean + join bronze into gold	an Airflow chain of notebooks	Lakeflow Declarative Pipelines
Run the pipeline at 06:00, then a report	a time.sleep in the pipeline	Lakeflow Jobs schedule
Order silver before gold	a manual task_a >> task_b	engine-derived DAG

Code.

# Inside a pipeline you NEVER schedule. You declare datasets and let the
# engine order them. This whole file is one pipeline; the engine derives:
#   raw_orders -> silver_orders -> gold_daily_revenue

import dlt
from pyspark.sql import functions as F

@dlt.table
def silver_orders():
    # depends on raw_orders (ingested by a Lakeflow Connect connector or Auto Loader)
    return (dlt.read_stream("raw_orders")
              .where(F.col("amount_cents") > 0))

@dlt.table
def gold_daily_revenue():
    # depends on silver_orders; edge derived automatically
    return (dlt.read("silver_orders")
              .groupBy(F.to_date("event_ts").alias("day"))
              .agg(F.sum("amount_cents").alias("revenue_cents")))

// Lakeflow Jobs (orchestration) schedules the pipeline UPDATE as a task.
// This lives OUTSIDE the pipeline definition, not inside it.
{
  "name": "daily-revenue-workflow",
  "schedule": { "quartz_cron_expression": "0 0 6 * * ?", "timezone_id": "UTC" },
  "tasks": [
    { "task_key": "refresh_pipeline",
      "pipeline_task": { "pipeline_id": "abc-123-declarative-pipeline" } },
    { "task_key": "email_report",
      "depends_on": [ { "task_key": "refresh_pipeline" } ],
      "notebook_task": { "notebook_path": "/Reports/DailyRevenue" } }
  ]
}

Step-by-step explanation.

1. Ingestion is not transformation. Pulling from Salesforce or a database belongs to Lakeflow Connect (or Auto Loader for files) — writing a @dlt.table that calls a REST API is an anti-pattern because it puts non-idempotent I/O inside the declarative graph.
2. Transformation is the pipeline. silver_orders and gold_daily_revenue are two dataset definitions in one pipeline file. The engine sees dlt.read("silver_orders") inside gold_daily_revenue and draws the edge; you never write the ordering.
3. Orchestration is external. The schedule ("06:00 daily") and cross-system chaining ("then email a report") belong to Lakeflow Jobs, which triggers the pipeline update as one task and the report notebook as a dependent task.
4. Why no internal scheduling. A pipeline's internal order is a pure function of its data dependencies. Introducing sleep/cron inside it would break the engine's ability to reason about the graph and to recompute incrementally. Scheduling is a job concern, not a pipeline concern.
5. The clean separation. Connect fills bronze, Declarative Pipelines produces silver/gold, Jobs decides when the pipeline runs and what happens after. Each pillar has exactly one job; mixing them is the most common design smell interviewers probe.

Output.

Pillar	Owns	Formerly called
Lakeflow Connect	managed ingestion	(new / Fivetran-style connectors)
Lakeflow Declarative Pipelines	streaming tables, materialized views, expectations	Delta Live Tables (DLT)
Lakeflow Jobs	scheduling, task chaining	Databricks Jobs / Workflows

Rule of thumb. If the requirement is when something runs or what runs after it, that is Lakeflow Jobs. If it is what a dataset should contain, that is a Declarative Pipeline. If it is how raw data arrives, that is Lakeflow Connect (or Auto Loader). Never solve one pillar's problem with another's tool.

Worked example — the dependency DAG the engine derives

Detailed explanation. The single most important thing to internalise is that the DAG is derived, not authored. Every dlt.read/dlt.read_stream/STREAM() reference becomes an edge. Understanding this lets you predict exactly what recomputes when a source changes and why a typo in a dataset name breaks the graph at validation time, not runtime.

• Nodes. Each @dlt.table / @dlt.materialized_view / SQL CREATE OR REFRESH ... is a node.
• Edges. Each read of another pipeline dataset is a directed edge into the current node.
• Validation. The engine resolves the whole graph before running anything; a missing or cyclic reference fails validation up front.

Question. Given a bronze → silver → two-gold pipeline, draw the derived DAG and state what recomputes when only the bronze source gets new rows.

Input.

Dataset	Reads	Type
bronze_clicks	Auto Loader source files	streaming table
silver_clicks	bronze_clicks	streaming table
gold_daily	silver_clicks	materialized view
gold_top_pages	silver_clicks	materialized view

Code.

import dlt
from pyspark.sql import functions as F

@dlt.table
def bronze_clicks():
    return (spark.readStream.format("cloudFiles")
              .option("cloudFiles.format", "json")
              .load("/Volumes/main/raw/clicks/"))

@dlt.table
def silver_clicks():
    return (dlt.read_stream("bronze_clicks")           # edge: bronze -> silver
              .where(F.col("url").isNotNull()))

@dlt.materialized_view
def gold_daily():
    return (dlt.read("silver_clicks")                  # edge: silver -> gold_daily
              .groupBy(F.to_date("ts").alias("day"))
              .agg(F.count("*").alias("clicks")))

@dlt.materialized_view
def gold_top_pages():
    return (dlt.read("silver_clicks")                  # edge: silver -> gold_top_pages
              .groupBy("url")
              .agg(F.count("*").alias("clicks"))
              .orderBy(F.desc("clicks")))

Step-by-step explanation.

1. The parse. Before any compute, the engine reads all four functions and resolves the references: silver_clicks reads bronze_clicks; both gold views read silver_clicks. That produces a DAG with one root (bronze_clicks), one middle node, and two leaves.
2. Validation first. If gold_daily had read "silver_click" (typo), the whole pipeline fails at validation with "dataset not found" — before wasting compute. Cycles are rejected the same way. This up-front resolution is a major reliability win over imperative jobs that discover a missing table at runtime.
3. Streaming vs materialized recompute. New bronze rows flow incrementally into silver_clicks (streaming table, append-only). The two materialized views then recompute their aggregates — incrementally where the engine can, or fully when the query shape forces it.
4. What does not recompute. Nothing upstream of a change recomputes. If only gold_top_pages's definition changes, bronze_clicks and silver_clicks are untouched; only that one leaf refreshes.
5. The lineage dividend. Because edges are derived, the pipeline UI can render exact column-level lineage and impact analysis for free. Delete silver_clicks and the engine immediately flags both gold views as broken dependents.

Output.

Change	Recomputes	Untouched
New rows in bronze source files	bronze → silver (incremental) → both gold (refresh)	—
Edit gold_top_pages query	gold_top_pages only	bronze, silver, gold_daily
Typo in a dlt.read name	nothing (validation fails first)	whole pipeline (safe stop)
Delete silver_clicks	validation error	flags both gold as broken

Rule of thumb. The DAG is a pure function of your dlt.read references. To predict what recomputes, trace the edges downstream from the change; to catch breakage early, trust that the engine resolves the entire graph before running a single node.

Senior interview question on the declarative model

A senior interviewer might open with: "Your team runs a bronze→silver→gold medallion as three separate Spark Structured Streaming jobs glued together with Airflow, and it keeps breaking on checkpoint drift and out-of-order restarts. Make the case for rewriting it as lakeflow declarative pipelines, show the shape of the rewrite, and name exactly what operational burden disappears."

Solution Using a single declarative pipeline that derives the DAG, manages state, and enforces quality

# pipeline.py — the ENTIRE bronze->silver->gold medallion as one declarative pipeline.
# No orchestrator, no checkpoint paths, no restart logic. The engine owns all of it.
import dlt
from pyspark.sql import functions as F

# --- BRONZE: raw ingest via Auto Loader (append-only streaming table) ---
@dlt.table(comment="Raw orders, exactly-once file ingest.", table_properties={"quality": "bronze"})
def bronze_orders():
    return (spark.readStream.format("cloudFiles")
              .option("cloudFiles.format", "json")
              .option("cloudFiles.schemaLocation", "/Volumes/main/schemas/orders")
              .load("/Volumes/main/raw/orders/"))

# --- SILVER: clean + typed (streaming table); quality gate declared inline ---
@dlt.table(comment="Cleaned, typed orders.", table_properties={"quality": "silver"})
@dlt.expect_or_drop("valid_amount", "amount_cents > 0")
@dlt.expect_or_drop("has_customer", "customer_id IS NOT NULL")
def silver_orders():
    return (dlt.read_stream("bronze_orders")
              .withColumn("event_ts", F.to_timestamp("event_ts_raw"))
              .withColumn("amount_cents", F.col("amount_cents").cast("bigint"))
              .drop("event_ts_raw"))

# --- GOLD: business aggregate (materialized view; recomputes incrementally) ---
@dlt.materialized_view(comment="Daily revenue per region.")
def gold_daily_revenue():
    return (dlt.read("silver_orders")
              .groupBy(F.to_date("event_ts").alias("day"), "region")
              .agg(F.sum("amount_cents").alias("revenue_cents"),
                   F.count("*").alias("order_count")))

-- The identical pipeline expressed in SQL — same three datasets, same derived DAG.
CREATE OR REFRESH STREAMING TABLE bronze_orders
  AS SELECT * FROM STREAM read_files('/Volumes/main/raw/orders/', format => 'json');

CREATE OR REFRESH STREAMING TABLE silver_orders (
  CONSTRAINT valid_amount EXPECT (amount_cents > 0) ON VIOLATION DROP ROW,
  CONSTRAINT has_customer EXPECT (customer_id IS NOT NULL) ON VIOLATION DROP ROW
) AS
  SELECT to_timestamp(event_ts_raw) AS event_ts,
         CAST(amount_cents AS BIGINT) AS amount_cents,
         customer_id, region
  FROM STREAM(bronze_orders);

CREATE OR REFRESH MATERIALIZED VIEW gold_daily_revenue AS
  SELECT to_date(event_ts) AS day, region,
         SUM(amount_cents) AS revenue_cents,
         COUNT(*)          AS order_count
  FROM silver_orders
  GROUP BY 1, 2;

Step-by-step trace.

Step	Before (3 jobs + Airflow)	After (one declarative pipeline)
Dependency ordering	Airflow bronze >> silver >> gold	derived from dlt.read/STREAM()
Checkpoints	2 hand-set paths (drift-prone)	engine-managed, one per streaming table
Restart after failure	manual re-run, risk of double-write	engine retries; last-good Delta version kept
Data quality	separate validation job	inline expect_or_drop on silver
Gold recompute	full recompute each run	incremental materialized-view refresh
Observability	scattered logs	one structured event log table

After the rewrite, the three jobs and the Airflow DAG collapse into one pipeline file. Checkpoint drift is impossible because the engine owns the checkpoints. A transient failure in silver no longer leaves gold reading a half-written table — the engine keeps the last good version until the new one is complete. Quality violations are dropped and counted in the event log rather than silently passing through.

Output:

Metric	Before	After
Files/artifacts to maintain	3 jobs + 1 DAG + 2 checkpoints	1 pipeline definition
Checkpoint-drift incidents	recurring	eliminated (engine-owned)
Ordering bugs on restart	possible	impossible (derived DAG)
Data-quality visibility	none inline	pass/drop metrics per run
Gold refresh cost	full recompute	incremental where possible

Why this works — concept by concept:

• Declarative dataset definitions — each table is defined by its query, not by a script. The engine derives the bronze → silver → gold DAG from the dlt.read/STREAM() references, removing the entire orchestration layer.
• Engine-managed streaming state — streaming tables carry their own Structured Streaming checkpoints, keyed to pipeline storage. Checkpoint drift, the root cause of the old breakages, cannot occur because engineers no longer set paths.
• Inline expectations — expect_or_drop moves data quality into the table definition, so the constraint travels with the data and produces metrics, replacing a brittle separate validation job.
• Materialized-view incremental refresh — gold_daily_revenue recomputes only what changed when the query shape allows it, turning an O(all-history) nightly recompute into an O(delta) refresh.
• Cost — one pipeline definition versus three jobs plus a DAG plus two checkpoints. Compute is O(delta) per run for streaming tables and incremental views rather than O(full) per job; the eliminated cost is the human time spent debugging checkpoint drift and restart ordering.

ETL
Topic — etl
ETL problems on medallion pipeline design

Practice →

Streaming Topic — streaming Streaming problems on incremental pipelines

Practice →

---

2. Core model — streaming tables vs materialized views

streaming tables process each source row once; materialized views recompute a query result — pick by source semantics, not by speed

The mental model in one line: a streaming table incrementally appends the result of a query over an append-only (or CDC) source, processing every source row exactly once and keeping a checkpoint, while a materialized view is the persisted result of a batch query that the engine keeps fresh by recomputing it — incrementally when the query shape permits, fully otherwise — so the choice between them is decided by whether your source is an append-only stream you want to consume once, or a dataset whose current aggregate/joined result you want always-correct. These are the only two persistent dataset kinds in lakeflow declarative pipelines (plus temporary views), and every senior design question about the framework reduces to picking the right one for each medallion layer.

Streaming table — the append/incremental dataset.

• What it is. A Delta table populated by a streaming query. Each input row is processed once; the engine tracks progress with a checkpoint so a restart resumes, not reprocesses.
• Source requirement. The source must be append-only from the streaming query's point of view (raw files via Auto Loader, a Kafka topic, or another streaming table). You cannot stream from a source that gets arbitrary updates/deletes without extra handling.
• When to use it. Bronze ingestion and silver cleaning — high-volume, append-heavy layers where you want to touch each row exactly once and never re-scan history.
• Cost profile. O(new rows) per run. Ideal for large, continuously-growing sources.

Materialized view — the recomputed query result.

• What it is. A Delta table defined by a batch query whose result the engine keeps up to date. Reading it is just reading a table; refreshing it re-derives the result.
• Correctness guarantee. A materialized view always reflects the full current result of its query against current inputs — even for aggregates, joins, and window functions that a single-pass stream cannot express correctly.
• When to use it. Gold aggregates, dimension joins, and any layer where "the answer must be right over all current data" beats "process each row once."
• Cost profile. Incremental when the engine can compute the delta (many aggregates/joins qualify); full recompute otherwise. Still cheaper than a hand-written full rebuild because the engine chooses incremental automatically.

The decorators and SQL keywords — three ways to spell each.

• Python (DLT-era, still supported). @dlt.table for a streaming table, @dlt.view for a temporary view; expectations via @dlt.expect*. This is what most existing code uses and it keeps running.
• Python (Lakeflow / Spark Declarative Pipelines). from pyspark import pipelines as dp then @dp.table, @dp.materialized_view, @dp.temporary_view — the forward-looking, open-source-aligned spelling.
• SQL. CREATE OR REFRESH STREAMING TABLE and CREATE OR REFRESH MATERIALIZED VIEW. STREAM(...) marks a streaming read of another dataset; a plain FROM name is a batch read.

The dependency DAG and the medallion.

• Bronze. Streaming tables from Auto Loader / Connect — raw, append-only, each file processed once.
• Silver. Streaming tables reading bronze via STREAM() / dlt.read_stream — cleaned, typed, deduped, quality-gated.
• Gold. Materialized views reading silver as batch — aggregates and dimension joins that must be correct over all current rows.
• Reads set the edges. dlt.read_stream("x") / STREAM(x) = streaming edge; dlt.read("x") / FROM x = batch edge. The engine composes these into one DAG.

Worked example — a streaming table for silver, a materialized view for gold

Detailed explanation. The canonical pairing: silver is a streaming table (append-only, one pass per row), gold is a materialized view (correct aggregate over all current silver rows). Building both shows exactly where the streaming edge and the batch edge fall, and why the aggregate belongs in a materialized view rather than a streaming table.

• Silver. Streaming table over bronze; incremental append.
• Gold. Materialized view over silver; recomputed aggregate.
• Edge kinds. STREAM(bronze) into silver; plain FROM silver into gold.

Question. Define a silver streaming table and a gold materialized view, and justify why the aggregate cannot simply be a streaming table.

Input.

Dataset	Kind	Read of upstream	Why
silver_sales	streaming table	STREAM(bronze_sales)	append-only, one pass/row
gold_region_totals	materialized view	FROM silver_sales (batch)	correct SUM over all current rows

Code.

-- SILVER: streaming table, incremental append, one pass per row
CREATE OR REFRESH STREAMING TABLE silver_sales AS
  SELECT sale_id,
         region,
         CAST(amount_cents AS BIGINT) AS amount_cents,
         to_timestamp(sold_at_raw)    AS sold_at
  FROM STREAM(bronze_sales)
  WHERE amount_cents IS NOT NULL;

-- GOLD: materialized view, correct aggregate over ALL current silver rows
CREATE OR REFRESH MATERIALIZED VIEW gold_region_totals AS
  SELECT region,
         SUM(amount_cents) AS revenue_cents,
         COUNT(*)          AS sale_count,
         MAX(sold_at)      AS last_sale_at
  FROM silver_sales
  GROUP BY region;

# The same pair in Python (Lakeflow / pyspark.pipelines spelling)
from pyspark import pipelines as dp
from pyspark.sql import functions as F

@dp.table  # streaming table
def silver_sales():
    return (dp.read_stream("bronze_sales")
              .where(F.col("amount_cents").isNotNull())
              .withColumn("amount_cents", F.col("amount_cents").cast("bigint"))
              .withColumn("sold_at", F.to_timestamp("sold_at_raw")))

@dp.materialized_view
def gold_region_totals():
    return (dp.read("silver_sales")           # batch read -> full-correct aggregate
              .groupBy("region")
              .agg(F.sum("amount_cents").alias("revenue_cents"),
                   F.count("*").alias("sale_count"),
                   F.max("sold_at").alias("last_sale_at")))

Step-by-step explanation.

1. Silver is append-only. Each bronze row maps to at most one silver row after cleaning. A streaming table is the exact fit: process each row once, checkpoint progress, never re-scan history. STREAM(bronze_sales) marks the read as streaming, so the engine treats new bronze rows incrementally.
2. Gold is an aggregate over all rows. SUM(amount_cents) GROUP BY region is not something a single-pass append can maintain by itself the way a plain projection can — the current total per region depends on every row seen so far. A materialized view persists that result and refreshes it.
3. The batch edge. FROM silver_sales (no STREAM) tells the engine gold reads silver as a batch snapshot. The engine can still refresh incrementally under the hood, but the contract is "the correct aggregate over current silver," not "each new silver row once."
4. Why not a streaming aggregate. You can express windowed streaming aggregates, but for an unbounded "total per region forever" a materialized view is simpler and always correct: it recomputes (incrementally where possible) rather than forcing you to manage streaming aggregation state and watermarks.
5. The pairing generalises. Append-heavy cleaning layers → streaming tables. Correctness-over-all-rows layers (totals, joins to dimensions, top-N) → materialized views. This split is the backbone of a Lakeflow medallion.

Output.

Layer	Kind	Refresh cost	Correctness
silver_sales	streaming table	O(new rows)	each row once
gold_region_totals	materialized view	O(delta), else O(full)	correct over all current rows

Rule of thumb. Ask "does each source row map to one output row I want to touch once?" → streaming table. Ask "must this be the correct answer over all current rows (an aggregate, a join, a top-N)?" → materialized view. Source semantics decide, not a guess about which is faster.

Worked example — temporary views and the dlt.read vs dlt.read_stream distinction

Detailed explanation. Not every node needs to be persisted. A temporary view (@dlt.view / @dp.temporary_view) is an unmaterialised query used to factor logic or stage a transformation; it is not published as a table. The dlt.read vs dlt.read_stream choice on reading it determines whether the edge is batch or streaming, independent of how the upstream was defined.

• Temporary view. Logic reuse; not a published dataset; recomputed each time it's read.
• dlt.read. Batch read of a dataset (full snapshot).
• dlt.read_stream. Streaming read (incremental, append-only source).

Question. Use a temporary view to factor a shared enrichment, then consume it once as a stream (silver) and once as a batch (a quality report).

Input.

Node	Kind	Read style
enriched	temporary view	defined once
silver_events	streaming table	dlt.read_stream("enriched")
qa_report	materialized view	dlt.read("enriched")

Code.

import dlt
from pyspark.sql import functions as F

@dlt.view(comment="Shared enrichment; not published as a table.")
def enriched():
    return (dlt.read_stream("bronze_events")
              .withColumn("domain", F.regexp_extract("url", r"https?://([^/]+)", 1)))

@dlt.table  # streaming table: consume the enrichment incrementally
def silver_events():
    return dlt.read_stream("enriched").where(F.col("domain") != "")

@dlt.materialized_view  # batch read of the same logic for a QA snapshot
def qa_report():
    return (dlt.read("enriched")
              .groupBy("domain")
              .agg(F.count("*").alias("rows"),
                   F.sum(F.when(F.col("domain") == "", 1).otherwise(0)).alias("bad_domain")))

Step-by-step explanation.

1. The temporary view factors logic. enriched computes the domain column once, in one place. Both downstream nodes reuse it, so the regex lives in a single definition — the declarative equivalent of a shared function.
2. It is not a table. @dlt.view produces no published Delta table; it is inlined/recomputed wherever it is read. Use it for intermediate logic you do not want to persist or expose.
3. read_stream makes a streaming edge. silver_events consumes enriched incrementally; because enriched itself reads bronze as a stream, the whole chain stays append-incremental and silver processes each event once.
4. read makes a batch edge. qa_report reads enriched as a batch snapshot to compute counts — a materialized view that recomputes on refresh. The same logical view is consumed two different ways.
5. The lesson. How you read (read vs read_stream) sets the edge type; what you declare (@dlt.table vs @dlt.materialized_view vs @dlt.view) sets whether and how the node is persisted. These two axes are independent.

Output.

Node	Published table?	Edge into it	Recompute
enriched	no (view)	streaming (from bronze)	each read
silver_events	yes (streaming table)	streaming	O(new rows)
qa_report	yes (materialized view)	batch	O(delta)/O(full)

Rule of thumb. Reach for a temporary view to factor shared logic without publishing a table; pick read_stream when you want each row once and read when you want a correct batch snapshot. The read style and the dataset kind are separate decisions.

Worked example — evolving a materialized view when its query changes

Detailed explanation. A frequent operational question: what happens when you change a materialized view's query (add a column, change the grouping)? Because the view is its query result, the engine simply re-derives it; there is no migration script. Understanding full-vs-incremental refresh behaviour here prevents surprise costs.

• Change. Add channel to the group-by of a gold aggregate.
• Effect. The engine recomputes the view to match the new definition.
• Cost. Adding a grouping key usually forces a full recompute the first time; steady-state refreshes are incremental again.

Question. Change gold_region_totals to also group by channel and explain the recompute behaviour.

Input.

Aspect	Before	After
Group keys	region	region, channel
First refresh	—	full recompute (new grain)
Subsequent refreshes	incremental	incremental again

Code.

-- Just change the query. No ALTER, no backfill script, no migration.
CREATE OR REFRESH MATERIALIZED VIEW gold_region_totals AS
  SELECT region,
         channel,                         -- new grouping key
         SUM(amount_cents) AS revenue_cents,
         COUNT(*)          AS sale_count
  FROM silver_sales
  GROUP BY region, channel;

Step-by-step explanation.

1. The definition is the schema. Because a materialized view is its query result, editing the query is editing the table. You do not write ALTER TABLE ... ADD COLUMN; you change the SELECT and re-run the pipeline.
2. New grain forces a recompute. Grouping by region, channel changes the result's grain, so the first refresh after the change recomputes the whole view — the old per-region rows cannot be reused for a per-region-per-channel result.
3. Steady state returns to incremental. Once the new definition is materialised, subsequent refreshes go back to incremental: only rows for changed (region, channel) groups get recomputed, assuming the query shape supports it.
4. No downstream break at read time. Consumers reading gold_region_totals see the new schema after the refresh completes; the engine keeps the prior version available until the new one is ready, so there is no half-migrated window.
5. Contrast with a streaming table. You cannot arbitrarily change a streaming table's schema the same way, because it holds accumulated append state and a checkpoint; certain changes require a full refresh of the streaming table. Materialized views are the more edit-friendly of the two.

Output.

Refresh	Rows recomputed	Trigger
First after grain change	all groups (full)	new group-by key
Next (few new sales)	only affected (region, channel)	incremental
Query unchanged, no new data	none	no-op

Rule of thumb. Editing a materialized view's query needs no migration — the engine re-derives the result. Expect one full recompute when you change the grain or aggregation, then incremental refreshes resume. Reserve schema-changing edits for materialized views where they are cheap; plan a full refresh when a streaming table's shape must change.

Senior interview question on streaming tables vs materialized views

A senior interviewer might ask: "Design a three-layer Lakeflow pipeline for a clickstream: raw JSON files landing in cloud storage, a cleaned event layer, and two gold outputs — a per-day click count and an always-correct 'sessions per user' table. For each dataset say whether it's a streaming table or a materialized view and defend the choice on source semantics and cost."

Solution Using streaming tables for bronze/silver and materialized views for the gold aggregates

-- BRONZE: streaming table — Auto Loader over raw files, each file once
CREATE OR REFRESH STREAMING TABLE bronze_clicks
  AS SELECT *, _metadata.file_path AS source_file
     FROM STREAM read_files('/Volumes/main/raw/clicks/', format => 'json');

-- SILVER: streaming table — clean + type, append-incremental, quality-gated
CREATE OR REFRESH STREAMING TABLE silver_clicks (
  CONSTRAINT has_user  EXPECT (user_id IS NOT NULL) ON VIOLATION DROP ROW,
  CONSTRAINT has_url   EXPECT (url IS NOT NULL)     ON VIOLATION DROP ROW
) AS
  SELECT user_id,
         url,
         to_timestamp(ts_raw) AS ts,
         source_file
  FROM STREAM(bronze_clicks);

-- GOLD 1: materialized view — per-day click count (correct over all rows)
CREATE OR REFRESH MATERIALIZED VIEW gold_daily_clicks AS
  SELECT to_date(ts) AS day, COUNT(*) AS clicks
  FROM silver_clicks
  GROUP BY to_date(ts);

-- GOLD 2: materialized view — sessions per user (needs full-history correctness)
CREATE OR REFRESH MATERIALIZED VIEW gold_sessions_per_user AS
  WITH ordered AS (
    SELECT user_id, ts,
           LAG(ts) OVER (PARTITION BY user_id ORDER BY ts) AS prev_ts
    FROM silver_clicks
  ),
  marked AS (
    SELECT user_id, ts,
           CASE WHEN prev_ts IS NULL
                 OR ts > prev_ts + INTERVAL 30 MINUTES
                THEN 1 ELSE 0 END AS is_new_session
    FROM ordered
  )
  SELECT user_id, SUM(is_new_session) AS sessions
  FROM marked
  GROUP BY user_id;

Step-by-step trace.

Dataset	Kind	Why this kind	Refresh cost
bronze_clicks	streaming table	append-only file ingest, one pass/file	O(new files)
silver_clicks	streaming table	one cleaned row per raw row	O(new rows)
gold_daily_clicks	materialized view	correct COUNT over all rows	O(delta)
gold_sessions_per_user	materialized view	window/LAG needs full-history view	O(delta)/O(full)

After deployment, raw files land and bronze_clicks consumes each exactly once. silver_clicks cleans incrementally, dropping and counting rows that fail the two expectations. gold_daily_clicks maintains a correct per-day count. gold_sessions_per_user uses a windowed session-gap calculation that inherently needs to see a user's full history in order, which is why it is a materialized view rather than a streaming aggregate — the engine recomputes affected users incrementally where it can.

Output:

Requirement	Chosen kind	Correctness property
Raw ingest	streaming table	exactly-once file processing
Cleaned events	streaming table	each row once; quality-gated
Per-day clicks	materialized view	correct total per day
Sessions per user	materialized view	correct over ordered full history

Why this works — concept by concept:

• Streaming tables for append layers — bronze and silver each map source rows one-to-one, so the exactly-once, checkpointed streaming table is the natural fit; cost stays O(new rows).
• Materialized views for aggregates — gold_daily_clicks is a running total that must be correct over all rows; the materialized view persists and refreshes that result rather than forcing manual streaming-aggregate state.
• Window functions demand a materialized view — LAG(...) OVER (PARTITION BY user_id ORDER BY ts) needs an ordered view of each user's full history; a single-pass streaming table cannot express it, but a materialized view recomputes it correctly and incrementally.
• Expectations on the streaming layer — quality gates live where the raw-to-clean transition happens (silver), so bad rows are dropped and counted before any gold aggregate consumes them.
• Cost — streaming layers are O(new rows) per run; materialized views are O(delta) when the engine can compute the change and O(full) only when the grain/shape forces it. Compared to a hand-written full nightly rebuild of all four datasets, the declarative version pays incremental cost by default and only recomputes what changed.

Data transformation
Topic — data-transformation
Data-transformation problems on medallion layers

Practice →

Streaming Topic — streaming Streaming problems on streaming tables vs views

Practice →

---

3. Data quality and expectations — EXPECT, DROP, FAIL

data quality expectations are constraints declared on a table — EXPECT warns, EXPECT OR DROP filters, EXPECT OR FAIL halts — with pass/fail metrics in the event log

The mental model in one line: an expectation is a boolean constraint you attach to a streaming table or materialized view definition, and its ON VIOLATION clause decides what happens to rows that fail it — keep them but record the violation (EXPECT, the default warn/track behaviour), drop them from the output while counting them (EXPECT OR DROP), or fail the entire pipeline update (EXPECT OR FAIL) — so that data quality expectations become executable, versioned, observable constraints that live with the table rather than in a separate validation job. This is one of the features that most cleanly separates lakeflow declarative pipelines from hand-wired ETL: quality is a property of the dataset, enforced on every row, with metrics you can query.

The three violation actions.

• EXPECT (warn / track). The default: rows that violate the constraint are kept in the output, but the violation is counted and surfaced in the event log. Use for soft rules you want visibility on without discarding data.
• EXPECT OR DROP (filter). Violating rows are removed from this table's output and counted. Downstream never sees them; the count tells you how bad the source is. Use for rows that are unusable but shouldn't stop the pipeline.
• EXPECT OR FAIL (halt). A single violating row fails the pipeline update. Use for invariants that must never be breached (a primary key must be non-null, an amount must be non-negative) where continuing would corrupt downstream.
• The trade-off ladder. Warn (keep + count) → drop (remove + count) → fail (stop). Pick the least-disruptive action that still protects downstream correctness.

Declaring expectations — Python and SQL.

• Single, Python. @dlt.expect("name", "constraint"), @dlt.expect_or_drop(...), @dlt.expect_or_fail(...) stacked as decorators on the table function.
• Multiple, Python. @dlt.expect_all({...}), @dlt.expect_all_or_drop({...}), @dlt.expect_all_or_fail({...}) take a dict of name → constraint.
• SQL. CONSTRAINT name EXPECT (predicate) [ON VIOLATION DROP ROW | FAIL UPDATE] inside the CREATE ... TABLE (...) column list.
• The predicate. Any SQL boolean expression over the row's columns — amount_cents > 0, email RLIKE '@', ts <= current_timestamp().

Observability — where the metrics land.

• The event log. Every pipeline writes a structured event log (available as a queryable table/view) containing, per expectation per run: passing records, failing records, and the action taken.
• The pipeline UI. Renders the same metrics visually — a data-quality panel per table showing pass rate over time.
• Alerting. Because the event log is a table, you can build alerts ("dropped-row rate > 2%") as ordinary queries in Lakeflow Jobs.
• Why it matters. Quality stops being invisible. A silent 0.5% drop that a hand-wired filter would hide becomes a number you can trend and alert on.

The quarantine pattern — keep the bad rows instead of dropping them.

• The idea. EXPECT OR DROP discards violating rows entirely. Sometimes you want to route them to a side table for inspection/reprocessing instead.
• The build. Add a boolean is_valid column via the same predicate, publish a clean table filtered to is_valid and a quarantine table filtered to NOT is_valid.
• When to use. Regulatory data you may not discard, or noisy sources where bad rows are fixable and worth replaying.
• The payoff. Nothing is lost; downstream gets only clean rows; an analyst can query the quarantine table to find and fix root causes.

Worked example — the three actions on one silver table

Detailed explanation. The cleanest way to learn the actions is to put all three on one table: a soft rule that warns, a usability rule that drops, and an invariant that fails. Seeing them together clarifies which action protects what.

• Warn. event_ts should not be in the future — track but keep.
• Drop. user_id null — unusable, remove.
• Fail. amount_cents negative — an impossible invariant; stop the pipeline.

Question. Declare a silver table with one warn, one drop, and one fail expectation, and describe the outcome for a batch containing one violation of each.

Input.

Constraint	Action	Rationale
event_ts <= current_timestamp()	warn (EXPECT)	clock issues; keep + count
user_id IS NOT NULL	drop	unusable row
amount_cents >= 0	fail	must never happen

Code.

CREATE OR REFRESH STREAMING TABLE silver_orders (
  CONSTRAINT ts_not_future  EXPECT (event_ts <= current_timestamp()),                 -- warn + track
  CONSTRAINT has_user       EXPECT (user_id IS NOT NULL)     ON VIOLATION DROP ROW,   -- filter
  CONSTRAINT nonneg_amount  EXPECT (amount_cents >= 0)       ON VIOLATION FAIL UPDATE -- halt
) AS
  SELECT order_id, user_id,
         CAST(amount_cents AS BIGINT) AS amount_cents,
         to_timestamp(event_ts_raw)   AS event_ts
  FROM STREAM(bronze_orders);

# Same three actions in Python
import dlt
from pyspark.sql import functions as F

@dlt.table
@dlt.expect("ts_not_future", "event_ts <= current_timestamp()")     # warn
@dlt.expect_or_drop("has_user", "user_id IS NOT NULL")              # drop
@dlt.expect_or_fail("nonneg_amount", "amount_cents >= 0")           # fail
def silver_orders():
    return (dlt.read_stream("bronze_orders")
              .withColumn("amount_cents", F.col("amount_cents").cast("bigint"))
              .withColumn("event_ts", F.to_timestamp("event_ts_raw")))

Step-by-step explanation.

1. Warn keeps the row. A record with event_ts slightly in the future (a skewed client clock) passes through to silver_orders unchanged, but the event log increments ts_not_future's failing count. You get visibility without data loss.
2. Drop removes the row. A record with user_id IS NULL never appears in silver_orders; the has_user failing count rises. Downstream gold aggregates never see the unusable row, and you can trend the drop rate to detect a broken upstream.
3. Fail stops everything. A single record with amount_cents = -5 fails the entire pipeline update. Nothing is published for this run; the last-good version of silver_orders remains readable. This is correct for an invariant whose violation means the source is corrupt.
4. Order of severity. The three actions form a ladder: warn (observe), drop (protect downstream but continue), fail (protect downstream and stop). You choose per constraint based on how catastrophic a violation is.
5. Metrics for all three. Every action, including fail, records counts. Even the run that failed tells you which expectation failed and how many rows breached it — far better than a hand-written filter that silently drops or a job that dies with a stack trace.

Output.

Constraint	Violating row fate	Event-log effect	Pipeline effect
ts_not_future (warn)	kept	failing count +1	continues
has_user (drop)	removed	dropped count +1	continues
nonneg_amount (fail)	—	failing count +1	update fails

Rule of thumb. Map each rule to the least-disruptive action that still protects downstream: warn for observability, drop for unusable-but-non-fatal rows, fail only for true invariants. Every action still emits counts, so you never trade enforcement for blindness.

Worked example — the quarantine pattern (route bad rows, don't discard them)

Detailed explanation. When you cannot afford to drop bad rows — regulatory retention, or fixable/replayable data — build a quarantine. Compute an is_valid flag once, then split into a clean table and a quarantine table. Nothing is lost and downstream still sees only clean rows.

• Flag. One boolean expression capturing all validity rules.
• Clean. WHERE is_valid.
• Quarantine. WHERE NOT is_valid, carrying a reason.

Question. Implement a quarantine split for silver_orders where an order is valid only if it has a user, a non-negative amount, and a parseable timestamp.

Input.

Output	Filter	Consumer
silver_orders_clean	is_valid	gold aggregates
silver_orders_quarantine	NOT is_valid	data-quality analyst

Code.

import dlt
from pyspark.sql import functions as F

VALID_RULES = {
    "has_user":      "user_id IS NOT NULL",
    "nonneg_amount": "amount_cents >= 0",
    "parseable_ts":  "event_ts IS NOT NULL",
}
VALID_EXPR = " AND ".join(f"({r})" for r in VALID_RULES.values())

@dlt.view
def orders_flagged():
    return (dlt.read_stream("bronze_orders")
              .withColumn("amount_cents", F.col("amount_cents").cast("bigint"))
              .withColumn("event_ts", F.to_timestamp("event_ts_raw"))
              .withColumn("is_valid", F.expr(VALID_EXPR)))

@dlt.table(comment="Only clean rows; feeds gold.")
@dlt.expect_all_or_drop(VALID_RULES)   # metrics + guarantee clean
def silver_orders_clean():
    return dlt.read_stream("orders_flagged").where("is_valid")

@dlt.table(comment="Bad rows retained for inspection / replay.")
def silver_orders_quarantine():
    return (dlt.read_stream("orders_flagged")
              .where("NOT is_valid")
              .withColumn("quarantined_at", F.current_timestamp()))

Step-by-step explanation.

1. One source of truth for validity. VALID_RULES is a dict of named predicates; VALID_EXPR ANDs them into a single boolean. The orders_flagged view computes is_valid once so the clean and quarantine tables agree exactly on the definition of "valid."
2. Clean table gets metrics too. silver_orders_clean filters to is_valid and declares expect_all_or_drop(VALID_RULES). The filter guarantees cleanliness; the expectations produce per-rule drop metrics so you know which rule fails most.
3. Quarantine retains everything else. silver_orders_quarantine keeps NOT is_valid rows and stamps quarantined_at. No data is lost — an analyst can query this table, find the dominant failure reason, fix the source, and replay.
4. Downstream sees only clean. Gold aggregates read silver_orders_clean, so a bad row can never skew a total. The quarantine is a side channel, off the main lineage into gold.
5. Why not just drop. EXPECT OR DROP alone discards the row's content — you keep a count but lose the data. Quarantine keeps the content, which is mandatory for regulated data and invaluable for debugging noisy sources.

Output.

Row	is_valid	Lands in	Downstream impact
user present, amount 500, ts ok	true	silver_orders_clean	feeds gold
user null	false	silver_orders_quarantine	inspected/replayed
amount -5	false	silver_orders_quarantine	inspected/replayed

Rule of thumb. When "drop" would lose data you must keep or want to fix, quarantine instead: compute is_valid once, publish clean and quarantine tables from the same flag, and keep the drop metrics on the clean table so you still trend quality.

Worked example — querying the event log for quality trends

Detailed explanation. Expectations are only as useful as your ability to see them. The event log is a queryable dataset; extracting per-expectation pass/fail counts lets you build alerts and dashboards with plain SQL — no custom instrumentation.

• Source. The pipeline event log (exposed as a table/view for the pipeline).
• Signal. flow_progress events carry data_quality metrics with per-expectation counts.
• Use. Trend the drop rate; alert when it crosses a threshold.

Question. Write a query over the event log that reports, per expectation, passing and failing record counts for the latest run.

Input.

Field	Meaning
event_type	e.g. flow_progress
details:flow_progress.data_quality.expectations	array of per-expectation metrics
name / passed_records / failed_records	per-expectation fields

Code.

-- Latest-run data-quality summary from the pipeline event log.
-- (event_log(...) exposes the pipeline's structured events as a table.)
WITH exp AS (
  SELECT
    timestamp,
    explode(
      from_json(
        details:flow_progress.data_quality.expectations,
        'array<struct<name:string, dataset:string,
                      passed_records:bigint, failed_records:bigint>>'
      )
    ) AS e
  FROM event_log(pipeline_id => 'abc-123-declarative-pipeline')
  WHERE event_type = 'flow_progress'
)
SELECT
  e.dataset,
  e.name                                   AS expectation,
  SUM(e.passed_records)                    AS passed,
  SUM(e.failed_records)                    AS failed,
  ROUND(100.0 * SUM(e.failed_records)
        / NULLIF(SUM(e.passed_records + e.failed_records), 0), 3) AS fail_pct
FROM exp
GROUP BY e.dataset, e.name
ORDER BY fail_pct DESC;

Step-by-step explanation.

1. The event log is data. event_log(...) returns the pipeline's structured events as rows, so quality metrics are queryable with ordinary SQL — no scraping of driver logs.
2. Expectations live under flow_progress. Each flow-progress event carries a data_quality.expectations array; from_json parses it into a typed array and explode turns each expectation into its own row.
3. Aggregate per expectation. Summing passed_records and failed_records per (dataset, name) gives the run's totals; fail_pct turns them into a rate you can threshold.
4. Sort by worst. Ordering by fail_pct DESC surfaces the expectation degrading most, which is exactly what an on-call engineer wants first.
5. From query to alert. Wrap this in a Lakeflow Job that runs after the pipeline and raises an alert when fail_pct > 2 for any drop/fail expectation. Quality alerting becomes a scheduled SQL query, not bespoke code.

Output.

dataset	expectation	passed	failed	fail_pct
silver_orders	has_user	998,900	1,100	0.110
silver_orders	ts_not_future	999,850	150	0.015
silver_orders	nonneg_amount	1,000,000	0	0.000

Rule of thumb. Treat the event log as a first-class table: parse the data_quality.expectations array, trend fail_pct per expectation, and alert on thresholds from a scheduled query. If you cannot see your expectation metrics, you are only half-using the feature.

Senior interview question on data-quality expectations

A senior interviewer might ask: "You ingest a payments feed where roughly 1% of rows have a null merchant_id, a tiny fraction have negative amounts (which must never reach finance), and some timestamps are clock-skewed into the future. Design the silver-layer expectations so finance never sees a bad row, nothing is silently lost, and on-call gets alerted when quality degrades. Show the pipeline and the alert query."

Solution Using warn/drop/fail expectations plus a quarantine table and an event-log alert

# silver_payments.py — quality-first silver with quarantine + metrics
import dlt
from pyspark.sql import functions as F

RULES = {
    "has_merchant":  "merchant_id IS NOT NULL",
    "parseable_ts":  "event_ts IS NOT NULL",
}
VALID_EXPR = " AND ".join(f"({r})" for r in RULES.values())

@dlt.view
def payments_flagged():
    return (dlt.read_stream("bronze_payments")
              .withColumn("amount_cents", F.col("amount_cents").cast("bigint"))
              .withColumn("event_ts", F.to_timestamp("event_ts_raw"))
              .withColumn("is_valid", F.expr(VALID_EXPR)))

@dlt.table(comment="Finance-safe payments; clean rows only.")
@dlt.expect_or_fail("nonneg_amount", "amount_cents >= 0")   # invariant: never ship negatives
@dlt.expect("ts_not_future", "event_ts <= current_timestamp()")  # warn: clock skew
@dlt.expect_all_or_drop(RULES)                              # drop unusable, keep metrics
def silver_payments_clean():
    return dlt.read_stream("payments_flagged").where("is_valid")

@dlt.table(comment="Retained bad rows for replay/audit.")
def silver_payments_quarantine():
    return (dlt.read_stream("payments_flagged")
              .where("NOT is_valid")
              .withColumn("quarantined_at", F.current_timestamp()))

-- Alert query (schedule after the pipeline in a Lakeflow Job):
-- page when any drop/fail expectation exceeds 2% for the latest run.
WITH exp AS (
  SELECT explode(from_json(
           details:flow_progress.data_quality.expectations,
           'array<struct<name:string,dataset:string,passed_records:bigint,failed_records:bigint>>'
         )) AS e
  FROM event_log(pipeline_id => 'pay-pipeline')
  WHERE event_type = 'flow_progress'
)
SELECT e.name,
       ROUND(100.0*SUM(e.failed_records)
             /NULLIF(SUM(e.passed_records+e.failed_records),0),3) AS fail_pct
FROM exp
GROUP BY e.name
HAVING fail_pct > 2.0;

Step-by-step trace.

Rule	Action	What finance sees	What ops sees
nonneg_amount	fail update	never a negative amount	run fails + count if breached
has_merchant / parseable_ts	drop (via filter + expect_all_or_drop)	no null-merchant rows	drop counts trend
ts_not_future	warn	future-ts rows kept + flagged	failing count trend
bad rows	quarantined	nothing	queryable side table

After deployment, silver_payments_clean contains only rows with a merchant and a parseable timestamp, and the pipeline fails loudly if any negative amount appears — so finance is structurally protected from the one catastrophic case. Null-merchant and unparseable rows are dropped from clean but retained in silver_payments_quarantine, so nothing is lost. Clock-skewed rows are kept but flagged. The alert query pages on-call when any drop/fail expectation crosses 2%.

Output:

Concern	Mechanism	Guarantee
Negatives to finance	expect_or_fail	impossible (update halts)
Unusable rows downstream	filter + expect_all_or_drop	removed + counted
Data loss	quarantine table	nothing discarded
Silent degradation	event-log alert query	paged at >2%

Why this works — concept by concept:

• EXPECT OR FAIL for invariants — a negative payment is a correctness catastrophe, so it fails the whole update; the last-good table stays readable and finance never ingests the bad batch.
• EXPECT OR DROP for unusable rows — null-merchant and unparseable-timestamp rows are removed from the clean output and counted, protecting downstream without stopping the pipeline.
• EXPECT (warn) for soft anomalies — clock-skewed timestamps are kept and flagged, giving visibility without discarding legitimate-but-late data.
• Quarantine for zero-loss retention — routing NOT is_valid rows to a side table satisfies audit/replay needs that a bare drop cannot, while keeping the main lineage clean.
• Cost — expectations are per-row boolean evaluations (near-free relative to the transformation) plus one extra quarantine table write; the event-log alert is one scheduled query. Compared to a separate validation job plus custom metrics plumbing, quality-as-declaration is dramatically cheaper and never drifts out of sync with the table it guards.

ETL
Topic — etl
ETL problems on data-quality gates

Practice →

Data transformation Topic — data-transformation Data-transformation problems on validation + quarantine

Practice →

---

4. Streaming and batch unified — Auto Loader and AUTO CDC

auto loader ingests files incrementally, AUTO CDC INTO applies a change feed as SCD Type 1 or Type 2 — one definition serves batch and streaming

The mental model in one line: lakeflow declarative pipelines unifies batch and streaming by making the dataset definition independent of the run cadence — auto loader (cloudFiles) incrementally discovers and ingests new files into a streaming table whether the pipeline runs once (batch trigger) or continuously, and AUTO CDC INTO (the successor to APPLY CHANGES INTO) consumes a change data capture feed and maintains a target streaming table as an SCD Type 1 (overwrite) or Type 2 (history) dimension — so the same declarative code handles a nightly batch and a continuous stream by changing only how the pipeline is triggered. You stop writing separate batch and streaming implementations of the same logic.

Auto Loader — incremental file ingestion.

• What it is. A Structured Streaming source (format("cloudFiles")) that tracks which files in a cloud directory it has already processed, so each file is ingested exactly once without you diffing directory listings.
• Schema handling. cloudFiles.schemaLocation persists the inferred schema; cloudFiles.schemaEvolutionMode controls how new columns are handled (add them, rescue them, or fail).
• Rescue column. _rescued_data captures fields that don't match the expected schema, so malformed/extra data is never silently dropped.
• Batch or stream. With a batch trigger (e.g. availableNow), Auto Loader processes all new files then stops; with continuous execution it keeps watching. Same source, two cadences.

AUTO CDC INTO — apply a change feed (was APPLY CHANGES).

• What it replaces. AUTO CDC and AUTO CDC FROM SNAPSHOT are the current APIs; they replace APPLY CHANGES / APPLY CHANGES FROM SNAPSHOT with the same syntax. The old form still runs, but new code should use AUTO CDC.
• What it does. Reads a stream of change rows (insert/update/delete) and maintains a target streaming table, handling out-of-order events via SEQUENCE BY and deletes via APPLY AS DELETE WHEN.
• SCD Type 1. STORED AS SCD TYPE 1 — the target holds the current row per key; updates overwrite, deletes remove.
• SCD Type 2. STORED AS SCD TYPE 2 — the target keeps history, opening/closing rows with __START_AT / __END_AT so you can query the state as of any point in time.

Sequencing, keys, and deletes — the correctness knobs.

• KEYS. The business key(s) that identify a logical row across changes.
• SEQUENCE BY. The column that orders changes for a key (a monotonic version, LSN, or timestamp). This is what makes out-of-order delivery safe — the engine applies changes in sequence order, not arrival order.
• APPLY AS DELETE WHEN. A predicate marking a change row as a delete (e.g. operation = 'DELETE'), so tombstones are handled rather than treated as upserts.
• COLUMNS * EXCEPT (...). Drop CDC-control columns (operation, sequenceNum) from the target so it holds only business columns.

Incremental refresh across the two dataset kinds.

• Streaming tables append incrementally. Auto Loader and AUTO CDC targets consume only new input; a restart resumes from the checkpoint, never reprocessing.
• Materialized views refresh incrementally. Downstream gold aggregates over a CDC-maintained dimension recompute only the affected groups when the query shape allows.
• Full refresh escape hatch. A full refresh reprocesses a streaming table from the source's start (needed after certain schema/logic changes); a normal refresh is incremental.
• The unification. Because cadence is a trigger setting, you develop against a batch trigger and promote to continuous without rewriting the transformation.

Worked example — Auto Loader bronze ingestion with schema evolution

Detailed explanation. The canonical bronze layer: Auto Loader watches a cloud directory, ingests each new JSON file once, persists the schema, and rescues unexpected fields. The same definition runs as a nightly batch or a continuous stream depending only on the pipeline trigger.

• Source. cloudFiles, format json.
• Schema. persisted at schemaLocation; evolution mode addNewColumns.
• Safety. _rescued_data catches off-schema fields.

Question. Write an Auto Loader bronze streaming table that ingests JSON orders once each, evolves the schema when new columns appear, and rescues unexpected fields.

Input.

Option	Value	Purpose
cloudFiles.format	json	source format
cloudFiles.schemaLocation	a volume path	persist inferred schema
cloudFiles.schemaEvolutionMode	addNewColumns	absorb new columns
cloudFiles.inferColumnTypes	true	type inference

Code.

import dlt

@dlt.table(
    name="bronze_orders",
    comment="Raw orders via Auto Loader; each file processed exactly once.",
    table_properties={"quality": "bronze"},
)
def bronze_orders():
    return (
        spark.readStream.format("cloudFiles")
             .option("cloudFiles.format", "json")
             .option("cloudFiles.schemaLocation", "/Volumes/main/schemas/orders")
             .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
             .option("cloudFiles.inferColumnTypes", "true")
             .load("/Volumes/main/raw/orders/")
             .selectExpr("*", "_metadata.file_path AS source_file",
                              "_metadata.file_modification_time AS ingested_from_ts")
    )

-- The SQL equivalent using read_files (Auto Loader under the hood)
CREATE OR REFRESH STREAMING TABLE bronze_orders
  AS SELECT *,
            _metadata.file_path              AS source_file,
            _metadata.file_modification_time AS ingested_from_ts
     FROM STREAM read_files(
            '/Volumes/main/raw/orders/',
            format               => 'json',
            schemaEvolutionMode  => 'addNewColumns'
          );

Step-by-step explanation.

1. Exactly-once files. Auto Loader tracks processed files in its own state (under the pipeline's storage), so a file is ingested once even across restarts. You never diff the directory or maintain a "seen files" list.
2. Persisted schema. schemaLocation stores the inferred schema so restarts don't re-infer from scratch. inferColumnTypes promotes strings to real types (bigint, timestamp) instead of leaving everything as strings.
3. Schema evolution. addNewColumns means when a producer starts sending a new field, Auto Loader adds it to the table (the stream restarts to pick up the new schema) rather than dropping it. Other modes: rescue (route to _rescued_data), failOnNewColumns, none.
4. Rescue safety net. Any field that doesn't fit the schema lands in _rescued_data (present automatically), so you never silently lose data from a malformed record — you can inspect and reprocess it.
5. Batch or stream, same code. Run the pipeline with a batch/availableNow-style trigger and it drains all new files then stops; run it continuously and it keeps watching the directory. The definition is identical; only the trigger differs.

Output.

Behaviour	Result
New file arrives	ingested once; source_file recorded
Duplicate re-list of same file	skipped (tracked as processed)
Producer adds a column	column added to bronze_orders
Off-schema field	captured in _rescued_data

Rule of thumb. For file ingestion, always use Auto Loader with a persisted schemaLocation and an explicit schemaEvolutionMode; rely on _rescued_data so nothing is lost. Never hand-roll directory diffing — exactly-once file tracking is the whole point.

Worked example — AUTO CDC INTO for an SCD Type 2 dimension

Detailed explanation. The headline CDC feature: feed a change stream into AUTO CDC INTO and get a full SCD Type 2 history dimension with open/close timestamps, correct even under out-of-order events. This is the successor to APPLY CHANGES INTO, same syntax.

• Source. A CDC feed (from Debezium/Auto Loader) with operation and a monotonic sequenceNum.
• Target. A streaming table stored as SCD Type 2.
• Correctness. SEQUENCE BY sequenceNum orders changes; APPLY AS DELETE WHEN handles tombstones.

Question. Build an SCD Type 2 dim_customers from a customer change feed, keeping history and handling deletes.

Input.

Clause	Value	Role
KEYS	(customer_id)	business key
SEQUENCE BY	sequenceNum	orders changes
APPLY AS DELETE WHEN	operation = 'DELETE'	tombstone handling
STORED AS	SCD TYPE 2	keep history

Code.

-- 1. Declare the target streaming table (schema inferred, or declare it explicitly).
CREATE OR REFRESH STREAMING TABLE dim_customers;

-- 2. Define the AUTO CDC flow that maintains it as SCD Type 2.
CREATE FLOW dim_customers_cdc AS AUTO CDC INTO
  dim_customers
FROM
  STREAM(cdc_customers)                 -- change feed: insert/update/delete rows
KEYS
  (customer_id)
APPLY AS DELETE WHEN
  operation = 'DELETE'
SEQUENCE BY
  sequenceNum                            -- out-of-order safe: applied in this order
COLUMNS * EXCEPT
  (operation, sequenceNum)               -- keep only business columns
STORED AS
  SCD TYPE 2;                            -- history: __START_AT / __END_AT

# The Python equivalent (create_auto_cdc_flow, successor to apply_changes)
import dlt
from pyspark.sql.functions import col

dlt.create_streaming_table("dim_customers")

dlt.create_auto_cdc_flow(
    target      = "dim_customers",
    source      = "cdc_customers",
    keys        = ["customer_id"],
    sequence_by = col("sequenceNum"),
    apply_as_deletes = "operation = 'DELETE'",
    except_column_list = ["operation", "sequenceNum"],
    stored_as_scd_type = 2,      # SCD Type 2 history
)

Step-by-step explanation.

1. Declare the target first. CREATE OR REFRESH STREAMING TABLE dim_customers creates the empty target; the CDC flow populates it. The two-step (table + flow) is required — the flow needs a target streaming table to apply changes into.
2. Keys define the logical row. KEYS (customer_id) tells the engine which column identifies the same customer across an insert, several updates, and maybe a delete. All change rows for one customer_id are one logical entity's history.
3. Sequence makes out-of-order safe. SEQUENCE BY sequenceNum is the correctness linchpin: even if change events arrive out of order (common with Kafka partitions), the engine applies them in sequenceNum order, so the final history is correct regardless of arrival order.
4. Deletes are tombstones, not upserts. APPLY AS DELETE WHEN operation = 'DELETE' marks delete rows so the engine closes the current SCD Type 2 record (sets __END_AT) instead of upserting a garbage row. Without this, a delete would look like an update.
5. Type 2 keeps history. STORED AS SCD TYPE 2 makes each change open a new current row and close the prior one via __START_AT / __END_AT. You can then query "what did this customer look like on 2026-06-01?" — impossible with a Type 1 overwrite.

Output.

customer_id	name	__START_AT	__END_AT
7	Ada (old address)	seq 10	seq 25
7	Ada (new address)	seq 25	(open)
9	Grace	seq 12	seq 40 (deleted)

Rule of thumb. For any dimension that needs history, use AUTO CDC INTO ... STORED AS SCD TYPE 2 with KEYS, a monotonic SEQUENCE BY, and APPLY AS DELETE WHEN. The sequence column — not arrival order — guarantees correctness; never skip it.

Worked example — the same logic as nightly batch and continuous stream

Detailed explanation. The unification payoff: write the transformation once, then choose batch or streaming at trigger time. The bronze/silver/CDC definitions above do not change; only the pipeline's execution mode does. This is what "streaming + batch as one definition" means concretely.

• Development / cost-sensitive. Triggered (batch) mode: run, process all available data, stop, release compute.
• Low-latency. Continuous mode: keep running, process as data arrives.
• No code change. Same tables, same expectations, same CDC flow.

Question. Show how the identical pipeline runs as a triggered nightly batch and as a continuous stream, and what changes.

Input.

Mode	Trigger	Compute lifetime	Latency
Triggered (batch)	schedule / manual	starts, drains, stops	minutes–hours
Continuous	always on	runs indefinitely	seconds

Code.

# The pipeline code is IDENTICAL in both modes — bronze (Auto Loader),
# silver (streaming table), dim (AUTO CDC). Nothing here mentions cadence.
import dlt
from pyspark.sql import functions as F

@dlt.table
def bronze_orders():
    return (spark.readStream.format("cloudFiles")
              .option("cloudFiles.format", "json")
              .option("cloudFiles.schemaLocation", "/Volumes/main/schemas/orders")
              .load("/Volumes/main/raw/orders/"))

@dlt.table
@dlt.expect_or_drop("valid_amount", "amount_cents > 0")
def silver_orders():
    return (dlt.read_stream("bronze_orders")
              .withColumn("amount_cents", F.col("amount_cents").cast("bigint")))

// Cadence is a PIPELINE SETTING, not code. Triggered (batch) nightly:
{
  "name": "orders-pipeline",
  "continuous": false,          // triggered: run, drain, stop  (batch)
  "development": false,
  "libraries": [ { "glob": { "include": "/Workspace/pipelines/orders/**" } } ]
}

// The SAME pipeline promoted to continuous (low-latency streaming):
{
  "name": "orders-pipeline",
  "continuous": true,           // continuous: always on  (streaming)
  "development": false,
  "libraries": [ { "glob": { "include": "/Workspace/pipelines/orders/**" } } ]
}

Step-by-step explanation.

1. The transformation is cadence-agnostic. bronze_orders, silver_orders, and any CDC flow are defined purely by their queries. Nothing in the code says "batch" or "streaming" — that decision lives entirely in the pipeline configuration.
2. Triggered mode drains and stops. With continuous: false, an update starts compute, processes all currently-available files/changes exactly once (resuming from checkpoints), then stops and releases the cluster. This is the cheap, schedule-it-nightly mode.
3. Continuous mode stays on. With continuous: true, the same pipeline runs indefinitely, processing new files and changes within seconds of arrival. You pay for always-on compute in exchange for low latency.
4. Checkpoints make the switch safe. Because streaming tables checkpoint progress, flipping between modes does not reprocess history — a triggered run and a continuous run share the same exactly-once semantics; only when work happens differs.
5. Develop cheap, promote for latency. Teams typically develop and backfill in triggered mode (cheaper, deterministic) and promote latency-critical pipelines to continuous. The unification means promotion is a config flag, not a rewrite — the core selling point over maintaining separate batch and streaming jobs.

Output.

Aspect	Triggered (batch)	Continuous (streaming)
Code	identical	identical
Compute	starts/stops per run	always on
Latency	schedule interval	seconds
Cost	pay per run	pay continuously
Semantics	exactly-once	exactly-once

Rule of thumb. Write the transformation once and treat cadence as a deployment setting (continuous true/false). Develop and backfill in triggered mode; promote to continuous only where seconds-level latency justifies always-on compute. Never fork the code into separate batch and streaming versions.

Senior interview question on unified streaming and batch

A senior interviewer might ask: "You get a Debezium change feed for a products table landing as JSON files in cloud storage, and you need a current-state products table for the app plus a full-history version for analytics — both correct under out-of-order events. Build the Lakeflow pipeline with Auto Loader ingestion and AUTO CDC for SCD Type 1 and Type 2, and explain how you'd run it nightly first and then promote to continuous."

Solution Using Auto Loader bronze plus two AUTO CDC targets (SCD Type 1 and Type 2)

-- 1. BRONZE: Auto Loader ingests the Debezium change files exactly once.
CREATE OR REFRESH STREAMING TABLE bronze_products_cdc
  AS SELECT
       payload:after.product_id::BIGINT   AS product_id,
       payload:after.name::STRING         AS name,
       payload:after.price_cents::BIGINT  AS price_cents,
       payload:op::STRING                 AS operation,   -- c/u/d
       payload:ts_ms::BIGINT              AS sequenceNum, -- monotonic order
       _metadata.file_path                AS source_file
     FROM STREAM read_files('/Volumes/main/raw/products_cdc/', format => 'json');

-- 2. CURRENT-STATE (SCD Type 1) for the app — overwrite per key, honor deletes.
CREATE OR REFRESH STREAMING TABLE products_current;

CREATE FLOW products_current_cdc AS AUTO CDC INTO
  products_current
FROM STREAM(bronze_products_cdc)
KEYS (product_id)
APPLY AS DELETE WHEN operation = 'd'
SEQUENCE BY sequenceNum
COLUMNS * EXCEPT (operation, sequenceNum, source_file)
STORED AS SCD TYPE 1;

-- 3. FULL-HISTORY (SCD Type 2) for analytics — keep every version.
CREATE OR REFRESH STREAMING TABLE products_history;

CREATE FLOW products_history_cdc AS AUTO CDC INTO
  products_history
FROM STREAM(bronze_products_cdc)
KEYS (product_id)
APPLY AS DELETE WHEN operation = 'd'
SEQUENCE BY sequenceNum
COLUMNS * EXCEPT (operation, sequenceNum, source_file)
STORED AS SCD TYPE 2;

// Run nightly first (triggered), then flip continuous for low latency — no code change.
{ "name": "products-cdc", "continuous": false, "development": false }   // phase 1: batch
{ "name": "products-cdc", "continuous": true,  "development": false }   // phase 2: streaming

Step-by-step trace.

Layer	Kind	Key clauses	Result
bronze_products_cdc	streaming table (Auto Loader)	read_files json	each change file once
products_current	AUTO CDC SCD 1	KEYS, SEQUENCE BY, delete-on-d	one current row per product
products_history	AUTO CDC SCD 2	same + STORED AS SCD TYPE 2	full history w/ __START_AT/__END_AT
cadence	config	continuous false→true	batch first, then streaming

After deployment, Auto Loader ingests each Debezium file exactly once into bronze, parsing the op and ts_ms into operation and sequenceNum. Two AUTO CDC flows read the same bronze change feed: one maintains products_current (SCD Type 1 — the app reads the latest price per product, deletes remove the row) and one maintains products_history (SCD Type 2 — analytics can reconstruct any product's price on any date). Both apply changes in sequenceNum order, so an out-of-order Kafka delivery still produces the correct final state. The pipeline runs nightly in triggered mode during rollout, then flips to continuous: true for seconds-level freshness with no code change.

Output:

Consumer	Table	Shape	Freshness after promotion
App	products_current	current row per product	seconds
Analytics	products_history	full SCD Type 2 history	seconds
Both	—	correct under out-of-order events	via SEQUENCE BY

Why this works — concept by concept:

• Auto Loader bronze — read_files/cloudFiles gives exactly-once file ingestion of the Debezium output with schema persistence; no directory diffing, no double-processing.
• AUTO CDC INTO (was APPLY CHANGES) — one declarative statement turns a raw change feed into a maintained dimension; the same feed drives both a Type 1 and a Type 2 target from two flows.
• SEQUENCE BY for out-of-order safety — changes are applied in sequenceNum (ts_ms) order, not arrival order, so partitioned/out-of-order delivery still yields the correct current state and history.
• APPLY AS DELETE WHEN — Debezium op = 'd' rows close the SCD Type 2 record and remove the SCD Type 1 row instead of being mis-applied as upserts.
• Cost — bronze is O(new files); each AUTO CDC target is O(changes) per run; promotion to continuous trades always-on compute for latency. Compared to hand-writing MERGE-based SCD logic with your own out-of-order handling and two separate batch/streaming jobs, the declarative version is a fraction of the code and cannot drift between the batch and streaming implementations because there is only one.

Streaming
Topic — streaming
Streaming problems on CDC and Auto Loader

Practice →

ETL Topic — etl ETL problems on SCD Type 1 / Type 2 dimensions

Practice →

---

5. Ops, migration, and interview signals

Development vs production mode, serverless unit economics, and the DLT → Lakeflow migration — how you run and pay for a declarative pipeline

The mental model in one line: running a lakeflow declarative pipelines job well means understanding three operational levers — development mode (reuses a cluster and disables retries for fast iteration) versus production mode (fresh compute, automatic retries, resilient); the unit economics of serverless pipeline compute (you pay DBUs for the graph the engine actually runs, so right-sizing and incremental refresh are the cost dials); and the migration from Delta Live Tables to Lakeflow (your @dlt code keeps running, the forward-looking spelling is pyspark.pipelines / Spark Declarative Pipelines, and AUTO CDC supersedes APPLY CHANGES). These are exactly the areas interviewers use to separate people who have run a pipeline in production from people who have only read the getting-started guide.

Development vs production mode.

• Development mode. Reuses the same cluster across updates (no cold-start between iterations) and disables automatic retries so failures surface immediately with full stack traces. This is for the write-run-fix loop.
• Production mode. Provisions fresh compute per run, enables automatic retries with escalating backoff, and is resilient to transient infra failures. This is for scheduled/continuous runs.
• The toggle. A single pipeline setting (development: true/false), not a code change. The same definition behaves differently in each mode.
• The trap. Leaving a pipeline in development mode in production hides transient failures you should be retrying and can leave a long-lived cluster running; leaving it in production while iterating makes every fix wait for a cold start.

Serverless unit economics.

• What you pay for. Serverless pipelines bill DBUs for the compute the engine actually uses to run the graph. Incremental refresh (streaming tables, incremental materialized views) means you pay for deltas, not full rebuilds.
• Photon. The vectorized engine accelerates most SQL/DataFrame work; it raises the DBU rate but usually lowers total cost by finishing faster.
• The cost dials. (1) Incremental over full refresh — avoid unnecessary full refreshes. (2) Triggered over continuous where latency allows — don't pay for always-on compute you don't need. (3) Right-size / let serverless scale — don't pin an oversized cluster.
• When declarative is cheaper — and when it isn't. Cheaper when incremental refresh replaces nightly full rebuilds and when the engine consolidates many small jobs. More expensive if you run everything continuously, force full refreshes, or use a pipeline for a trivial one-shot transform a plain job would do for less.

The DLT → Lakeflow migration.

• Your code still runs. Existing import dlt / @dlt.table / @dlt.expect* pipelines continue to work under Lakeflow Declarative Pipelines — the rename did not break the API.
• The forward spelling. New/portable code uses from pyspark import pipelines as dp with @dp.table, @dp.materialized_view, @dp.temporary_view — this is Spark Declarative Pipelines, the open-source core (Apache Spark 4.0) that Lakeflow extends.
• CDC API. AUTO CDC / create_auto_cdc_flow replace APPLY CHANGES / apply_changes (same syntax); prefer the new names in new code.
• What to change first. Nothing is forced. Migrate opportunistically: new pipelines in the pyspark.pipelines spelling, new CDC flows with AUTO CDC, and leave stable @dlt pipelines alone until you touch them.

Interview signals on ops and migration.

• Name development vs production mode and why (cluster reuse + no retries vs fresh compute + retries) — senior signal.
• Say "my old @dlt code keeps running" when asked about the rename — required answer.
• Name incremental refresh and triggered-vs-continuous as the two biggest cost dials — senior signal.
• Know that AUTO CDC replaced APPLY CHANGES with the same syntax — senior signal.

Worked example — configuring dev vs prod correctly

Detailed explanation. The same pipeline should run in development mode while you iterate and production mode once deployed. Setting this wrong is a common on-call and cost incident. Walk through both configs and what each changes.

• Dev. cluster reuse, no retries, fast fail.
• Prod. fresh compute, retries, resilience.
• One flag. development.

Question. Provide the development and production pipeline settings and state the behavioural difference for a transient S3 read error.

Input.

Setting	Development	Production
development	true	false
Cluster	reused across runs	fresh per run
Retries	disabled	enabled (backoff)
Transient error	fails fast	retried automatically

Code.

// Development — fast iterate: reuse cluster, fail fast, see the stack trace.
{
  "name": "orders-pipeline",
  "development": true,
  "continuous": false,
  "channel": "CURRENT",
  "libraries": [ { "glob": { "include": "/Workspace/pipelines/orders/**" } } ]
}

// Production — resilient: fresh compute, auto-retry transient failures.
{
  "name": "orders-pipeline",
  "development": false,
  "continuous": false,
  "channel": "CURRENT",
  "libraries": [ { "glob": { "include": "/Workspace/pipelines/orders/**" } } ]
}

Step-by-step explanation.

1. Dev reuses the cluster. With development: true, consecutive updates reuse the running cluster, so your edit-run-edit loop doesn't pay a cold start each time. Great for iteration, wrong for production (a long-lived idle cluster costs money and masks fresh-start bugs).
2. Dev disables retries. A transient S3 read error in dev fails immediately with the full stack trace — exactly what you want while debugging, because a silent retry would hide the error you're trying to see.
3. Prod uses fresh compute. With development: false, each run provisions clean compute, so a run never inherits corrupted state from a previous one — important for reproducibility and for surfacing "works only because the cluster was warm" bugs.
4. Prod retries transient failures. The same S3 error in production is retried with backoff; a blip in object storage doesn't page on-call. This is the resilience you want for scheduled/continuous runs.
5. It's a flag, not a fork. The pipeline code is identical; only development differs. The classic incident is shipping with development: true left on — transient failures stop retrying and a cluster lingers. Always flip it to false for deployed pipelines.

Output.

Event	Development mode	Production mode
Edit + re-run	reuse cluster (fast)	fresh cluster (slow)
Transient S3 error	fails fast (visible)	retried (resilient)
Idle between runs	cluster lingers	compute released
Use case	iteration	scheduled/continuous

Rule of thumb. Develop with development: true (cluster reuse, fail-fast) and deploy with development: false (fresh compute, auto-retry). The single most common ops mistake is leaving development mode on in production — check the flag in code review.

Worked example — driving down cost with incremental refresh and triggered mode

Detailed explanation. Two dials move pipeline cost the most: refreshing incrementally instead of fully, and running triggered instead of continuous where latency permits. Quantifying them on a concrete workload shows why "declarative" isn't automatically cheaper — you have to use the incremental path.

• Dial 1. Incremental materialized-view refresh vs forced full recompute.
• Dial 2. Triggered (nightly drain) vs continuous (always on).
• Anti-pattern. Continuous + frequent full refresh = maximum cost.

Question. For a gold aggregate over a 2-billion-row silver table with ~5M new rows/day, compare the cost of incremental-triggered vs full-continuous.

Input.

Choice	Rows processed/day	Compute lifetime
Incremental + triggered	~5M (delta)	~minutes/day
Full + continuous	~2B repeatedly	24h always-on

Code.

-- CHEAP: a materialized view the engine can refresh incrementally.
-- Simple grouped SUM/COUNT over an append-mostly silver table qualifies.
CREATE OR REFRESH MATERIALIZED VIEW gold_daily_revenue AS
  SELECT to_date(event_ts) AS day, region,
         SUM(amount_cents) AS revenue_cents,
         COUNT(*)          AS orders
  FROM silver_orders
  GROUP BY 1, 2;
-- Run it in a TRIGGERED pipeline (continuous:false) once/hour or nightly.

-- EXPENSIVE anti-pattern to avoid: a non-incrementalizable query that forces
-- a full recompute, run CONTINUOUSLY. e.g. a view whose logic can't be
-- computed from deltas will rescan all 2B rows on every refresh.

# Cost estimate: incremental-triggered vs full-continuous
NEW_ROWS_DAY   = 5_000_000
TOTAL_ROWS     = 2_000_000_000
RUNS_PER_DAY_TRIGGERED   = 24        # hourly
# Incremental+triggered processes only the delta per run:
incremental_rows = NEW_ROWS_DAY                     # ~5M/day total work
# Full+continuous reprocesses everything on every refresh:
full_rows = TOTAL_ROWS * RUNS_PER_DAY_TRIGGERED     # ~48B/day of scanning
print(f"incremental/day: {incremental_rows:,} rows scanned")
print(f"full+continuous/day: {full_rows:,} rows scanned  (~{full_rows/incremental_rows:,.0f}x)")
# incremental/day: 5,000,000 rows scanned
# full+continuous/day: 48,000,000,000 rows scanned  (~9,600x)

Step-by-step explanation.

1. Incremental refresh scans the delta. A simple grouped SUM/COUNT over an append-mostly silver table is incrementalizable: the engine updates only the (day, region) groups touched by the ~5M new rows, so daily work is ~O(new rows), not O(all rows).
2. Full refresh scans everything. If the query shape prevents incremental computation (or you force a full refresh), each refresh rescans all 2B rows. Multiply by refresh frequency and the cost explodes.
3. Triggered vs continuous. Triggered mode runs, drains the delta, and stops — you pay for minutes of compute. Continuous mode keeps compute always on; for a gold aggregate that only needs hourly freshness, that's paying 24h for a few minutes of actual work.
4. The multiplier is enormous. The estimate shows full+continuous scanning on the order of thousands of times more rows than incremental+triggered for the same business output. This is why "declarative" is not automatically cheap — the savings come from using the incremental, triggered path.
5. Design for incremental. Keep gold queries incrementalizable (grouped aggregates, joins the engine can maintain), avoid gratuitous full refreshes, and run continuously only where seconds-level latency is a real requirement.

Output.

Strategy	Rows scanned/day	Compute	Relative cost
Incremental + triggered	~5M	minutes	1x
Full + triggered hourly	~48B	hours	~1000s x
Full + continuous	~48B+	24h always-on	highest

Rule of thumb. The two biggest cost dials are incremental-vs-full refresh and triggered-vs-continuous execution. Keep gold queries incrementalizable, run triggered unless you truly need continuous, and reserve full refreshes for schema/logic changes. A declarative pipeline is only cheap if you stay on the incremental path.

Worked example — migrating a DLT pipeline to the Lakeflow spelling

Detailed explanation. Migration is opportunistic, not forced. The safest path: leave working @dlt code alone, write new code in the pyspark.pipelines spelling, and switch CDC flows to AUTO CDC. Show a side-by-side so the mechanical mapping is obvious.

• Keep. Existing @dlt.table pipelines run unchanged.
• Adopt. from pyspark import pipelines as dp for new pipelines.
• Switch. apply_changes → create_auto_cdc_flow; APPLY CHANGES INTO → AUTO CDC INTO.

Question. Show the DLT-era code and its Lakeflow-spelling equivalent for a streaming table and a CDC flow, and state what actually has to change.

Input.

DLT era	Lakeflow spelling
import dlt	from pyspark import pipelines as dp
@dlt.table	@dp.table
@dlt.expect_or_drop	@dp.expect_or_drop
dlt.apply_changes(...)	dp.create_auto_cdc_flow(...)
APPLY CHANGES INTO	AUTO CDC INTO

Code.

# ---- DLT era (still runs today, unchanged) ----
import dlt
from pyspark.sql.functions import col

@dlt.table
@dlt.expect_or_drop("valid_amount", "amount_cents > 0")
def silver_orders():
    return dlt.read_stream("bronze_orders")

dlt.create_streaming_table("dim_customers")
dlt.apply_changes(                       # legacy CDC API
    target="dim_customers", source="cdc_customers",
    keys=["customer_id"], sequence_by=col("sequenceNum"),
    apply_as_deletes="operation = 'DELETE'", stored_as_scd_type=2,
)

# ---- Lakeflow / Spark Declarative Pipelines spelling (forward-looking) ----
from pyspark import pipelines as dp
from pyspark.sql.functions import col

@dp.table
@dp.expect_or_drop("valid_amount", "amount_cents > 0")
def silver_orders():
    return dp.read_stream("bronze_orders")

dp.create_streaming_table("dim_customers")
dp.create_auto_cdc_flow(                 # AUTO CDC replaces apply_changes
    target="dim_customers", source="cdc_customers",
    keys=["customer_id"], sequence_by=col("sequenceNum"),
    apply_as_deletes="operation = 'DELETE'", stored_as_scd_type=2,
)

Step-by-step explanation.

1. Nothing is forced. The top block still runs under Lakeflow Declarative Pipelines. The rename did not deprecate @dlt; you migrate because you want the forward-compatible, open-source-aligned spelling, not because you must.
2. The decorator mapping is mechanical. dlt → dp (from pyspark import pipelines as dp), @dlt.table → @dp.table, @dlt.expect_or_drop → @dp.expect_or_drop. The function bodies and semantics are identical.
3. The CDC API is the real change. dlt.apply_changes(...) becomes dp.create_auto_cdc_flow(...) (same arguments), and SQL APPLY CHANGES INTO becomes AUTO CDC INTO. The syntax matches, so it's a find-and-replace, but the new names are what documentation and new features target.
4. Migrate opportunistically. Write new pipelines in the dp spelling and new CDC flows with AUTO CDC; leave stable @dlt pipelines until you're editing them anyway. A big-bang rewrite buys nothing.
5. Why bother at all. pyspark.pipelines is the open-source Spark Declarative Pipelines API (Spark 4.0), so code in that spelling is portable to any Spark that supports it, and it's where new capabilities land first. It's a strategic, not urgent, migration.

Output.

Element	Change required?	Effort
@dlt.table pipelines	no (keep running)	none
New pipeline code	adopt dp spelling	trivial
apply_changes	→ create_auto_cdc_flow	rename
APPLY CHANGES INTO	→ AUTO CDC INTO	rename

Rule of thumb. Don't rewrite working @dlt pipelines. Adopt from pyspark import pipelines as dp for new code and AUTO CDC for new CDC flows; treat migration as opportunistic modernisation, and you keep both the old investment and the forward path.

Senior interview question on running a declarative pipeline in production

A senior interviewer might ask: "You're taking a Lakeflow pipeline from prototype to production. It has Auto Loader bronze, a couple of streaming silver tables, and two gold materialized views. Walk me through the dev-to-prod configuration, how you'd control cost, how you'd monitor it, and what you'd say when leadership asks 'is this cheaper than the old Airflow + Spark jobs?'"

Solution Using production mode, triggered incremental refresh, event-log monitoring, and an honest cost story

// 1. Production pipeline config: fresh compute + retries, triggered nightly,
//    serverless so you pay for the graph the engine actually runs.
{
  "name": "orders-medallion",
  "development": false,        // fresh compute, auto-retry transient failures
  "continuous": false,         // triggered: drain the delta, stop, release compute
  "serverless": true,          // pay DBUs for actual work; incremental where possible
  "channel": "CURRENT",
  "photon": true,
  "libraries": [ { "glob": { "include": "/Workspace/pipelines/orders/**" } } ]
}

-- 2. Monitoring: run this after each update (as a Lakeflow Job task) to trend
--    per-dataset row counts and any expectation failures from the event log.
SELECT
  timestamp,
  details:flow_progress.status                                   AS status,
  details:flow_progress.metrics.num_output_rows                  AS output_rows,
  details:flow_progress.data_quality.dropped_records            AS dropped
FROM event_log(pipeline_id => 'orders-medallion')
WHERE event_type = 'flow_progress'
ORDER BY timestamp DESC
LIMIT 50;

# 3. Cost story (illustrative): incremental+triggered vs the old full nightly rebuild.
OLD_FULL_REBUILD_ROWS = 2_000_000_000     # Airflow+Spark rescanned everything nightly
NEW_INCREMENTAL_ROWS  = 5_000_000         # Lakeflow scans only the delta
savings = 1 - NEW_INCREMENTAL_ROWS / OLD_FULL_REBUILD_ROWS
print(f"rows scanned/night: {OLD_FULL_REBUILD_ROWS:,} -> {NEW_INCREMENTAL_ROWS:,}")
print(f"scan reduction: {savings:.2%}")
# rows scanned/night: 2,000,000,000 -> 5,000,000
# scan reduction: 99.75%

Step-by-step trace.

Concern	Setting / action	Effect
Reliability	development:false	fresh compute + auto-retry
Cost — cadence	continuous:false	pay per run, not always-on
Cost — compute	serverless:true + incremental MVs	DBUs for deltas, not rebuilds
Speed	photon:true	faster runs, lower total cost
Monitoring	event-log query in a Job	row counts + drop metrics trended
Cost narrative	incremental vs full rebuild	~99% fewer rows scanned/night

After promotion, the pipeline runs in production mode (fresh compute, retries), triggered nightly on serverless so compute is released between runs. Gold materialized views refresh incrementally, so the nightly work is ~O(new rows) rather than the old O(all rows) full rebuild. An event-log query runs after each update to trend output rows and dropped-record counts, wired to an alert. The honest cost answer to leadership: cheaper because the old jobs did full nightly rebuilds and this does incremental refresh — but only as long as we stay triggered and incremental; flip everything to continuous with full refreshes and the comparison reverses.

Output:

Dimension	Old (Airflow + Spark)	New (Lakeflow, tuned)
Reliability	manual retries	auto-retry, fresh compute
Nightly rows scanned	~2B (full rebuild)	~5M (incremental)
Compute between runs	cluster management	released (serverless)
Monitoring	scattered logs	event-log metrics + alerts
Cost	full-rebuild baseline	~99% fewer rows scanned

Why this works — concept by concept:

• Production mode — development:false gives fresh compute and automatic retries, so transient infra blips don't page on-call and runs don't inherit stale cluster state.
• Triggered + serverless — continuous:false on serverless means you pay DBUs only while a run actually processes the delta, then compute is released; you're not funding a 24h cluster for a nightly job.
• Incremental materialized views — gold refreshes scale with new rows, not total rows, which is the entire source of the cost win over the old full-rebuild jobs.
• Event-log monitoring — querying flow_progress events turns row counts and drop metrics into trendable, alertable data with no custom instrumentation.
• Cost — the win is real (~99% fewer rows scanned nightly) but conditional: it depends on staying incremental and triggered. Framing it honestly — "cheaper because we replaced full rebuilds with incremental refresh; it would not be cheaper if we ran continuous full refreshes" — is the senior signal, because it shows you understand the unit economics rather than assuming declarative equals cheap.

Design
Topic — design
Design problems on pipeline operations and cost

Practice →

ETL
Topic — etl
ETL problems on incremental refresh and migration

Practice →

---

Cheat sheet — Lakeflow Declarative Pipelines recipes

• The rename in one line. Delta Live Tables (DLT) is now Lakeflow Declarative Pipelines, the transformation pillar of the Lakeflow family (Lakeflow Connect = ingest, Declarative Pipelines = transform, Lakeflow Jobs = orchestrate). Your import dlt / @dlt.table code keeps running unchanged; the forward spelling is from pyspark import pipelines as dp (Spark Declarative Pipelines, Spark 4.0).
• Declarative vs imperative. You declare target datasets + their queries; the engine derives the DAG from dlt.read/dlt.read_stream/STREAM() references, orders execution, manages checkpoints and retries, enforces expectations, and emits an event log. No orchestrator, no checkpointLocation, no manual task edges in your code.
• Streaming table vs materialized view. Streaming table = append/incremental over an append-only source, each row processed once, checkpointed (bronze/silver). Materialized view = persisted query result kept fresh by incremental (or full) recompute, always correct over all current rows (gold aggregates, joins, window functions). Pick by source semantics, not by "which is faster".
• The three dataset kinds. @dlt.table / CREATE OR REFRESH STREAMING TABLE (streaming table); @dlt.materialized_view / CREATE OR REFRESH MATERIALIZED VIEW (materialized view); @dlt.view / @dp.temporary_view (temporary view — logic reuse, not published). Read style sets the edge: read_stream/STREAM() = streaming edge, read/plain FROM = batch edge.
• Expectations — three actions. EXPECT (predicate) warns + counts but keeps the row; EXPECT (predicate) ON VIOLATION DROP ROW filters + counts; EXPECT (predicate) ON VIOLATION FAIL UPDATE halts the update. Python: @dlt.expect / @dlt.expect_or_drop / @dlt.expect_or_fail, plus expect_all* for dicts. Pick the least-disruptive action that protects downstream.
• Quarantine pattern. Compute one is_valid boolean, publish a clean table (WHERE is_valid + expect_all_or_drop for metrics) and a quarantine table (WHERE NOT is_valid + quarantined_at). Nothing lost; downstream stays clean; analysts fix and replay from quarantine. Use for regulated or fixable data where a bare drop is unacceptable.
• Observability. Every pipeline writes a structured event log you can query: event_log(pipeline_id => ...), event_type = 'flow_progress', parse details:flow_progress.data_quality.expectations. Trend fail_pct per expectation and alert from a scheduled Lakeflow Job query. If you can't see the metrics, you're half-using expectations.
• Auto Loader recipe. spark.readStream.format("cloudFiles") (or SQL read_files(...)) with cloudFiles.format, a persisted cloudFiles.schemaLocation, and an explicit cloudFiles.schemaEvolutionMode (addNewColumns / rescue / failOnNewColumns). Exactly-once file tracking; off-schema fields land in _rescued_data. Never hand-roll directory diffing.
• AUTO CDC INTO (was APPLY CHANGES). CREATE OR REFRESH STREAMING TABLE t; CREATE FLOW f AS AUTO CDC INTO t FROM STREAM(src) KEYS (k) APPLY AS DELETE WHEN <cond> SEQUENCE BY <seq> COLUMNS * EXCEPT (op, seq) STORED AS SCD TYPE {1|2};. Python: create_auto_cdc_flow(...). SEQUENCE BY (not arrival order) guarantees out-of-order correctness; SCD Type 2 adds __START_AT/__END_AT.
• Batch vs streaming is a setting. Same code; cadence is the pipeline's continuous flag: false = triggered (drain, stop, cheap), true = continuous (always-on, seconds latency). Develop/backfill triggered, promote to continuous only where latency demands. Streaming-table checkpoints make the switch safe.
• Dev vs prod mode. development: true reuses the cluster and disables retries (fast iteration, fail-fast). development: false provisions fresh compute and auto-retries (resilient). The classic incident is shipping with dev mode left on — check it in code review.
• Cost dials. (1) Incremental over full refresh — keep gold queries incrementalizable; reserve full refresh for schema/logic changes. (2) Triggered over continuous where latency allows. (3) Serverless + Photon + right-sizing. Declarative is cheaper than full nightly rebuilds only if you stay on the incremental, triggered path.
• Migration checklist. Keep @dlt pipelines as-is; write new pipelines with pyspark.pipelines (dp); switch CDC flows from apply_changes/APPLY CHANGES INTO to create_auto_cdc_flow/AUTO CDC INTO. Opportunistic, not big-bang — you keep the old investment and the forward path.

Frequently asked questions

Is Delta Live Tables the same as Lakeflow Declarative Pipelines?

Yes — lakeflow declarative pipelines is Delta Live Tables, renamed. At the 2025 Data + AI Summit Databricks reorganised its data-engineering products into the Lakeflow family — Lakeflow Connect (managed ingestion), Lakeflow Declarative Pipelines (transformation, the former DLT), and Lakeflow Jobs (orchestration, the former Databricks Jobs/Workflows). The declarative framework itself is unchanged in substance: you still define streaming tables, materialized views, and expectations, and your existing import dlt / @dlt.table code continues to run. What's new is the family framing, a forward-looking pyspark.pipelines API aligned with the open-source Spark Declarative Pipelines in Apache Spark 4.0, and the AUTO CDC API superseding APPLY CHANGES.

Streaming table vs materialized view — when do I pick each?

Pick a streaming table when your source is append-only and you want to process each row exactly once and never re-scan history — this is the natural fit for bronze ingestion (via auto loader) and silver cleaning, where each input row maps to at most one output row and cost stays O(new rows). Pick a materialized view when the dataset is the result of a query that must be correct over all current rows — aggregates, dimension joins, top-N, and anything using window functions — because a materialized view persists that result and refreshes it (incrementally where the query shape allows, fully otherwise) rather than forcing you to manage streaming-aggregate state and watermarks. The decision is driven by source semantics (append-once vs correct-over-all-rows), not by a guess about which runs faster.

What are expectations and what do EXPECT / DROP / FAIL do?

Data quality expectations are boolean constraints you declare directly on a table definition; every row is checked and the results are recorded in the event log. EXPECT (predicate) is the warn/track action — violating rows are kept but counted, giving you visibility without discarding data. EXPECT (predicate) ON VIOLATION DROP ROW filters violating rows out of the table's output while still counting them, protecting downstream without stopping the run. EXPECT (predicate) ON VIOLATION FAIL UPDATE fails the entire pipeline update on a single violation, which is the right action for true invariants (a non-null key, a non-negative amount) where continuing would corrupt downstream. In Python these are @dlt.expect, @dlt.expect_or_drop, and @dlt.expect_or_fail (with expect_all* variants for dicts of rules). Choose the least-disruptive action that still protects downstream correctness.

What replaced APPLY CHANGES INTO?

AUTO CDC INTO (SQL) and create_auto_cdc_flow() (Python) replaced APPLY CHANGES INTO and apply_changes() — with the same syntax. You declare a target streaming table, then a flow that reads a change data capture feed and maintains the target: KEYS identify the logical row, SEQUENCE BY orders changes so out-of-order delivery is handled correctly, APPLY AS DELETE WHEN turns tombstones into proper deletes, and STORED AS SCD TYPE 1 or SCD TYPE 2 chooses current-state overwrite or full history (with __START_AT/__END_AT). The old APPLY CHANGES form still works, but Databricks recommends AUTO CDC for new code, and AUTO CDC gets the newer capabilities such as bitemporal storage and column-subset partial updates.

Does my old @dlt code still work?

Yes. The DLT → Lakeflow rename did not break the API — existing pipelines using import dlt, @dlt.table, @dlt.view, @dlt.expect*, dlt.read/dlt.read_stream, and dlt.apply_changes continue to run under Lakeflow Declarative Pipelines. Migration is opportunistic, not forced: the recommended pattern is to leave stable @dlt pipelines alone, write new pipelines in the forward-looking from pyspark import pipelines as dp spelling (@dp.table, @dp.materialized_view, @dp.temporary_view), and use AUTO CDC / create_auto_cdc_flow for new CDC flows instead of APPLY CHANGES / apply_changes. That way you keep the existing investment while new work lands on the open-source-aligned, forward-compatible API.

Is a declarative pipeline cheaper than a hand-wired Spark job?

It depends entirely on whether you use the incremental path — declarative is not automatically cheaper. It is cheaper when incremental refresh replaces full rebuilds: a streaming table processes only new rows and a materialized view refreshes only the affected groups, so a nightly job that used to rescan two billion rows now scans the few million that changed. It is also cheaper when the engine consolidates many small jobs and removes the human cost of debugging checkpoint drift and restart ordering. But it can be more expensive if you run everything continuously when triggered would do, force full refreshes on non-incrementalizable queries, or use a pipeline for a trivial one-shot transform a plain job would handle for less. The senior framing is honest: cheaper because we replaced full rebuilds with incremental refresh and released compute between triggered runs — conditional on staying incremental and triggered, not an automatic property of the word "declarative".

Practice on PipeCode

• Drill the ETL practice library → for the medallion-pipeline, incremental-load, SCD Type 1/Type 2, and data-quality-gate problems that Lakeflow pipelines are built to solve.
• Rehearse on the streaming practice library → for the Auto Loader, streaming-table, AUTO CDC, and out-of-order-event scenarios senior interviewers open with when declarative pipelines are on the table.
• Sharpen the modelling axis with the data-transformation practice library → for the streaming-table-vs-materialized-view, temporary-view factoring, and quarantine-split patterns that separate a clean medallion from a tangled one.
• Layer in the design practice library → for the dev-vs-prod-mode, unit-economics, and DLT→Lakeflow migration trade-offs that turn a working pipeline into a production-grade one.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the declarative-ETL decision map — streaming table vs materialized view, warn/drop/fail, batch vs continuous — against real graded inputs.

Lock in declarative-pipeline muscle memory

Docs explain the syntax. PipeCode drills explain the decision — when a streaming table beats a materialized view, when EXPECT OR FAIL is the only safe action, when AUTO CDC's SEQUENCE BY saves you from out-of-order corruption, when triggered-and-incremental is the difference between a cheap pipeline and a runaway bill. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs data engineers actually face on Databricks Lakeflow.

Practice ETL problems →
Practice streaming problems →