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

Part 1 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 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:

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:

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:

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:

    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:

Try itWhat an expectation does with bad lines

On a bad line

WarnDropFailQuarantine

The new file, checked against 2 rules

order_id	status	amount	broken rule
1004	shipped	40.00	none
1007	refunded	9.99	known_status
1008	placed	-5.00	positive_amount

@dp.expect_all(rules)

What this update adds to silver

order_id	status	amount
1004	shipped	40.00
1007	refunded	9.99
1008	placed	-5.00

The update succeeds. Metrics: 2 of 3 records failed, 1 for each rule.Succeeds

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:

-- 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:

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 qualityData QualityHow far data can be trusted: complete, correct, on time and in the expected shape. Pipelines check it with rules and stop or flag rows that fail. concept has more kinds of rules.