# Data quality in pipelines: expectations, quarantine tables and quality gates

Write each data quality rule as a SQL condition that is true for a good row, and decide what a row that breaks it does. On Databricks, an expectation in a Lakeflow pipeline either keeps the row and counts it (warn, the default), drops it, or fails the update. To keep the bad rows for someone to fix, split them into a quarantine table with the rules they broke. Write every rule so that a missing value makes it false, not NULL, or the row passes.

- Page: https://tecyfy.com/data-engineering/course/medallion-pipelines/expectations
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 7: Pipelines: medallion with declarative pipelines, Rules, quarantine and a quality gate
- Parts of this module: A first pipeline, bronze to gold (https://tecyfy.com/data-engineering/course/medallion-pipelines); Streaming tables, materialized views and full refresh (https://tecyfy.com/data-engineering/course/medallion-pipelines/refresh); Rules, quarantine and a quality gate (https://tecyfy.com/data-engineering/course/medallion-pipelines/expectations); Lakeflow pipelines on Databricks (https://tecyfy.com/data-engineering/course/medallion-pipelines/on-databricks); Build it: the shop's pipeline, checked (https://tecyfy.com/data-engineering/course/medallion-pipelines/build-it)

At the end of this part you can:

- Split a table's lines into good rows and a quarantine table that says which rules each bad line broke
- Write rules that treat a missing value as a failure
- Choose between warn, drop, fail and quarantine for a rule on Databricks, and stop a run when too many lines fail

[Part 1](/data-engineering/course/medallion-pipelines) built the shop's pipeline, `bronze.py`, `silver.py` and `gold.sql` in a project made by `spark-pipelines init --name shop`, and [part 2](/data-engineering/course/medallion-pipelines/refresh) ran it again on the file of 2026-10-03. To start here, make part 1's project, write its 2 days with `make_files.py`, add part 2's file of 2026-10-03, and run `spark-pipelines run --conf spark.sql.catalogImplementation=hive`, the command for every run below.

## Write the rules once

Part 1's silver kept out 1 kind of bad line, one that did not parse. A line can also parse and still be wrong. Here is the file of 2026-10-04:

```bash
printf 'order_id,customer_id,status,amount,updated_at\n1004,C01,shipped,40.00,2026-10-04 09:00:00\n1006,C03,placed,,2026-10-04 10:00:00\n1007,C01,refunded,9.99,2026-10-04 11:00:00\n1008,C02,placed,-5.00,2026-10-04 12:00:00\n' > landing/orders-2026-10-04.csv
```

Order 1006 has no amount, 1007 a status the shop's rules do not know, and 1008 a negative amount. Rather than a `where` per problem, write the rules down once, each true for a good line, and keep the lines that break any of them in a table of their own, a quarantine. Replace `transformations/silver.py` with:

```python
from pyspark import pipelines as dp
from pyspark.sql import functions as F

# Each rule is true for a good line. A line that breaks any rule goes to rejected_orders.
RULES = {
    "parsed": "_bad IS NULL",
    "known_status": "status IN ('placed', 'paid', 'shipped', 'cancelled')",
    "positive_amount": "amount > 0",
}


@dp.temporary_view
def checked_orders():
    broken = [F.when(~F.expr(rule), name) for name, rule in RULES.items()]
    return (
        spark.read.table("bronze_orders")
        .select("order_id", "customer_id", F.lower(F.trim("status")).alias("status"), "amount", "updated_at", "_bad", "source_file")
        .select("*", F.concat_ws(", ", *broken).alias("broken_rules"))
    )


@dp.materialized_view(comment="1 row per order, in its newest good state")
def orders():
    newest = F.max_by(F.struct("customer_id", "status", "amount", "updated_at"), "updated_at")
    return (
        spark.read.table("checked_orders")
        .where("broken_rules = ''")
        .groupBy("order_id")
        .agg(newest.alias("order"))
        .select("order_id", "order.*")
    )


@dp.materialized_view(comment="Lines that broke a rule, and the rules they broke")
def rejected_orders():
    return (
        spark.read.table("checked_orders")
        .where("broken_rules != ''")
        .select("order_id", "status", "amount", "broken_rules", "source_file")
    )
```

`checked_orders` is a temporary view: it exists only during a run and is never stored, so it costs nothing to share between the 2 tables. It also cleans `status`, which part 2 showed belongs here and not in bronze. `broken_rules` names every rule a line broke, and is empty for a good line.

## The NULL trap

Run the pipeline, then `python show.py orders rejected_orders customer_spend`:

```text
orders
+--------+-----------+---------+------+-------------------+
|order_id|customer_id|status   |amount|updated_at         |
+--------+-----------+---------+------+-------------------+
|1001    |C01        |shipped  |25.50 |2026-10-02 08:00:00|
|1002    |C02        |cancelled|12.00 |2026-10-03 09:30:00|
|1004    |C01        |shipped  |40.00 |2026-10-04 09:00:00|
|1005    |C02        |placed   |18.25 |2026-10-03 16:20:00|
|1006    |C03        |placed   |NULL  |2026-10-04 10:00:00|
+--------+-----------+---------+------+-------------------+

rejected_orders
+--------+--------+------+---------------+---------------------+
|order_id|status  |amount|broken_rules   |source_file          |
+--------+--------+------+---------------+---------------------+
|1003    |placed  |NULL  |parsed         |orders-2026-10-01.csv|
|1007    |refunded|9.99  |known_status   |orders-2026-10-04.csv|
|1008    |placed  |-5.00 |positive_amount|orders-2026-10-04.csv|
+--------+--------+------+---------------+---------------------+

customer_spend
+-----------+------+-----+
|customer_id|orders|spend|
+-----------+------+-----+
|C01        |2     |65.50|
|C02        |1     |18.25|
|C03        |1     |NULL |
+-----------+------+-----+
```

Order 1006, with no amount, is in silver, and C03 now spends `NULL`. For a missing amount, `amount > 0` is not false but NULL, `NOT NULL` is NULL too, and `when` adds a rule name only when its test is true. Order 1003 shows the same trap: only `parsed` is named, not `positive_amount`. Make a rule that comes out NULL count as broken, by changing the line in `checked_orders` to:

```python
    broken = [F.when(~F.coalesce(F.expr(rule), F.lit(False)), name) for name, rule in RULES.items()]
```

Run again, and silver is the 4 good orders, 1001, 1002, 1004 and 1005. `rejected_orders` holds 1003 with `parsed, positive_amount`, 1006 with `positive_amount`, 1007 with `known_status` and 1008 with `positive_amount`, and C03 is gone from gold. Every line is in exactly 1 of the 2 tables, with its reason.

## Expectations on Databricks

On Databricks, Lakeflow pipelines have expectations, rules declared on a table, which the open-source framework does not have. Each one is a name and a SQL condition, and its action decides what a row that breaks it does. Press each action to see what it writes for the file of 2026-10-04:

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/medallion-pipelines/expectations)

