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

A medallion pipeline keeps data in 3 layers: bronze holds the records as they arrived, silver the cleaned and checked records, and gold the tables people query, such as spend per customer. With Spark Declarative Pipelines you write each table as a query, a streaming table for bronze and materialized views for silver and gold. spark-pipelines run reads which table each query reads, and updates them in that order.

- Page: https://tecyfy.com/data-engineering/course/medallion-pipelines
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 7: Pipelines: medallion with declarative pipelines, A first pipeline, bronze to gold
- 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:

- Say what bronze, silver and gold each hold, and who reads each
- Write a pipeline of a streaming table and 2 materialized views, in Python and SQL, and run it with spark-pipelines
- Read a run's log to see the order the pipeline chose

[Module 6](/data-engineering/course/getting-data-in) 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](/data-engineering/course/getting-data-in), the script that writes the order files for 2026-10-01 and 2026-10-02.

## Bronze, silver and gold

The [medallion architecture](https://tecyfy.com/concepts/medallion-architecture) 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:

```bash
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.

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

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

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

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

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

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

```text
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](/data-engineering/course/medallion-pipelines/refresh) runs the pipeline again as new files arrive, and changes a query.

## Check yourself

1. The run log imports silver.py first, then gold.sql, then bronze.py. In what order are the 3 tables updated?

   A. silver, gold, bronze, the order the files were imported
   B. bronze_orders, then orders, then customer_spend, because each query names the table it reads
   C. All 3 at once, since they are in separate files
   D. In alphabetical order of the table names

2. The file of 2026-10-02 has a second line for order 1001, now shipped. Why does customer_spend count 1001 once?

   A. The stream skipped the older line
   B. Gold removes duplicate lines with DISTINCT
   C. bronze_orders keeps only the newest line of each order
   D. orders keeps 1 row per order, its newest line by updated_at, and gold reads orders

### Answers

1. B. The files only declare tables. The pipeline builds a graph from the tables each query reads, so orders waits for bronze_orders and customer_spend waits for orders, whatever order the files are in. Tables that do not depend on each other can run at the same time.
2. D. Bronze keeps both lines of 1001, as they arrived. Silver groups by order_id and max_by picks the line with the latest updated_at, so gold counts orders, not lines. Keeping every line in bronze is what lets silver be rebuilt with a different rule later.

## Questions

### What is the medallion architecture?

A way to arrange a lakehouse's tables in 3 layers of quality. Bronze holds raw data as it was loaded, with no cleanup. Silver holds data that has been cleaned and validated, such as 1 row per order with bad lines removed. Gold holds the tables built for the people who use them, such as totals per customer for a dashboard. Databricks recommends it but says it is not a requirement.

### What is Spark Declarative Pipelines?

A framework in Apache Spark for building pipelines by declaring the tables you want rather than the steps to make them. You write each table as a query in Python or SQL; the pipeline works out the order from the queries, runs what can run in parallel and keeps the checkpoints of its streams. On Databricks, Lakeflow pipelines extend it with more features, such as AUTO CDC.

## Sources

- [What is the medallion lakehouse architecture?](https://docs.databricks.com/aws/en/lakehouse/medallion)
- [Spark Declarative Pipelines programming guide](https://spark.apache.org/docs/latest/declarative-pipelines-programming-guide.html)
- [Pipeline concepts](https://docs.databricks.com/aws/en/ldp/concepts)
- [Spark Declarative Pipelines on Lakeflow](https://docs.databricks.com/aws/en/ldp/)