# Build a medallion pipeline with Spark Declarative Pipelines, a quality gate and a check

A pipeline is trustworthy when bad lines are kept aside with a reason, a run stops when too much is bad, and something outside the pipeline checks the result. shop_pipeline does all 3: silver splits lines into orders and rejected_orders by named rules, a quality_gate table fails the run when more than a third of the lines are rejected, and check.py proves that every line reached bronze, every order is in silver or the quarantine once, and gold's totals match silver.

- Page: https://tecyfy.com/data-engineering/course/medallion-pipelines/build-it
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 7: Pipelines: medallion with declarative pipelines, Build it: the shop's pipeline, checked
- 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:

- Build a pipeline of bronze, silver and gold tables with a quarantine and a quality gate
- Decide what to change when a quality gate stops a run, and apply a rule change to old data without a full refresh
- Check after every run that each layer agrees with the files and with the layers around it

The module so far built the shop's pipeline on your computer: a bronze streaming table in [part 1](/data-engineering/course/medallion-pipelines), full refreshes in [part 2](/data-engineering/course/medallion-pipelines/refresh), and rules, a quarantine and a gate in [part 3](/data-engineering/course/medallion-pipelines/expectations). This part puts them together as the ingestion and modelling steps of the shop's final project, adds a table of sales per day, and checks the result. It needs part 1's `bronze.py` and `show.py`, and part 3's `quality_gate.sql`.

## Build it: the shop's pipeline, with a gate and a check

Make the project and give it a script that writes 1 day of order files at a time, all 4 days of this module:

```bash
spark-pipelines init --name shop_pipeline
cd shop_pipeline
rm transformations/example_*
```

```python
"""Write the shop's order file for 1 day into landing/, as the shop's system would drop it."""
import sys
from pathlib import Path

DAYS = {
    "2026-10-01": [
        "1001,C01,placed,25.50,2026-10-01 09:15:00",
        "1002,C02,placed,12.00,2026-10-01 10:02:00",
        "1003,C03,placed,abc,2026-10-01 11:30:00",
    ],
    "2026-10-02": [
        "1001,C01,shipped,25.50,2026-10-02 08:00:00",
        "1004,C01,placed,40.00,2026-10-02 12:45:00",
    ],
    "2026-10-03": [
        "1002,C02,cancelled,12.00,2026-10-03 09:30:00",
        "1005,C02,placed,18.25,2026-10-03 16:20:00",
    ],
    "2026-10-04": [
        "1004,C01,shipped,40.00,2026-10-04 09:00:00",
        "1006,C03,placed,,2026-10-04 10:00:00",
        "1007,C01,refunded,9.99,2026-10-04 11:00:00",
        "1008,C02,placed,-5.00,2026-10-04 12:00:00",
    ],
}
day = sys.argv[1]
landing = Path("landing")
landing.mkdir(exist_ok=True)
lines = ["order_id,customer_id,status,amount,updated_at", *DAYS[day]]
(landing / f"orders-{day}.csv").write_text("\n".join(lines) + "\n")
print("wrote", landing / f"orders-{day}.csv")
```

Save that as `make_files.py`, copy `show.py` beside it, and put `bronze.py` and `quality_gate.sql` in `transformations/`. Silver is part 3's, with the NULL fix, and with 1 more column, `placed_at`, the first time an order was seen. Save it as `transformations/silver.py`:

```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():
    # A rule that comes out NULL, such as amount > 0 on a missing amount, counts as broken.
    broken = [F.when(~F.coalesce(F.expr(rule), F.lit(False)), 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: its newest good state and when it was placed")
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"), F.min("updated_at").alias("placed_at"))
        .select("order_id", "order.*", "placed_at")
    )


@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")
    )
```

Gold has 2 tables in 1 file, `transformations/gold.sql`, spend per customer and sales per day:

```sql
CREATE MATERIALIZED VIEW customer_spend AS
SELECT customer_id, count(*) AS orders, sum(amount) AS spend
FROM orders
WHERE status != 'cancelled'
GROUP BY customer_id;

CREATE MATERIALIZED VIEW daily_sales AS
SELECT to_date(placed_at) AS day, count(*) AS orders, sum(amount) AS revenue
FROM orders
WHERE status != 'cancelled'
GROUP BY to_date(placed_at);
```

## A check from outside the pipeline

The gate stops a run on too many bad lines, but nothing yet says the tables are right. Save this as `check.py`, beside `make_files.py`:

```python
"""Check the pipeline's tables against the files in landing/ and against each other."""
import sys
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("check").enableHiveSupport().getOrCreate()
problems = []

lines = spark.read.text("landing").filter(~F.col("value").startswith("order_id,")).count()
bronze = spark.table("bronze_orders").count()
if lines != bronze:
    problems.append(f"landing/ has {lines} lines, bronze_orders {bronze} rows")

orders = spark.table("orders")
if orders.count() != orders.select("order_id").distinct().count():
    problems.append("orders has an order_id more than once")
seen = spark.table("bronze_orders").select("order_id")
kept = orders.select("order_id").union(spark.table("rejected_orders").select("order_id"))
lost = seen.subtract(kept).count()
if lost:
    problems.append(f"{lost} orders are in neither orders nor rejected_orders")

sold = orders.filter("status NOT IN ('cancelled', 'refunded')").agg(F.sum("amount")).first()[0]
spend = spark.table("customer_spend").agg(F.sum("spend")).first()[0]
revenue = spark.table("daily_sales").agg(F.sum("revenue")).first()[0]
if not sold == spend == revenue:
    problems.append(f"silver sells {sold}, customer_spend {spend}, daily_sales {revenue}")

print(f"bronze {bronze} rows, silver {orders.count()} orders, {sold} sold")
for problem in problems:
    print("PROBLEM:", problem)
sys.exit(1 if problems else 0)
```

