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

The module so far built the shop's pipeline on your computer: a bronze streaming table in part 1, full refreshes in part 2, and rules, a quarantine and a gate in part 3. 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:

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

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

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:

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:

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

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:

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:

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

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:

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.

Comparisondbt vs Spark Declarative Pipelinesdbt 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…