Lakeflow Declarative Pipelines Expectations Design Guide: Quality Monitoring & Invalid Data Handling

Expectations in Lakeflow Declarative Pipelines (formerly Delta Live Tables / DLT) are a declarative way to validate and control data quality inside a pipeline. You embed rules like "this column must not be NULL" or "amount must be non-negative" directly in the table definition, and choose whether violating records are logged, dropped, or whether the entire pipeline should fail.

This article walks through the 3 operating modes of Expectations, both SQL and PySpark syntax, the quarantine pattern, quality monitoring via the event log, and how Expectations compare to Delta Table Constraints (NOT NULL / CHECK).

The 3 Operating Modes of Expectations

Expectations come in 3 modes that differ in how violating records are handled. You pick a mode based on the nature of the pipeline (batch vs. streaming) and business requirements (how much data loss is acceptable).

EXPECT mode is the right choice when you do not want to stop the pipeline on violations but still want to track their count. A typical use case is initial Bronze-layer ingestion, where you monitor violation trends via the event log.

SQL Syntax: Defining Expectations with CONSTRAINT

To define Expectations in SQL, you use the CONSTRAINT clause inside a CREATE OR REFRESH statement. Quality rules are declared inline as part of the table definition.

-- EXPECT: 違反を記録するが全レコードを出力に含める
CREATE OR REFRESH MATERIALIZED VIEW silver_orders
(
  CONSTRAINT valid_amount EXPECT (amount > 0),
  CONSTRAINT not_null_id EXPECT (order_id IS NOT NULL)
)
AS SELECT * FROM LIVE.bronze_orders;

-- EXPECT OR DROP: 違反レコードを出力から除外
CREATE OR REFRESH MATERIALIZED VIEW silver_orders_clean
(
  CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW,
  CONSTRAINT not_null_id EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW
)
AS SELECT * FROM LIVE.bronze_orders;

-- EXPECT OR FAIL: 1件でも違反があればパイプラインを停止
CREATE OR REFRESH MATERIALIZED VIEW gold_revenue
(
  CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION FAIL UPDATE
)
AS SELECT region, SUM(amount) AS total FROM LIVE.silver_orders_clean
GROUP BY region;

CONSTRAINT names must be unique within the pipeline. Since the event log uses these names as the key for tracking violation counts, business-meaningful names like valid_amount or not_null_customer_id are recommended.

PySpark Syntax: The @dlt.expect Decorator

In PySpark, Expectations are defined via decorators. In practice, the @dlt.expect_all family is most commonly used because it accepts multiple rules as a dictionary.

import dlt
from pyspark.sql.functions import col

# 単一条件の EXPECT（記録のみ）
@dlt.table
@dlt.expect("valid_amount", "amount > 0")
@dlt.expect("not_null_id", "order_id IS NOT NULL")
def silver_orders():
    return dlt.read_stream("bronze_orders")

# 複数条件をまとめて DROP
quality_rules = {
    "valid_amount": "amount > 0",
    "not_null_id": "order_id IS NOT NULL",
    "valid_region": "region IN ('APAC','EMEA','NA')"
}

@dlt.table
@dlt.expect_all_or_drop(quality_rules)
def silver_orders_clean():
    return dlt.read_stream("bronze_orders")

# FAIL: 1件でも違反でパイプライン停止
@dlt.table
@dlt.expect_all_or_fail({"positive_revenue": "total > 0"})
def gold_revenue():
    return (
        dlt.read("silver_orders_clean")
        .groupBy("region")
        .agg({"amount": "sum"})
        .withColumnRenamed("sum(amount)", "total")
    )

@dlt.expect_all takes a dictionary and applies every rule in EXPECT (log) mode. @dlt.expect_all_or_drop and @dlt.expect_all_or_fail correspond to DROP and FAIL modes respectively. Extracting the dictionary into an external variable makes rule management and test-time substitution much easier.

Quarantine Pattern: Isolating Violating Data for Later Processing

EXPECT OR DROP makes records disappear, and EXPECT lets bad data flow downstream. When you want to keep violating data but prevent it from reaching downstream tables, use the quarantine table pattern.

import dlt
from pyspark.sql.functions import expr

@dlt.table(comment="品質条件を満たすレコードのみ")
def silver_orders_valid():
    return (
        dlt.read_stream("bronze_orders")
        .filter("amount > 0 AND order_id IS NOT NULL")
    )

@dlt.table(comment="品質違反レコードを隔離")
def silver_orders_quarantine():
    return (
        dlt.read_stream("bronze_orders")
        .filter("amount <= 0 OR order_id IS NULL")
        .withColumn("quarantine_reason",
            expr("""CASE
                WHEN amount <= 0 THEN 'invalid_amount'
                WHEN order_id IS NULL THEN 'null_order_id'
                ELSE 'unknown'
            END"""))
    )

Adding a violation-reason column to the quarantine table makes it useful for data quality reporting, root-cause analysis by ops teams, and reinjection workflows after the data is fixed.

Reviewing Quality Metrics in the Event Log

Expectations violation counts are recorded automatically in the pipeline event log. You can see them on the pipeline detail page in the UI, or query the event_log table directly with SQL.

SELECT
  details:flow_progress.data_quality.expectations.name AS expectation_name,
  details:flow_progress.data_quality.expectations.dataset AS dataset,
  details:flow_progress.data_quality.expectations.passed_records AS passed,
  details:flow_progress.data_quality.expectations.failed_records AS failed,
  timestamp
FROM event_log(TABLE(my_catalog.my_schema.my_pipeline))
WHERE event_type = 'flow_progress'
  AND details:flow_progress.data_quality IS NOT NULL
ORDER BY timestamp DESC;

This query returns the pass and fail counts for each Expectation over time. Combine it with alerting on sudden spikes in violation rates to catch upstream data-quality regressions early.

Choosing Between Expectations and Delta Constraints (NOT NULL / CHECK)

DLT Expectations and Delta Table Constraints both relate to data quality, but they operate at different layers and behave differently.

In practice, the robust pattern is to use DLT Expectations to check quality progressively through the pipeline, then place Delta Constraints on the final Gold table so that only data that has cleared every quality check can be written.

Design Guidelines: Placement in the Medallion Architecture

• Bronze: use EXPECT (log only) to visualize raw-data quality trends. Do not drop or fail.
• Silver: use EXPECT OR DROP to filter out bad records. Route to a quarantine table when retention is needed.
• Gold: use EXPECT OR FAIL to guarantee zero quality violations. Apply to billing and aggregation tables.

This staged quality gating minimizes the risk of upstream quality issues propagating downstream, while preventing data loss at the Bronze layer.

What Gets Tested on the Exam

• Can you precisely distinguish the violation behavior of the 3 modes (EXPECT / EXPECT OR DROP / EXPECT OR FAIL)?
• What happens to records dropped by EXPECT OR DROP (they are not auto-retained — quarantine must be implemented yourself)
• How to retrieve pass and fail counts for Expectations from the event log
• The difference in where Delta Constraints apply (in-pipeline vs. at table write time)

Check Your Understanding

Frequently Asked Questions