How do I add data-quality expectations to a pipeline?

Summary

• In open-source Apache Spark, the Spark Declarative Pipelines guide (Spark 4.2.0) does not document expectations, so you write the checks as explicit DataFrame logic that splits valid rows from invalid ones.
• In Lakeflow Pipelines, an expectation is an optional clause on a streaming table, materialized view or view. It checks each record against a SQL boolean constraint and applies an action.
• The three actions are warn (the default, dp.expect), drop (dp.expect_or_drop, ON VIOLATION DROP ROW) and fail (dp.expect_or_fail, ON VIOLATION FAIL UPDATE). Quarantine is a pattern, not a fourth action.
• Metrics for warn and drop appear in the pipeline UI Data quality tab and in the event log. A fail action records no metrics.

How do I add data-quality expectations to a pipeline?
In open-source Apache Spark, write the checks as DataFrame logic that routes valid and invalid rows, as shown below. On Databricks, Lakeflow Pipelines have built-in expectations: "optional clauses in pipeline materialized view, streaming table, or view creation statements that apply data quality checks on each record passing through a query."

How do I add expectations in open-source Spark?

Spark Declarative Pipelines (SDP) is the open-source pipeline framework (from pyspark import pipelines as dp). Its Spark 4.2.0 programming guide documents @dp.materialized_view, @dp.table, @dp.temporary_view and @dp.append_flow, and does not mention expectations or constraints. So express the rule as a SQL predicate and split the rows yourself:

rule = "amount > 0 AND user_id IS NOT NULL"
df = spark.read.table("raw_orders")

valid = df.filter(f"coalesce({rule}, false)")
invalid = df.filter(f"NOT coalesce({rule}, false)")

valid.write.mode("append").saveAsTable("orders")
invalid.write.mode("append").saveAsTable("orders_quarantine")

The coalesce matters. In Spark SQL a comparison with NULL returns NULL, NOT NULL is NULL, and a filter keeps only rows where the condition is true. Without coalesce, a row with a NULL amount and a present user_id evaluates to NULL and lands in neither output. To mimic the fail action, check invalid.count() and raise an error before writing.

How do I add expectations in Lakeflow Pipelines?

An expectation has a name, a constraint and an action. The constraint is a SQL condition that must evaluate to true or false for each record. In Python:

from pyspark import pipelines as dp

@dp.table
@dp.expect("valid_timestamp", "timestamp > '2012-01-01'")
@dp.expect_or_drop("valid_current_page", "current_page_id IS NOT NULL AND current_page_title IS NOT NULL")
@dp.expect_or_fail("valid_count", "count > 0")
def clean_clicks():
    return spark.readStream.table("raw_clicks")

The same expectations in SQL:

CREATE OR REFRESH STREAMING TABLE clean_clicks (
  CONSTRAINT valid_timestamp EXPECT (timestamp > '2012-01-01'),
  CONSTRAINT valid_current_page EXPECT (current_page_id IS NOT NULL AND current_page_title IS NOT NULL) ON VIOLATION DROP ROW,
  CONSTRAINT valid_count EXPECT (count > 0) ON VIOLATION FAIL UPDATE
)
AS SELECT * FROM STREAM(raw_clicks);

Action	SQL	Python	Result
warn (default)	EXPECT	dp.expect	Invalid records are written to the target
drop	EXPECT ... ON VIOLATION DROP ROW	dp.expect_or_drop	Invalid records are dropped before data is written to the target
fail	EXPECT ... ON VIOLATION FAIL UPDATE	dp.expect_or_fail	Invalid records prevent the update from succeeding

To reuse one set of rules across datasets, pass a dictionary of names and constraints to the expect_all family:

valid_pages = {
    "valid_count": "count > 0",
    "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL",
}

@dp.table
@dp.expect_all_or_drop(valid_pages)
def prepared_clicks():
    return spark.readStream.table("raw_clicks")

Databricks recommends storing expectation definitions separately from pipeline logic. Loading expectations dynamically from a file is not supported in SQL.

What happens when an expectation fails?

For a table update, a fail action stops the update and "the system atomically rolls back the transaction". In a triggered pipeline, the failure of one flow does not cause other parallel flows to fail. In a continuous pipeline, it stops the flow and all dependent flows. Lakeflow Pipelines report the violation with a dedicated error message that names the violated expectations and, for many queries, the input record behind it. Fix the pipeline code to handle the invalid data before you re-run.
Expectations enforce data quality, not orchestration. A validation table does not make downstream tables wait, so a failed validation does not stop them from updating. To stop downstream processing, split the validation and the downstream work into separate pipelines and orchestrate them with a job.

How do I quarantine invalid records?

Quarantine is a pattern, not an action. The Databricks docs combine expectations with temporary tables and views. A temporary table applies @dp.expect_all(rules) and adds an is_quarantined column. Two views then filter on is_quarantined=false for valid rows and is_quarantined=true for invalid rows.

Where do I see the results?

For warn and drop, open the dataset in the pipeline UI and select the Data quality tab. A fail action records no metrics. The event log keeps passed and failed counts in details:flow_progress.data_quality.expectations and the dropped count in details:flow_progress.data_quality, on events of type flow_progress. Query it with SELECT * FROM event_log(<pipelineId>).

What are the limits?

• Constraints cannot contain custom Python functions, external service calls or subqueries referencing other tables.
• An expectation name must be unique for a dataset.
• Only streaming tables, materialized views and temporary views support expectations.
• Only Python can group expectations, with expect_all, expect_all_or_drop and expect_all_or_fail, which take a dictionary of names and constraints.
• Expectations are not supported with AUTO CDC FROM SNAPSHOT.

FAQs

Can I put several expectations on one dataset?

Yes. In SQL, separate the CONSTRAINT clauses with commas. In Python, stack the decorators or pass a dictionary to the expect_all family. Each name must be unique for the dataset.

What about table-level freshness and completeness?

Expectations check records as they flow through a query. For tables in Unity Catalog, Databricks documents data quality monitoring, whose anomaly detection analyzes "freshness" and "completeness" of enabled tables.

Should I use warn, drop or fail?

Use warn to keep invalid records while collecting metrics, drop to keep them out of the target, and fail when invalid records are unacceptable.