Spark Declarative Pipelines tutorial: a medallion pipeline with bronze, silver and gold

Module 6 loaded the shop's order files into a bronze table with a stream and a checkpoint, and cleaned nothing. This module builds the rest of the pipeline, silver and gold, with Spark Declarative Pipelines. This part needs only make_files.py from module 6, part 1, the script that writes the order files for 2026-10-01 and 2026-10-02.

Bronze, silver and gold

The medallion architectureMedallion ArchitectureOrganising a lakehouse in 3 layers of rising quality: bronze for raw data, silver for cleaned and joined data, and gold for tables shaped for reports and AI. arranges a lakehouse's tables in 3 layers, each better than the one before. In Databricks' words they are raw, validated and enriched:

• Bronze holds the data as it arrived, with no cleanup: every line of every order file, even the line with abc for an amount. Data engineers read it, to rebuild the layers above or to find out what came in.
• Silver holds data that is cleaned and checked: 1 row per order, in its newest state, with bad lines kept out. Analysts and data scientists read it.
• Gold holds tables built for the people who use them, such as spend per customer for a dashboard.

Each layer is made from the one below, so a pipeline is a chain of tables. Module 6's job made the first link by hand. Writing every link by hand means also deciding when each one runs, which is what a declarative pipeline does for you.

A pipeline is a set of table definitions

In Spark Declarative Pipelines you write each table as a query and leave the steps to the pipeline. A streaming table reads each new record once, for data that only grows, such as files landing. A materialized view is a query whose result is stored and recomputed when the pipeline runs, so it always matches what it reads. The pipeline reads every query, sees which table each one reads, and runs them in that order.

Install it beside PySpark, and make a pipeline project:

pip install "pyspark[pipelines]==4.2.0"
spark-pipelines init --name shop
cd shop
rm transformations/example_*

init writes spark-pipeline.yml, the pipeline's spec: its name, the libraries of code to load, here every file in transformations/, and storage, the absolute address of a folder where the pipeline keeps the checkpoints of its streams. Copy make_files.py into shop/ and run it for both days.

Bronze, silver and gold as code

Save this as transformations/bronze.py. It is module 6's read with no write: the pipeline writes the table and keeps the checkpoint.

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

ORDERS = "order_id BIGINT, customer_id STRING, status STRING, amount DECIMAL(10,2), updated_at TIMESTAMP"


@dp.table(comment="Every line of every order file, once, as it arrived")
def bronze_orders():
    return (
        spark.readStream.format("csv")
        .option("header", True)
        .option("enforceSchema", False)
        .option("columnNameOfCorruptRecord", "_bad")
        .schema(f"{ORDERS}, _bad STRING")
        .load("landing")
        .select("*", F.col("_metadata.file_name").alias("source_file"), F.current_timestamp().alias("loaded_at"))
    )

@dp.table on a function that returns a streaming read makes a streaming table named after the function. Save this as transformations/silver.py:

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


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

max_by takes, for each order, the line with the latest updated_at. Gold can be SQL in the same pipeline. Save this as transformations/gold.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;

A function that defines a table must only return its query. The pipeline runs this code more than once as it plans and runs, so count(), collect() or a write inside it would run at the wrong time.

Run it

spark-pipelines run --conf spark.sql.catalogImplementation=hive

The setting keeps the list of tables in a metastore_db folder in the project. Without it, Spark keeps that list in memory and forgets it when the run ends, so the next run finds the table files and stops with LOCATION_ALREADY_EXISTS. On Databricks, Unity Catalog keeps the list. Among Spark's own messages, the run prints its progress:

2026-10-18 08:14:20: Flow spark_catalog.default.bronze_orders is QUEUED.
2026-10-18 08:14:20: Flow spark_catalog.default.orders is QUEUED.
2026-10-18 08:14:20: Flow spark_catalog.default.customer_spend is QUEUED.
2026-10-18 08:14:20: Flow spark_catalog.default.bronze_orders is STARTING.
2026-10-18 08:14:20: Flow spark_catalog.default.bronze_orders is RUNNING.
2026-10-18 08:14:22: Flow spark_catalog.default.bronze_orders has COMPLETED.
2026-10-18 08:14:22: Flow spark_catalog.default.orders is PLANNING.
2026-10-18 08:14:22: Flow spark_catalog.default.orders is STARTING.
2026-10-18 08:14:22: Flow spark_catalog.default.orders is RUNNING.
2026-10-18 08:14:23: Flow spark_catalog.default.orders has COMPLETED.
2026-10-18 08:14:23: Flow spark_catalog.default.customer_spend is PLANNING.
2026-10-18 08:14:23: Flow spark_catalog.default.customer_spend is STARTING.
2026-10-18 08:14:23: Flow spark_catalog.default.customer_spend is RUNNING.
2026-10-18 08:14:23: Flow spark_catalog.default.customer_spend has COMPLETED.
2026-10-18 08:14:25: Run is COMPLETED.

A flow is the query that fills a table, and each table is named in full, catalog and schema first, here Spark's own spark_catalog.default. A few lines earlier, the log says it imported silver.py, then gold.sql, then bronze.py, yet bronze ran first: the order comes from the queries, not the files. To look at the tables, save this as show.py:

"""Print the tables the pipeline wrote, by name."""
import sys
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("show").enableHiveSupport().getOrCreate()
for name in sys.argv[1:]:
    print(name)
    table = spark.table(name).drop("loaded_at")
    table.orderBy(*table.columns).show(truncate=False)

Run python show.py 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        |placed |12.00 |2026-10-01 10:02:00|
|1004    |C01        |placed |40.00 |2026-10-02 12:45:00|
+--------+-----------+-------+------+-------------------+

customer_spend
+-----------+------+-----+
|customer_id|orders|spend|
+-----------+------+-----+
|C01        |2     |65.50|
|C02        |1     |12.00|
+-----------+------+-----+

Bronze holds all 5 lines, both of 1001's and order 1003 with its text in _bad. Silver has 1001 once, shipped, and leaves 1003 out. Here the tables are Parquet files in spark-warehouse/; on Databricks every pipeline table is a Delta table. Part 2 runs the pipeline again as new files arrive, and changes a query.