The quarantine above is Databricks' own pattern for keeping bad rows: 1 flag computed from all the rules, `NOT` of the rules joined by `AND`, and the rows split by it. Databricks' docs also say a constraint must come out true or false for every row, so write the NULL case into the rule itself: `amount IS NOT NULL AND amount > 0`.

## Stop the run when too much is bad

A quarantine keeps the pipeline running, which is right for a few bad lines. When a third of all lines fail, the file itself is probably broken, and a run should stop where people will see it. Save this as `transformations/quality_gate.sql`:

```sql
-- Fails this flow, and the run, when more than a third of all lines were rejected.
CREATE MATERIALIZED VIEW quality_gate AS
SELECT rejected, lines,
  IF(rejected * 3 > lines, raise_error(rejected || ' of ' || lines || ' lines were rejected'), 'ok') AS result
FROM (
  SELECT (SELECT count(*) FROM rejected_orders) AS rejected,
         (SELECT count(*) FROM bronze_orders) AS lines
);
```

`raise_error` fails the query with your message. 4 of the 11 lines are rejected, so the run fails:

```text
2026-10-18 08:18:19: Flow 'spark_catalog.default.quality_gate' has FAILED.
Error: [USER_RAISED_EXCEPTION] 4 of 11 lines were rejected SQLSTATE: P0001
2026-10-18 08:18:19: Flow spark_catalog.default.customer_spend is PLANNING.
2026-10-18 08:18:19: Flow spark_catalog.default.customer_spend is STARTING.
2026-10-18 08:18:19: Flow spark_catalog.default.customer_spend is RUNNING.
2026-10-18 08:18:20: Flow spark_catalog.default.customer_spend has COMPLETED.
```

