Databricks DE · Topic 5.3

Data-Quality Expectations

Boolean rules on each row in an SDP pipeline. On violation: warn (keep + metric), drop (remove), or fail (abort) — plus the quarantine pattern. This deep dive walks each with the code.

warn / drop / failexpect_all (dict)quarantine patternevent-log metrics

The one-liner

Expectations validate each record against a SQL boolean. Choose the action: warn (keep + record metric), drop (remove bad rows), fail (abort the update), or quarantine (route bad rows aside).

🛂 Analogy: expectations are a bouncer checking IDs. Warn = let them in but note it in the logbook; Drop = turn them away quietly; Fail = shut the whole door until the problem's sorted.

Pick an action — see what happens to bad rows

▸ rule: amount >= 0 · two rows violate it

1 · The three actions

Action	Bad row is…	Update	Metric?	Use for
Warn (default)	kept & written	continues	✅ yes	track quality without blocking
Drop	removed before write	continues	✅ yes	keep the table clean
Fail	—	aborts / rolls back	❌ no	truly critical invariants

2 · Syntax — click a language

CREATE OR REFRESH STREAMING TABLE silver_orders (
  CONSTRAINT valid_id  EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW,    -- drop
  CONSTRAINT valid_amt EXPECT (amount >= 0),                                  -- warn (default)
  CONSTRAINT has_date  EXPECT (order_date IS NOT NULL) ON VIOLATION FAIL UPDATE -- fail
) AS SELECT * FROM STREAM bronze_orders;

from pyspark import pipelines as dp

@dp.table
@dp.expect("valid_amt", "amount >= 0")                 # warn
@dp.expect_or_drop("valid_id", "order_id IS NOT NULL") # drop
@dp.expect_or_fail("has_date", "order_date IS NOT NULL")# fail
def silver_orders():
    return spark.readStream.table("bronze_orders")

# group many with a dict: @dp.expect_all / _all_or_drop / _all_or_fail
RULES = {"valid_id": "order_id IS NOT NULL", "valid_amt": "amount >= 0"}
@dp.table
@dp.expect_all_or_drop(RULES)               # drop a row failing ANY rule
def silver_orders2():
    return spark.readStream.table("bronze_orders")

Conditions are SQL booleans — no Python UDFs, external calls, or subqueries referencing other tables. (Legacy @dlt.expect* still works; prefer pyspark.pipelines.)

3 · The quarantine pattern — keep bad rows

Drop discards bad rows; fail halts the pipeline. To retain suspect rows, split the stream on the rules: clean → silver, bad → quarantine.

from pyspark.sql.functions import expr
RULES = {"valid_id": "order_id IS NOT NULL", "valid_amt": "amount >= 0"}
valid = " AND ".join(RULES.values())

@dp.table
def silver_orders():                      # clean rows only
    return spark.readStream.table("bronze_orders").filter(expr(valid))

@dp.table
def quarantined_orders():                 # bad rows kept for review/replay
    return spark.readStream.table("bronze_orders").filter(expr(f"NOT({valid})"))

4 · Metrics & monitoring

Warn and drop emit per-expectation metrics (failing-row counts) to the pipeline UI and the event log — trend quality over time or alert on spikes.
Fail aborts the update, so no metric is recorded for that run — you see the failure, not a count.

Why it matters

Bad data silently flowing to gold breaks dashboards & models — expectations catch it at the pipeline boundary with visible metrics.
"How do you enforce data quality in a pipeline?" → Expectations (warn/drop/fail/quarantine).

🧾 Real-world use case: an orders pipeline drops rows with null order_id, warns on amount < 0 (tracked so analysts notice), and fails the update if a critical reference table is empty.

Uses, edge cases & limitations

Uses: validate ingest at bronze/silver; track quality trends (warn metrics); hard-stop critical loads (fail); quarantine suspect rows.
Edge cases: fail rolls back the whole update (use sparingly); quarantine is a pattern (split the stream), not a keyword; a row failing any rule in an expect_all_or_drop/_or_fail dict triggers the action.
Limitation: SQL-only boolean logic — no UDFs/external calls/subqueries referencing other tables; applies within SDP pipelines only. Warn & drop produce metrics; fail does not (it aborts).

Common gotchas

❌ Using fail for non-critical rules → one bad row halts the pipeline.
❌ Expecting drop to keep bad rows somewhere — it discards them; use the quarantine pattern.
❌ Calling a Python UDF/subquery in an expectation — only SQL booleans.
❌ Forgetting warn still writes the bad rows (only records a metric).
❌ Using legacy @dlt.expect* in new code — prefer from pyspark import pipelines.