How should I structure a medallion pipeline in SDP?

Summary

• In Spark Declarative Pipelines (SDP), map each medallion layer to a dataset type: streaming tables (@dp.table with a streaming read) for bronze and most of silver, and materialized views (@dp.materialized_view) for gold aggregates.
• Keep all three layers in one pipeline project. SDP "analyzes the dependencies of your defined objects and orchestrates their order of execution and parallelization automatically."
• Read upstream layers with spark.readStream.table() or spark.table(), keep dataset functions free of writes, and run the project with the spark-pipelines CLI, which ships with Apache Spark 4.1 and later.

How should I structure a medallion pipeline in SDP?

Put the whole flow in one SDP pipeline project and give each layer the dataset type that matches how its data changes. The medallion pattern uses three layers: bronze (raw), silver (validated) and gold (enriched). SDP offers two main dataset types for them:

• A streaming table is "a definition of a table and one or more streaming flows written into it". It processes "only new data as it arrives".
• A materialized view is "a view that is precomputed into a table". It "always has exactly one batch flow writing to it".

Layer	What it holds	SDP dataset	Python	SQL
Bronze	Raw, unvalidated data, appended incrementally	Streaming table	@dp.table with spark.readStream	CREATE STREAMING TABLE
Silver	Cleaned, typed, validated records	Streaming table, or a materialized view for small batch inputs	@dp.table with spark.readStream.table()	CREATE STREAMING TABLE ... FROM STREAM
Gold	Aggregates for analytics and reporting	Materialized view	@dp.materialized_view with spark.table()	CREATE MATERIALIZED VIEW
Shared steps	Logic that several datasets reuse	Temporary view	@dp.temporary_view	CREATE TEMPORARY VIEW

You reference datasets inside the pipeline "in the same way you'd reference tables defined outside your pipeline". SDP works out the order from those references.

What goes in the bronze layer?

Bronze holds "raw, unvalidated data" that "is appended incrementally and grows over time", so keep transformations to a minimum. This streaming table follows the Kafka example in the SDP programming guide:

from pyspark import pipelines as dp
from pyspark.sql import DataFrame

@dp.table
def bronze_orders() -> DataFrame:
    return (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("subscribe", "orders")
        .load()
    )

The Kafka source returns key and value as binary columns, so leave parsing to silver. The connector is a separate artifact (spark-sql-kafka-0-10_2.13). spark-pipelines accepts spark-submit arguments, so add it with --packages.

What goes in the silver layer?

Silver is where you "perform data cleansing, deduplication, and normalization". When sources are append-only, the medallion guidance says to "configure most reads from bronze as streaming reads", so each run handles only new records:

from pyspark.sql import functions as sf

@dp.table
def silver_orders() -> DataFrame:
    schema = "order_id STRING, customer_id STRING, amount DOUBLE"
    return (
        spark.readStream.table("bronze_orders")
        .select(sf.from_json(sf.col("value").cast("string"), schema).alias("o"))
        .select("o.*")
        .filter(sf.col("order_id").isNotNull())
    )

from_json accepts a DDL-formatted string as its schema. The filter keeps only records that have an order_id.

What goes in the gold layer?

Gold "consists of aggregated data tailored for analytics and reporting". A materialized view reads silver in batch:

@dp.materialized_view
def gold_customer_totals() -> DataFrame:
    return (
        spark.table("silver_orders")
        .groupBy("customer_id")
        .agg(sf.sum("amount").alias("total_amount"))
    )

How do you run and check the pipeline?

A spark-pipeline.yml spec ties the files together. It needs name, libraries (the paths to your SQL or Python files) and storage, "a directory where checkpoints can be stored for streaming tables". spark-pipelines init --name my_pipeline generates a starter project.

• spark-pipelines dry-run catches syntax, analysis and graph validation errors ("e.g. cyclic dependencies") without reading or writing data.
• spark-pipelines run performs "a default incremental update".
• spark-pipelines run --full-refresh silver_orders,gold_customer_totals "clears all existing data and checkpoints for the specified datasets and recomputes them from scratch."

Keep dataset functions pure. "The function used to define a dataset must return a pyspark.sql.DataFrame", and the guide says to never use "methods that save or write to files or tables" there. Its list of operations to avoid includes collect(), toPandas(), saveAsTable() and start(). On Spark 4.1, each file must declare spark = SparkSession.active(). From Spark 4.2, SDP injects spark for you.

How Databricks approaches medallion pipelines

On Databricks, "Lakeflow pipelines extend and are interoperable with Spark Declarative Pipelines". Lakeflow Pipelines adds expectations for silver quality rules: dp.expect (warn) still writes invalid records, dp.expect_or_drop drops them, and dp.expect_or_fail stops the update. Its AUTO CDC APIs compute slowly changing dimensions (SCD) Type 1 and Type 2 from a CDC feed or from snapshots. A pipeline task in Lakeflow Jobs can run the pipeline on a schedule, on a table update, on file arrival or continuously.

FAQs

Can I run SDP without Databricks?

Yes. Apache Spark 4.1.0 added SDP, and you can install it with pip install pyspark[pipelines]. The CLI "is built on top of spark-submit, meaning that it supports all cluster managers supported by spark-submit."

Can I write SCD Type 2 logic with MERGE in a gold dataset?

Not inside a dataset function, because those must return a DataFrame and must not write to tables. Spark 4.2.0 added Auto CDC to SDP for "declarative SCD Type 1 upserts". Databricks AUTO CDC handles Type 2, and Databricks notes that these APIs "are not supported by Apache Spark Declarative Pipelines".

Which table format do SDP datasets use?

@dp.table and @dp.materialized_view take a format argument ("The format of the table, e.g. "parquet""), plus table_properties, partition_cols and cluster_by. Open-source Spark's spark.sql.sources.default is parquet. On Databricks, "Delta Lake is the default format for all operations".