The pipeline tried the gate 3 times in all, then `spark-pipelines` ended with exit code 1, which a scheduler reads as a failure. But `customer_spend` was still updated: it does not read `quality_gate`, so nothing waits for it. Databricks' docs say the same of a validation table. To keep gold from updating after a failed check, they advise 2 pipelines in a job, the second running only when the first succeeds, which module 10 builds. The [data quality](https://tecyfy.com/concepts/data-quality) concept has more kinds of rules.

## Check yourself

1. A line has no amount, and the rule positive_amount is amount > 0. The check is F.when(~F.expr(rule), name). Why is the line not rejected?

   A. The CSV reader filled the amount with 0
   B. amount > 0 is NULL for a missing amount, NOT NULL is NULL, and when() treats NULL as not true, so no rule name is added
   C. concat_ws drops the last rule
   D. The line was rejected by the parsed rule instead

2. On Databricks, silver uses @dp.expect_all_or_fail(rules) and a new file has 1 bad line among 1,000 good ones. What lands in silver from this update?

   A. The 999 good lines
   B. All 1,000 lines, with the bad one counted in the metrics
   C. The good lines up to the bad one
   D. Nothing, because the update fails and rolls back, and someone must fix the data or the rule before it can run

### Answers

1. B. In SQL a comparison with NULL is NULL, and so is its negation, so the row lands in neither branch of the test and counts as good. Wrapping the rule in coalesce(rule, false) turns NULL into a failure; on Databricks, write the rule as amount IS NOT NULL AND amount > 0, since a constraint must come out true or false.
2. D. Fail stops the flow on an invalid record and rolls the table's update back atomically, so even the good lines are not written, and no metrics are recorded. Use it for a rule that must never break; use drop or a quarantine when a few bad lines are expected.

## Questions

### What are expectations in Databricks pipelines?

Rules on a pipeline table, each a name and a SQL condition that must be true for a valid row, added with @dp.expect in Python or CONSTRAINT ... EXPECT in SQL. Each has an action for a row that breaks it. Warn keeps the row and counts it, drop leaves it out and counts it, and fail stops the update. The counts appear in the pipeline's Data quality tab.

### What is a quarantine table in a data pipeline?

A table that holds the rows that failed a data quality rule, with the reason, instead of dropping them. The clean table gets only the good rows, nothing is lost, and someone can fix the bad rows at the source and load them again. Databricks' pattern computes 1 flag from all the rules and splits the rows by it.

## Sources

- [Manage data quality with pipeline expectations](https://docs.databricks.com/aws/en/ldp/expectations)
- [Expectation recommendations and advanced patterns](https://docs.databricks.com/aws/en/ldp/expectation-patterns)
- [Spark Declarative Pipelines programming guide](https://spark.apache.org/docs/latest/declarative-pipelines-programming-guide.html)
- [raise_error function](https://docs.databricks.com/aws/en/sql/language-manual/functions/raise_error)