It counts the files' lines again, as module 6's check did, then asks 3 questions a pipeline cannot answer about itself. Is every order in silver once? Is every order that bronze saw in silver or the quarantine? Do the 2 gold tables add up to what silver sold? It is a script, not a table in the pipeline, because Databricks' docs say a validation table does not stop the tables downstream of it, and its exit code can stop whatever runs next. Its count of sales already leaves out refunds, ahead of gold.

## Run 4 days

For each of the first 3 days, write the file, run, and check:

```bash
python make_files.py 2026-10-01
spark-pipelines run --conf spark.sql.catalogImplementation=hive
python check.py
```

After the third day, `check.py` prints `bronze 7 rows, silver 4 orders, 83.75 sold` and exits with 0, and `python show.py daily_sales` gives 25.50, 40.00 and 18.25 for the 3 days. Now write `2026-10-04` and run. The gate stops the run, and `spark-pipelines` exits with 1:

```text
2026-10-18 08:24:18: Flow 'spark_catalog.default.quality_gate' has FAILED.
Error: [USER_RAISED_EXCEPTION] 4 of 11 lines were rejected SQLSTATE: P0001
```

`python show.py rejected_orders` says why:

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

1006 and 1008 are bad data, for the shop's system to fix. 1007 is not: the shop has started refunding orders, and `refunded` is a real status the rule does not know. The rule is wrong, so change it, and change what gold counts as a sale. Add `'refunded'` to `known_status` in `silver.py`, change `customer_spend`'s filter to `WHERE status NOT IN ('cancelled', 'refunded')`, and run again. The run completes, but the check does not pass:

```text
bronze 11 rows, silver 5 orders, 83.75 sold
PROBLEM: silver sells 83.75, customer_spend 83.75, daily_sales 93.74
```

`daily_sales` still has the old filter and counts the refund's 9.99 as revenue. Change its filter too, run, and check:

```text
bronze 11 rows, silver 5 orders, 83.75 sold
```

The check exits with 0. 1007 is in silver without a full refresh: silver is a materialized view, recomputed from every bronze line on each run, so a rule change there reaches old lines at once. `quality_gate` now holds 3 of 11 lines, under a third, and `ok`.

On Databricks, the same project is part 4's pipeline with 2 more files, `daily_sales` in gold and the gate. Module 10 runs the pipeline and then the check as 2 tasks of a job, so a failed check stops what follows.

- Comparison: [dbt vs Spark Declarative Pipelines](https://tecyfy.com/data-engineering/dbt-vs-spark-declarative-pipelines). dbt and Spark Declarative Pipelines both let you say what each table should contain and work out the order to build them. dbt compiles SQL models with Jinja and…

## Check yourself

1. You add 'refunded' to known_status in silver.py and run the pipeline again, with no new file. Does order 1007, loaded the day before, now reach orders?

   A. No, because bronze has already read its file; a full refresh of bronze is needed
   B. Yes, because orders is a materialized view, computed again from every bronze line on each run
   C. No, because rejected_orders keeps it until it is deleted
   D. Only after a full refresh of orders

2. After a fix, check.py prints "PROBLEM silver sells 83.75, customer_spend 83.75, daily_sales 93.74". What is the most likely cause?

   A. Bronze loaded a file twice
   B. silver has an order twice
   C. The quality gate let a bad line through
   D. daily_sales still counts order 1007's refund, 9.99, because its filter was not changed with customer_spend's

### Answers

1. B. Bronze still holds 1007's line, and orders and rejected_orders are recomputed from all of bronze on every run, so a rule change in silver applies to old lines at once. Only a change to the streaming table, bronze, would need a full refresh, as part 2 showed.
2. D. 93.74 is 83.75 plus 9.99, the refunded order. The other checks passed, so bronze and silver are sound, and the 2 gold tables disagree only in what they leave out. Every place that decides what counts as a sale must change together, and comparing the totals is what shows when one did not.

## Questions

### How do I test that a data pipeline's output is correct?

Check properties that must always hold, after every run, from outside the pipeline. Every source line is in the raw table once. Every key is in the clean table or the quarantine, and once. Totals computed from 2 tables that should agree, agree. Run the check as the next step after the pipeline, and stop what follows when it fails, the way a failing test stops a deploy.

### What should happen when a data quality gate fails?

The run stops where people see it, and someone reads why before anything is changed. The quarantine says which lines broke which rule. Either the data is wrong, and is fixed at its source, or the rule is wrong, as when the source starts sending a new valid value, and the rule changes. Loosening the gate to make the run pass hides the problem.

## Sources

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