How do I test an SDP pipeline?

Summary

• Run spark-pipelines dry-run first. It launches a run that "doesn't write or read any data" yet catches syntax errors, analysis errors and graph validation errors such as cyclic dependencies.
• Unit-test transformation logic by keeping it in plain functions that take and return DataFrames, then check them with pytest, a local SparkSession and assertDataFrameEqual.
• Integration-test with spark-pipelines run and a separate test spec (its own catalog, database and storage), using --full-refresh-all so each run starts clean, then assert on the output tables.
• On Databricks, the Lakeflow Pipelines Editor has a Dry run button, and pipeline unit tests with mock inputs are available in Beta.

How do I test an SDP pipeline?

Spark Declarative Pipelines (SDP) is the declarative pipeline framework in Apache Spark, added in Spark 4.1.0. The programming guide describes it as a framework for "reliable, maintainable, and testable data pipelines". There's no single test command, so test in layers: validate the graph, test each transformation alone, then run the whole pipeline on small data you control.

Layer	What it catches	Tool
Dry run	Syntax errors, analysis errors such as a missing table or column, graph validation errors such as cyclic dependencies	spark-pipelines dry-run
Unit tests	Wrong logic inside one transformation	pytest, a local SparkSession, assertDataFrameEqual
Integration run	Wiring between datasets, refresh behavior, streaming checkpoints	spark-pipelines run with a test spec
Output checks	Row counts, null keys and value ranges in the produced tables	SQL or DataFrame assertions after the run

What does spark-pipelines dry-run check?

spark-pipelines dry-run "launches an execution of a pipeline that doesn't write or read any data, but catches many kinds of errors that would be caught if the pipeline were to actually run." The guide lists three kinds: syntax errors (invalid Python or SQL), analysis errors (selecting from a table or column that doesn't exist) and graph validation errors (cyclic dependencies). It takes --spec plus the spark-submit arguments except --class. Because it reads and writes nothing, it suits a pull-request check, but it can't tell you whether your logic is right or your data is clean.

pip install "pyspark[pipelines]"
spark-pipelines init --name my_pipeline
cd my_pipeline
spark-pipelines dry-run

How do I unit test a transformation?

The guide says Python functions that define datasets "should include only the code required to define the table or view", and each one must return a pyspark.sql.DataFrame. So put the DataFrame logic in a plain function, call it from the decorated dataset function, and test the plain function. The decorated function then reduces to return valid_orders(spark.table("orders")). Make the logic module importable wherever the pipeline runs, for example by installing it as a package.
Don't import a file that applies the @dp decorators from a test. The decorators register datasets with the running pipeline, and outside one the Spark 4.2.0 source raises GRAPH_ELEMENT_DEFINED_OUTSIDE_OF_DECLARATIVE_PIPELINE. The PySpark testing guide says its testing utilities "are standalone, meaning they can be compatible with any test framework or CI test pipeline".

# order_logic.py (no pipeline decorators)
from pyspark.sql import DataFrame
from pyspark.sql.functions import col

def valid_orders(orders: DataFrame) -> DataFrame:
    return orders.filter(col("amount") > 0)

# test_orders.py
import pytest
from pyspark.sql import SparkSession
from pyspark.testing.utils import assertDataFrameEqual
from order_logic import valid_orders

@pytest.fixture
def spark_fixture():
    spark = SparkSession.builder.master("local[1]").appName("sdp-tests").getOrCreate()
    yield spark

def test_valid_orders_drops_non_positive_amounts(spark_fixture):
    source = spark_fixture.createDataFrame([(1, 100.0), (2, -50.0)], ["order_id", "amount"])
    expected = spark_fixture.createDataFrame([(1, 100.0)], ["order_id", "amount"])
    assertDataFrameEqual(valid_orders(source), expected)

How do I run an integration test?

Run the real pipeline on a small, known dataset and inspect what it writes. Create a second spec file in the project root, next to the real one, and name it something else: without --spec, the CLI auto-discovers only spark-pipeline.yml and spark-pipeline.yaml. Point catalog, database and storage (where streaming checkpoints live) at throwaway locations so test output never lands in real tables.

# spark-pipeline-test.yml
name: my_pipeline_test
libraries:
  - glob:
      include: transformations/**
storage: file:///tmp/my_pipeline_test/storage
catalog: test_catalog
database: test_db

spark-pipelines run --spec spark-pipeline-test.yml --full-refresh-all

--full-refresh-all performs "a full graph reset and recompute", so every test run starts from scratch. Without refresh options, run performs a default incremental update. To test incremental behavior, add rows to a source and run again without refresh options. Afterwards, read the output tables from a session that uses the same catalog and check them with assertDataFrameEqual or count, null and range queries.

What can I test on Databricks?

On Databricks, Lakeflow Pipelines "extend and are interoperable with Spark Declarative Pipelines". The Lakeflow Pipelines Editor has a Dry run button that validates the pipeline "without updating any data". From the CLI, databricks pipelines dry-run validates the pipeline's graph and doesn't materialize or publish any datasets, and databricks pipelines start-update accepts --validate-only to do the same for source code.
Pipeline unit tests are in Beta. You write Python tests in the editor and run them against mock inputs in a temporary test schema, using TestPipeline and test_spark. They need triggered mode and Databricks Runtime 18.1 or above. Isolation covers only operations that reference a table by name, so path-based reads and connectors such as Kafka or Auto Loader reach real data.
Expectations (EXPECT in SQL, @dp.expect in Python) apply data quality checks to each record as a Databricks pipeline runs. The Spark 4.2.0 SDP guide doesn't describe them.

FAQs

Does a passing dry run mean my pipeline works?

No. A dry run "doesn't write or read any data", so it can't see bad values, empty sources or wrong logic. Pair it with unit tests and an integration run.

Where do data assertions go?

After the run, in the test. The guide lists collect(), count(), pivot() and toPandas() among the operations that should never be used in SDP dataset code, so keep checks out of the dataset definitions.