Introduction
In Part 1 of this series, Integration Testing for LakeFlow Jobs with Pytest and Databricks Connect, we built a blueprint for testing LakeFlow Jobs: deploy the job, trigger it from pytest, and assert on the results with Databricks Connect, all from your local pytest environment.
Testing jobs are straightforward. You write functions, you call them, and you check what comes back. Spark Declarative Pipelines (SDP) change the model. You don't call anything. You declare the datasets you want, and the pipeline engine works out the execution graph, the streaming state, and how each table gets materialized. This is great for production workload, but how do you write tests for a table you never explicitly built?
This post will deep dive into how to test your SDP pipelines. We'll split the problem the way any solid testing strategy should: fast, isolated unit tests for your transformation logic, and integration tests that run the real pipeline end to end. We'll use a small NYC Taxi pipeline as the running example. By the end, you'll have a blueprint for moving an SDP pipeline from experiment to production with tests you can trust.
Unit Testing SDP Pipelines in Python
Unit tests have one job: to prove your transformation logic is correct. They should run locally, in isolation, and fast without requiring any additional services to integrate with. For an SDP pipeline, that logic is the code that transforms your data.
The catch is that SDP wraps that logic inside dataset definitions that only run within the pipeline engine. So the first move is structuring your code so there's something to point your tests to.
Decouple the transformation from the pipeline definition
An SDP dataset looks like an ordinary Python function, so it's tempting to test it like one. Let's take a simple example:
# ./nyctaxi_pipeline.py
from pyspark.sql import functions as F
@dp.table(comment="Raw NYC Taxi trips as streaming table")
def nyctaxi_trips_raw():
return spark.readStream.table("samples.nyctaxi.trips")
@dp.materialized_view(comment="Average trip distance by pickup zip")
def avg_distance():
return (
spark.read.table("nyctaxi_trips_raw")
.groupBy("pickup_zip")
.agg(F.avg("trip_distance").alias("avg"))
)
That function only means something inside a running SDP pipeline. The @DP.table and @DP.materialized_view decorators register these datasets with the pipeline engine, and spark.read.table("nyctaxi_trips_raw") points at a dataset that exists only once the pipeline runs. Call it from a unit test and there's nothing for it to run against
The fix is a design decision. Keep your pipeline definitions thin, and push the transformation logic into separate functions that take a DataFrame and return a DataFrame:
# ./transformations.py
def calculate_avg_distance(df: DataFrame) -> DataFrame:
return (
df.groupBy("pickup_zip")
.agg(F.avg("trip_distance").alias("avg"))
)
# ./nyctaxi_pipeline.py
@dp.table(comment="Raw NYC Taxi trips as streaming table")
def nyctaxi_trips_raw():
return spark.readStream.table("samples.nyctaxi.trips")
@dp.materialized_view(comment="Average trip distance by pickup zip")
def avg_distance():
return calculate_avg_distance(spark.read.table("nyctaxi_trips_raw"))
Now the pipeline definition is a thin wrapper that wires a source to a transformation. The transformation knows nothing about SDP, streaming, or catalogs, it's just a function over a DataFrame.
Test the transformation with pytest
Because the logic is decoupled, the test is ordinary pytest testing. Start a local Spark session, feed it a small sample DataFrame, call the function, and assert on the output:
@pytest.mark.unit_test
def test_calculate_avg_distance(nyctaxi_data_df):
avg = calculate_avg_distance(nyctaxi_data_df).first()["avg"]
assert round(avg) == 1
The test runs against a local Spark, so you get instant feedback around your test results.
Tools and environment setup
We walked through the project setup and uv dependency-group setup in more depth in Part 1. The same project layout carries straight over to SDP pipelines.
Unit tests should run on your laptop with nothing but Python. To pull that off, your unit tests should run against the open-source PySpark instead of Databricks Connect.
Databricks Connect replaces the Pyspark package. You can't have both installed at once as installing one will break the other. The integration tests will use Databricks Connect to run against Databricks compute, but for unit tests we'll use plain OSS PySpark.
UV dependency groups solve this cleanly. You declare a group per testing layer and select the one you need at run time. Let's look at an example of how to set this up:
# pyproject.toml
[dependency-groups]
dev = ["pytest>=8.3.4", "databricks-labs-pytester"]
unit-tests = ["pyspark>=4.0.0,<5.0.0"] # local, open-source Spark
integration-tests = ["databricks-connect==17.1.0"] # remote Databricks compute
[tool.uv]
default-groups = ["dev"]
So the unit-test run pulls in PySpark and any other dependency specific to unit testing:
uv run --group unit-tests pytest -m unit_test
Know what you can't unit test (and why)
Decoupling buys you fast, isolated testing for your logic, but it leaves some things out. A unit test runs plain Python and PySpark on your machine, so anything that lives only inside the pipeline runtime or the Databricks platform is out of reach:
- Change Data Capture: (APPLY_CHANGES) CDC processing is applied by the pipeline engine as it processes a stream. There's no engine in a local PySpark session, so there's nothing to apply the changes.
- Expectations: data-quality constraints are declared on a dataset and enforced by the pipeline as it runs. A direct function call never travels through that enforcement path.
- Auto Loader: incremental file ingestion needs real cloud storage and a streaming source behind it. There's nothing meaningful to stand in for that in a unit test.
- Unity Catalog: catalog and schema resolution, permissions, and lineage are workspace concerns. A local session has no catalog to resolve names against.
These features behave correctly only when the whole pipeline runs against the platform. To cover them, you have to run the actual pipeline. That's integration testing, and it's where we go next.
Integration Testing SDP Pipelines in Python or SQL
Unit tests prove your transformation logic works in isolation, but an SDP pipeline is more than transformations. The engine handles streaming state, dependency resolution, view materialization, expectations enforcement, Auto Loader coordination, and Unity Catalog assets access and management. None of which can be done currently in a local PySpark session. Integration testing is where you execute the pipeline as a whole: deploy it, run it, and assert that the outputs are correct in an actual Databricks environment using the SDP engine.
The challenge with SDP is that you can't just "call" the pipeline from test code the way you'd call a function. The pipeline runs inside the engine, so your test has to orchestrate from the outside: create or reference a pipeline, trigger an update, wait for completion, and then inspect the results.
There are two ways to approach this:
- Validate inside the pipeline itself: Use SDP expectations to embed logic and data-quality checks directly into your dataset definitions. The pipeline engine evaluates them on every run and fails the update if they're violated.
- Run it locally with pytest: Create an isolated test pipeline programmatically, trigger it via the Databricks SDK, wait for it to complete, and assert on the output tables using Databricks Connect. This gives you the full flexibility of pytest, such as, parameterization, fixtures and CI integration.
Both approaches have a place. Expectations catch data-quality regressions continuously in acceptance or production environments. Pytest-driven tests give you a controlled, repeatable integration test you can run before merging code. Let's look at each.
Approach 1: Validate data quality inside the pipeline with expectations
SDP has a built-in mechanism for asserting on data as it flows through your pipeline called expectations. You declare a constraint on a dataset, and the engine evaluates it on a row-by-row basis during each update. If the constraint is violated, the pipeline can log a warning, drop the offending rows, or fail the update entirely.
The integration testing pattern is to add a dedicated validation dataset to the end of your pipeline that reads from an output table and applies expectations to verify the results. This dataset isn't part of the pipeline business logic, it only exists to assert the pipeline's output:
from pyspark.sql import functions as F
@dp.expect_or_fail("valid_avg", "avg IS NOT NULL AND avg > 0")
@dp.temporary_view(comment="Validate avg_distance values are correct")
def validate_avg_distance():
return spark.read.table("avg_distance")
@dp.expect_or_fail("has_rows", "row_count > 0")
@dp.temporary_view(comment="Validate avg_distance has rows")
def validate_avg_distance_count():
return (
spark.read.table("avg_distance")
.select(F.count("*").alias("row_count"))
)
This creates dedicated validation steps in the pipeline graph. When the pipeline runs, it first materializes avg_distance, then executes the validation tables/views which read the output and apply expectations against it. If any expectation fails, the pipeline update fails. Because these are temporary views, they don't persist as physical tables, they only exist during the pipeline update for validation purposes.
You can expand this pattern with multiple validation datasets, each targeting a different output table or testing a different check. The pipeline engine handles the dependency ordering automatically.
For a deeper dive on using expectations as part of a DevOps testing strategy, including shared expectation libraries and CI integration, see Applying Software Development & DevOps Best Practices to Delta Live Table Pipelines.
Pros:
- Zero external tooling. The validation rule lives alongside the pipeline definition.
- Runs on every update automatically, catching regressions in acceptance and production
- Works for both Python and SQL pipelines
- Quality metrics are visible in the pipeline UI
Cons:
- Expectations evaluate SQL predicates row by row. Table-wide assertions (e.g., "row count should be greater than 0") require aggregating into a separate validation dataset first, adding complexity to the pipeline graph.
- Tightly coupled to the pipeline definition: Changing a constraint means redeploying the pipeline
- Additionally, if the validation should only be run on a specific environment, this requires a separate pipeline definition per environment with different parameters/configurations.
- No test isolation: A single pipeline is run for both the data processing and validation. Cannot run multiple isolated tests at the same time that test different scenarios.
- No local feedback loop: Requires you to deploy and run the pipeline to see results
Expectations are a strong first line of defense. But when you need full control over the test environment, isolated schemas, arbitrary assertions, parameterized runs and CI integration, you need to orchestrate from the outside. That's Approach 2.
Approach 2: Integration testing locally with pytest
Approach 1 stays inside the Databricks workspace. This approach brings the test back to where you already work: your IDE (or a Databricks test notebook), with pytest and a debugger. In Part 1 we tested a Lakeflow Job exactly this way, using pytest, Databricks Connect, and databricks-labs-pytester, and the same toolchain carries over to SDP pipelines. The new problem to solve is state. A pipeline's outputs (streaming tables and materialized views) persist across runs, so leftover checkpoints and table metadata from a previous run can poison the next one, and your test result stops reflecting the code you are actually testing.
Getting that isolation right is the core of this approach. You stay in your IDE, you keep pytest, and you let the pipeline run directly on Databricks.
How the approach works
Since a pipeline only runs inside the Lakeflow runtime, you don't run it locally. Instead, pytest acts as the orchestrator and the assertion engine, while the pipeline itself runs on Databricks serverless compute:
- pytest reads the spec of your already-deployed pipeline using Databricks SDK.
- pytester creates an ephemeral Unity Catalog schema to isolate the run
- pytester creates a fresh, ephemeral pipeline from that spec, pointed at the ephemeral schema.
- pytest then triggers a full-refresh update through the Databricks SDK and waits for it to finish.
- pytest reads the output table using Databricks Connect and asserts the data.
- When the test finishes, pytester tears down both ephemeral resources: it deletes the pipeline and drops the schema.
The division of labor is the whole trick:
- pytest is the test runner with breakpoints and a normal debugger.
- databricks-labs-pytester supplies the test fixtures, creates the ephemeral schema and pipeline, and guarantees cleanup when done.
- Databricks SDK reads and triggers the full-refresh pipeline update.
- Databricks Connect reads Unity Catalog tables back for assertions.
Pros and cons
Let us be clear about the trade-offs before you adopt this.
What you get:
- It runs the same way locally and in CI, with no extra test job to deploy.
- The full pytest ecosystem: fixtures, parametrization, markers, and reporting you already use.
- Real runtime parity. The pipeline runs on the Databricks runtime, so you can test the SDP features local Spark can't: Auto Loader, Unity Catalog reads/writes, AutoCDC (APPLY CHANGES), and SDP expectations.
- A tight feedback loop. You write, run, and debug the tests in one place, no context switching between the IDE and the workspace UI.
What it costs:
- The pipeline still runs on Databricks, so each test run costs compute and takes time to complete. Serverless instant compute keep that round trip short, with no cluster to wait on or keep warm.
- You need Unity Catalog privileges to create schemas in the target catalog.
- A full refresh recomputes everything. That's the point of a clean test, but it isn't free.
Use this for end-to-end confidence. Keep using unit tests for fast, local checks of your transformation logic.
Project structure
The pipeline, its deployment, and its tests live in one repo. Here is an example project structure for an SDP pipeline:
workflow-test-automation-blueprint/
├── databricks.yml # DAB bundle definition
├── pyproject.toml # uv project + dependency groups
├── conftest.py # pytest CLI options (--pipeline-name)
├── resources/
│ └── nyctaxi_sdp_pipeline.pipeline.yml # SDP pipeline resource (DAB)
├── src/ps_test_blueprint/
│ ├── nyctaxi_pipeline.py # the pipeline definition (flow)
│ ├── nyctaxi_functions.py # transformation logic (reused by unit tests)
│ └── utils.py # read_table helper (batch vs streaming)
└── tests/
└── integration/
└── test_nyctaxi_sdp_pipeline.py # the integration test
The pipeline is small on purpose: a streaming table and a materialized view. The transformation is decoupled from the flow, so the actual logic lives in nyctaxi_functions.py (and can be unit-tested on its own), while the pipeline file just wires sources to transforms:
# src/ps_test_blueprint/nyctaxi_pipeline.py
from pyspark import pipelines as dp
from ps_test_blueprint.nyctaxi_functions import calculate_avg_distance
from ps_test_blueprint.utils import read_table
@dp.table(comment="Raw NYC Taxi trips as streaming table")
def nyctaxi_trips_raw():
return read_table("samples.nyctaxi.trips", incremental=True)
@dp.materialized_view(comment="Average trip distance by pickup zip")
def avg_distance():
df = read_table("nyctaxi_trips_raw")
return calculate_avg_distance(df)
We deploy it with a Declarative Automation Bundle (DAB). serverless: true lets the test request serverless compute later, and root_path points the pipeline at the package source so its imports resolve:
# resources/nyctaxi_sdp_pipeline.pipeline.yml
resources:
pipelines:
nyctaxi_sdp_pipeline:
name: nyctaxi_sdp_pipeline
catalog: main
schema: default
serverless: true
root_path: ../src
libraries:
- file:
path: ../src/ps_test_blueprint/nyctaxi_pipeline.py
Dependency isolation is handled with uv groups. Databricks Connect replaces pyspark, so the two can't coexist; keeping them in separate groups stops them from overwriting each other:
# pyproject.toml
[dependency-groups]
dev = [
"pytest>=8.3.4",
"databricks-labs-pytester"
]
unit-tests = [
"pyspark>=4.0.0,<5.0.0"
]
integration-tests = [
"databricks-connect==17.1.0"
]
Environment setup
You need the Databricks CLI installed and authenticated to your dev workspace, plus uv. Then install the integration dependencies and point Databricks Connect at serverless compute:
# Install the integration test dependencies
uv sync --only-group integration-tests
# Tell Databricks Connect which workspace and compute to use
export DATABRICKS_HOST=<your-dev-workspace-url>
export DATABRICKS_SERVERLESS_COMPUTE_ID=auto
DATABRICKS_SERVERLESS_COMPUTE_ID=auto is what gives you on-demand serverless compute for the reads, so you don't keep a cluster warm just to run tests. If you'd rather use a classic cluster, set DATABRICKS_CLUSTER_ID instead.
Deploy the pipeline so there's a spec for the test to read:
databricks bundle deploy -t dev
In development mode the bundle prefixes the resource name, so the deployed pipeline is named like [dev your_name] nyctaxi_sdp_pipeline. You'll pass that exact name to the test.
The test, step by step
Here's the full integration test. We'll walk through it.
# tests/integration/test_nyctaxi_sdp_pipeline.py
import pytest
from collections import namedtuple
from datetime import timedelta
try:
from databricks.connect import DatabricksSession as SparkSession
except ImportError:
from pyspark.sql import SparkSession as SparkSession
CreatedPipeline = namedtuple("CreatedPipeline", ["pipeline_id", "catalog", "schema"])
@pytest.fixture
def deployed_pipeline_spec(ws, request):
"""Look up the deployed pipeline by name and return its spec.
The test creates a fresh, isolated pipeline from this spec so that each
run starts with clean streaming state -- no stale checkpoints or table
metadata from previous ephemeral schemas.
"""
pipeline_name = request.config.getoption("--pipeline-name")
deployed = next(
(p for p in ws.pipelines.list_pipelines() if p.name == pipeline_name),
None,
)
if deployed is None:
raise ValueError(f"Pipeline '{pipeline_name}' not found.")
return ws.pipelines.get(deployed.pipeline_id).spec
@pytest.fixture
def test_pipeline(make_schema, make_pipeline, deployed_pipeline_spec):
"""Create a fresh, isolated pipeline mirroring the deployed one.
Inherits the deployed pipeline's full configuration and overrides
only what must be unique to this run.
"""
catalog_name = "main"
schema_name = make_schema(catalog_name=catalog_name).name
test_pipeline_spec = deployed_pipeline_spec.as_shallow_dict()
test_pipeline_spec.pop("id", None)
test_pipeline_spec.pop("deployment", None)
test_pipeline_spec["clusters"] = [] # serverless: stop make_pipeline injecting a classic cluster
test_pipeline_spec["name"] = f"test_{deployed_pipeline_spec.name}_{schema_name}"
test_pipeline_spec["catalog"] = catalog_name
test_pipeline_spec["schema"] = schema_name
created = make_pipeline(**test_pipeline_spec)
return CreatedPipeline(
pipeline_id=created.pipeline_id,
catalog=catalog_name,
schema=schema_name,
)
@pytest.mark.integration_test
def test_sdp_pipeline(spark, ws, test_pipeline):
ws.pipelines.start_update(pipeline_id=test_pipeline.pipeline_id, full_refresh=True)
result = ws.pipelines.wait_get_pipeline_idle(
pipeline_id=test_pipeline.pipeline_id,
timeout=timedelta(minutes=15),
)
assert result.state.name == "IDLE", f"Pipeline ended in state {result.state}"
latest_update = ws.pipelines.list_updates(test_pipeline.pipeline_id).updates[0]
assert latest_update.state.value == "COMPLETED", (
f"Pipeline update {latest_update.state.value}: check pipeline events"
)
trips_df = spark.read.table(f"{test_pipeline.catalog}.{test_pipeline.schema}.nyctaxi_trips_raw")
assert trips_df.count() > 0
avg_df = spark.read.table(f"{test_pipeline.catalog}.{test_pipeline.schema}.avg_distance")
assert avg_df.count() > 0
Read the deployed spec, don't reuse the pipeline. The deployed_pipeline_spec fixture finds the deployed pipeline by name and returns its spec: its libraries, its root_path, and whether it's serverless. We reuse the spec, not the pipeline.
Get an isolated schema. Inside the test_pipeline fixture, make_schema (from pytester) creates a fresh Unity Catalog schema for this run and registers it for automatic cleanup. Every run gets its own namespace, so parallel and repeated runs don't collide.
Create a fresh pipeline for this run. The test_pipeline fixture builds a brand-new pipeline from the deployed spec using pytester's make_pipeline, which creates the pipeline and registers it for automatic cleanup. It calls as_shallow_dict() to copy the spec, drops the fields that must be unique (id, deployment), and overrides name, catalog, and schema to point at the ephemeral schema. Copying the whole spec means everything else, libraries, root_path, serverless, comes along for free, with no list of fields to keep in sync by hand. Creating a fresh pipeline per run is the most important design choice here (more on why below).
One non-obvious detail: make_pipeline is built to stand up a pipeline from scratch, so when you don't pass clusters it injects a default classic single-node cluster. Passing an empty clusters list suppresses that default and lets the serverless setting from the deployed spec win.
Trigger a full refresh and wait. start_update(full_refresh=True) clears all streaming checkpoints and table data, so the run starts from a known state. wait_get_pipeline_idle() blocks until the pipeline stops moving.
Verify it is actually completed. Reaching IDLE only means the pipeline stopped, not that it succeeded. We pull the latest update and assert its state is COMPLETED before trusting any data.
Assert on the data through Databricks Connect. spark.read.table(...) reads the streaming table and the materialized view from the ephemeral UC schema and checks both have rows.
Cleanup is automatic. Because the pipeline came from make_pipeline and the schema from make_schema, pytester tears both down at the end of the run, even if an assertion fails. There's no try/finally to maintain, and the test body stays focused on the run and the assertions.
Running it
Deploy the pipeline, then run the test, passing the deployed pipeline name:
databricks bundle deploy -t dev
uv run python -m pytest -rsx -v -m integration_test \
--pipeline-name="[dev your_name] nyctaxi_sdp_pipeline" \
tests/integration/test_nyctaxi_sdp_pipeline.py
Here's a run against a serverless dev workspace:
============================= test session starts ==============================
platform darwin -- Python 3.14.0, pytest-9.0.2, pluggy-1.6.0
cachedir: .pytest_cache
rootdir: /.../workflow-test-automation-blueprint
configfile: pytest.ini
plugins: databricks-labs-pytester-0.7.4
collecting ... collected 1 item
tests/integration/test_nyctaxi_sdp_pipeline.py::test_sdp_pipeline PASSED [100%]
======================== 1 passed in 72.26s (0:01:12) =========================
While the test runs, you can watch it work in the workspace. A new pipeline named test_<deployed-name>_<schema> appears, runs a full refresh on serverless, and then disappears when the test finishes.
The ephemeral test pipeline mid-run. The unique pipeline name (top) and the isolated dummy_* schema both tables write to (bottom) are created by the test and deleted afterward.
The pipeline and the schema are both gone after the run. Run it as many times as you like, each run is independent. Because every run gets its own uniquely named pipeline and schema, you can also run tests in parallel without them stepping on each other.
Conclusion & Next Steps
Testing an SDP pipeline comes down to splitting the problem in two:
- Unit tests prove your transformation logic in isolation: decouple the logic from the dataset definitions, hand it a small DataFrame on a local Spark, and assert on what comes back in seconds.
- Integration tests prove the pipeline actually runs: deploy it, trigger a full refresh against a fresh, ephemeral schema, and assert on the real streaming tables and materialized views through Databricks Connect, all without leaving your pytest environment.
Together, they give you a fast inner loop and end-to-end confidence, and they run the same way locally and in CI.
Everything in this series lives in a companion repo you can clone and run: spark-declarative-pipeline-testing-blueprint. It has the NYC Taxi pipeline, the unit tests, the integration test and its fixtures, the DAB bundle, and the uv setup, so you can go from git clone to a passing test against your own workspace.
If you're just getting started, read Part 1 for the Lakeflow Jobs version of this blueprint, clone the repo, and point the fixtures at one of your own pipelines. The patterns here, ephemeral schemas, a fresh pipeline per run, and asserting on COMPLETED before reading data, carry over to any SDP pipeline you own.
One more thing worth watching. The techniques in this post work today, on any workspace, by orchestrating the pipeline from the outside. Databricks is also building a unit testing library directly into Spark Declarative Pipelines. It's currently in preview and heading into beta, and it gives you an isolated SparkSession (test_spark) that redirects every read and write to a temporary schema, a TestPipeline.active() handle on the current pipeline, and a test_pipeline.run(...) call that can refresh a single table or the whole DAG, all asserted with ordinary pytest.
It runs in the Lakeflow editor today, with local IDE and CI/CD support on the roadmap. When that lands, expect much of the scaffolding in this post, standing up an ephemeral pipeline, wiring schemas by hand, to collapse into a few framework primitives. The testing mindset stays exactly the same: decouple your logic, isolate your state, and assert on real output.
