Lakeflow Declarative Pipelines data quality: expectations, quarantine tables and monitoring with the event log

Expectations are row-level rules you attach to a pipeline table. They are the cheapest way to stop bad data travelling from bronze to gold, and their results are recorded for every update.

#Three behaviours

Python:

import dlt
from pyspark.sql import functions as F

@dlt.table(comment="Cleaned orders")
@dlt.expect("valid_amount", "amount >= 0")
@dlt.expect_or_drop("has_order_id", "order_id IS NOT NULL")
@dlt.expect_or_fail("known_currency", "currency IN ('EUR','USD','GBP')")
def silver_orders():
    return dlt.read_stream("bronze_orders").withColumn("amount", F.col("amount").cast("decimal(12,2)"))

SQL:

CREATE OR REFRESH STREAMING TABLE silver_orders (
  CONSTRAINT valid_amount   EXPECT (amount >= 0),
  CONSTRAINT has_order_id   EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW,
  CONSTRAINT known_currency EXPECT (currency IN ('EUR','USD','GBP')) ON VIOLATION FAIL UPDATE
)
AS SELECT * FROM STREAM(bronze_orders);

Newer releases expose the Python API as pyspark.pipelines (import pyspark.pipelines as dp); the decorators are equivalent. Use whichever your runtime documents.

#Rule sets shared across tables

Keep rules in one place:

RULES = {
    "has_order_id": "order_id IS NOT NULL",
    "valid_amount": "amount >= 0",
    "recent_date": "order_date >= '2015-01-01' AND order_date <= current_date() + INTERVAL 1 DAY",
}

@dlt.table
@dlt.expect_all_or_drop(RULES)
def silver_orders():
    return dlt.read_stream("bronze_orders")

expect_all, expect_all_or_drop and expect_all_or_fail take a dict. You can also load rules from a Delta table maintained by data owners, and build the dict at pipeline start.

#Quarantine instead of drop

Dropping silently loses data you may need to fix. Split the stream instead:

quarantine_rule = "NOT({0})".format(" AND ".join(f"({r})" for r in RULES.values()))

@dlt.view
def orders_flagged():
    return dlt.read_stream("bronze_orders").withColumn("is_quarantined", F.expr(quarantine_rule))

@dlt.table(comment="Valid orders only")
def silver_orders():
    return dlt.read_stream("orders_flagged").filter("NOT is_quarantined").drop("is_quarantined")

@dlt.table(comment="Rows that failed at least one rule")
def silver_orders_quarantine():
    return (dlt.read_stream("orders_flagged").filter("is_quarantined")
            .withColumn("quarantined_at", F.current_timestamp()))

Watch for NULLs: NOT(amount >= 0) is NULL when amount is NULL, so the row lands in neither table. Wrap rules with coalesce(<rule>, false) when building quarantine_rule.

Tip

Give data owners SELECT on the quarantine table and a simple dashboard. Bad data then becomes a ticket for the source team, not an investigation for the platform team.

#Reading quality metrics from the event log

Every update records expectation results in the pipeline event log. With Unity Catalog, query it with the event_log() table function:

WITH flows AS (
  SELECT timestamp, origin.flow_name AS flow,
         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(TABLE(sales_prod.silver.silver_orders))
  WHERE event_type = 'flow_progress'
)
SELECT date_trunc('day', timestamp) AS day, e.dataset, e.name,
       sum(e.passed_records) AS passed, sum(e.failed_records) AS failed,
       round(100 * sum(e.failed_records) / nullif(sum(e.passed_records + e.failed_records), 0), 3) AS failed_pct
FROM flows
GROUP BY ALL
ORDER BY day DESC, failed_pct DESC;

The JSON path is the documented structure today; if it changes, inspect one details value and adjust.

#Alerting

Create a Databricks SQL alert on the query above: notify when failed_pct for a rule exceeds its normal level (for example 1%). Pair expect_or_fail with pipeline failure notifications so invariant breaks page someone.

#Design guidance

Uniqueness and cross-table checks are not row-level. Implement them as a small materialized view that aggregates (for example count(*) - count(DISTINCT order_id)) with an expect_or_fail that the value is 0.

#Common problems

#Prevent it next time

• Every silver table has expectations; every expectation has an owner.
• Weekly review of quarantine volumes with source teams.
• Version rule sets in Git alongside the pipeline code.