Every engineer who ships application code writes unit tests for it. Yet the data pipeline that feeds the same application is often left with a manual run and a dashboard that someone eventually notices is wrong.
The gap has been more about tooling than intent. Data and machine learning engineers have long argued that the practices that make software dependable, including tests, contracts, and validation, apply just as strongly to the data that feeds it. There has been recent work on data unit tests for pipelines that directly address this.
In data engineering, the most costly failures are rarely catastrophic crashes. A job that stops completely is flagged immediately. More damaging are runs that finish without errors but quietly produce flawed output. The corrupt results flow downstream, populate analytics, and are treated as accurate until someone uncovers a discrepancy weeks later.
With Lakeflow Spark Declarative Pipelines (SDP), you can now write unit tests that run natively inside the pipeline editor, against mock data you define, on the pipeline compute, with nothing to deploy and no separate testing setup to build.
This blog reveals whether you can finally build a data pipeline to the same standard you already hold for the rest of your code in different ecosystems.
In this post, we will walk through building a simple orders pipeline and demonstrate how silent errors can slip past unnoticed during a successful run. You will then see how native unit tests allow you to catch these bugs early and resolve silent failures before they reach production.
Summary
- Native unit tests run inside the Lakeflow editor on pipeline compute, with nothing to deploy.
- Mocking a source is one
CREATE TABLE, and a test is the execution of part of the pipeline, followed by an assert. - You will watch a successful run report wrong numbers, then catch the cause with a single test.
The Story
The example is a simple retail orders pipeline, consisting of four tables, a story most teams recognize. We start with the orders_raw bronze table, which ingests raw orders from a source table called orders_source that sits outside the pipeline. It is converted to a orders_clean table in the silver layer, undergoing transformations that drop rows with no order ID, no customer ID, or both; treat non-positive quantities or null prices as invalid; and computes a line total. There is also a customers_history table built with AUTO CDC as a Slowly Changing Dimension (SCD) Type 2 table, so every change to a customer tier is retained for the window during which it was valid. Finally, we have orders_curated, the gold table that joins orders to the current customer tier and aggregates revenue by tier.
Fig 1. The ETL: Four source tables starting from orders_source and ending with orders_curated
The transformation carries the data-quality contract as expectations, declared in the table's definition. The following blocks explain how it is built using PySpark.
from pyspark import pipelines as dp
from pyspark.sql import functions as F
from pyspark.sql.functions import col, expr
@dp.table
@dp.expect_or_drop("has_order_id", "order_id IS NOT NULL")
@dp.expect_or_drop("has_customer_id", "customer_id IS NOT NULL")
@dp.expect_or_fail("positive_quantity", "quantity > 0")
def orders_clean():
raw = spark.read.table("orders_raw")
return (
raw.withColumn("unit_price", col("unit_price").cast("double"))
.withColumn("quantity", col("quantity").cast("int"))
.withColumn(
"line_total",
F.when(
col("unit_price").isNotNull() & (col("quantity") > 0),
col("quantity") * col("unit_price"),
).otherwise(F.lit(None).cast("double")),
)
.select("order_id", "customer_id", "quantity", "unit_price", "line_total")
)
@DP.view
def customers_cdc():
return spark.readStream.table(f"{CATALOG}.{SCHEMA}.customers_cdf")
dp.create_streaming_table("customers_history")
dp.create_auto_cdc_flow(
target="customers_history",
source="customers_cdc",
keys=["customer_id"],
sequence_by=col("sequence_num"),
apply_as_deletes=expr("operation = 'DELETE'"),
except_column_list=["operation", "sequence_num"],
stored_as_scd_type="2",
)
Expectations enforce the contract at run time. The unit tests assert the transform logic itself. You want both, and the rest of this post shows why.
orders_source and ending with orders_curated
The pipeline passed, but is it production-ready?
If you run the pipeline, it succeeds. But does that mean we can move it to production?
Every table builds, every expectation passes, and the gold table orders_curated reports revenue and order counts by customer tier. The numbers look plausible, so no one looks closer.
Fig 3. Reconciliation of the data that is generated reveals the quality of the code we have written
Add up the order counts across the tiers, and they come to 25. We started with a seeded value of 15 orders. One of them dropped due to a null customer ID, so 14 orders should be attributed. The extra 11 come from a join that matched every historical version for each customer, rather than only the current version. A customer who moved from Silver to Gold has their orders counted under both, and the Silver tier shows 13 orders, while no customer is currently Silver at all.
The pipeline passed without any failures, so nothing raised an alert. If we shipped it to production, dashboard reports could show inflated revenue for weeks before someone traced the issue back to the pipeline. This is exactly the kind of failure a unit test is designed to catch - one that a pipeline run will never flag on its own.
What does a native SDP unit test look like?
There are two kinds of pipeline testing, and it helps to keep them apart:
- On checks the data that goes through the pipeline. On Databricks, expectations check each row when the pipeline builds a table. They can remove a bad row or stop the update. Data Quality Monitoring with Anomaly Detection checks the tables after the pipeline writes to them.
- The other kind checks the logic that produces the data, by running a transform against mock inputs you define, rather than against production data. That second kind is a unit test, and if you have written unit tests elsewhere, it is the same idea: you validate the transformation on inputs you control before the real tables are built.
The offering we are going to explore is Lakeflow Unit Testing, which is in beta on all cloud platforms as of the date of this article's publication.
The framework gives you three pieces, and together they are less code than a docstring 🙂. It starts with test_spark, a session fixture that redirects any table you reference by its fully qualified name to an isolated per-run schema. Hence, a test reads your mock instead of production data. TestPipeline.active() is a handle to the pipeline open in the editor. test_pipeline.run(test_spark, {...}) runs a selective refresh of only the tables you name, synchronously, so the moment it returns, you can read the result and assert on it. The run() call does not raise an error when the update fails. It returns a status object. Check the status.is_success() before you read a table. If you do not, a failed update shows as a "table not found" error, and you do not see the real cause.
The test file lives in a tests folder within the pipeline, but it declares no tables of its own, so it never becomes part of the pipeline's source code or graph. TestPipeline.active() returns the pipeline you have open in the editor, which is how a test knows which pipeline it belongs to. You run the tests from the editor itself, on the pipeline compute, not as a separate job.
The following example seeds a source table with a single CREATE TABLE, refreshes the silver transform, and asserts that the expected rows survive.
from pyspark.pipelines.testing import TestPipeline, test_spark
test_pipeline = TestPipeline.active()
CATALOG = "main"
SCHEMA = "demo_sdp_unit_testing"
FQ = f"{CATALOG}.{SCHEMA}"
CLEAN_CHAIN = {f"{FQ}.orders_raw", f"{FQ}.orders_clean"}
CURATED_CHAIN = {
f"{FQ}.orders_raw", f"{FQ}.orders_clean",
f"{FQ}.customers_history", f"{FQ}.orders_curated",
}
CDC_CHAIN = {f"{FQ}.customers_history"}
def run_chain(session, chain):
status = test_pipeline.run(session, chain)
assert status.is_success, f"{status.error_class}: {status.error_message}"
def mock_orders_source(session):
session.sql(f"""
CREATE OR REPLACE TABLE {FQ}.orders_source AS
SELECT * FROM VALUES
(1001, 'C1', 2, 50.00),
(1003, 'C2', 3, NULL),
(1004, NULL, 1, 10.00)
AS t(order_id, customer_id, quantity, unit_price)
""")
def test_clean_happy_path_row_count(test_spark):
mock_orders_source(test_spark)
run_chain(test_spark, CLEAN_CHAIN)
assert test_spark.table(f"{FQ}.orders_clean").count() == 2
This is the complete process. You simulate a data source, perform a refresh, and verify the results. A test that can be written quickly is more likely to be written and maintained, distinguishing between a pipeline you hope works and one you've actually validated. You can use Genie Code (as your Data Engineer AI co-worker) to write unit tests too!
Find the bug with one test
The test seeds one customer with a closed Silver version and a current Gold version, plus two orders, and then asserts that both orders are attributed solely to the Gold version.
def test_curated_attributes_to_current_tier(test_spark):
mock_curated_sources(test_spark)
run_chain(test_spark, CURATED_CHAIN)
rows = {r["tier"]: r for r in test_spark.table(f"{FQ}.orders_curated").collect()}
assert set(rows) == {"Gold"}
assert rows["Gold"]["order_count"] == 2
assert rows["Gold"]["revenue"] == 130.00
Run the tests, and this one test, test_curated_attributes_to_current_tier fails while the other eight pass. The assertion is specific: the current tier is the only tier that should appear, and the closed Silver version has crept in beside it.
Fig 6. Failing tests can be tracked visually in the navigation pane.
Fig 7. The test failure shows the assertion that failed and the line number.
The fix is a change to a single line of code. Since the orders_curated table reads the SCD type 2 history without filtering out the current rows, you need to add the filter back. In a Type 2 history table, the current version of a row is the one with __END_AT set to null, and older versions have an end timestamp.
# before
current_customers = (
spark.read.table("customers_history")
.select("customer_id", "tier")
)
# after
current_customers = (
spark.read.table("customers_history")
.filter(col("__END_AT").isNull())
.select("customer_id", "tier")
)
When you run the tests again, and all nine pass. Re-run the pipeline, and the order counts come to 14, revenue attributes to the current tier for each customer, and the two orders from the deleted customer sit under a null tier where they belong. The test that found the bug now guards against it returning, so the next person to touch this join finds out in seconds, well before it reaches a dashboard.
Fig 8. Cyclical runs after fixing the issues.
Fig 9. Rerun the reconciliation to check whether the output data is in the correct shape.
How do you prove an expectation fires?
Expectations declare what good data looks like, and a test proves the policy fires. orders_clean drops any row with a null order ID through @DP.expect_or_drop. The test seeds one valid row and one row with a null order ID, refreshes the transform, and asserts a single row survives.
def test_expectation_drops_row_missing_order_id(test_spark):
test_spark.sql(f"""
CREATE OR REPLACE TABLE {FQ}.orders_source AS
SELECT * FROM VALUES
(2001, 'C9', 1, 5.00),
(NULL, 'C9', 1, 5.00)
AS t(order_id, customer_id, quantity, unit_price)
""")
run_chain(test_spark, CLEAN_CHAIN)
assert test_spark.table(f"{FQ}.orders_clean").count() == 1
A dropped row is easy to assert on, because you count the output, and the bad row is gone. Testing the declared policy this way means a later change to the expectation, or to the transform that feeds it, cannot quietly weaken the contract without a test failing.
What can a unit test do that an expectation cannot?
Expectations and unit tests answer different questions. An expectation checks whether a row of real data is valid as it moves through the pipeline. A unit test checks whether the transform computes the right answer for the inputs you choose. You need both, and a unit test can do things that an expectation, by design, cannot.
A unit test gives you control of the input. You build the exact case you want to check: for example, missing data/presence of certain records as data moves downstream, or a change stream that arrives out of order, or an order with a null price. You check it before any of those appear in production. An expectation, on the other hand, only ever sees the data that actually arrives, so a case you have not hit is a case you have not tested. It can also pin the result to an exact value. A unit test can require that a given input produces one Gold row with a count of 2 and revenue of 130.00, and fails the moment the numbers drift. Expectations check each row on its own, so they cannot say what a join or an aggregate should add up to. That is how, in our case, the bug in the gold tables slipped past earlier: every row was valid, every expectation passed, but the total still came out at 25 instead of 14. And it runs before the pipeline sees real data, so it stops broken logic from shipping. An expectation checks each row when the pipeline builds the table. It can remove a bad row or stop the update. It can only do this after the bad row comes into the pipeline. Expectations and unit tests are complementary.
Expectations catch the data you did not see coming, at production scale. Unit tests catch the logic you reasoned through, on the cases you picked, before anything is built.
You can unit test order-dependent logic too: AUTO CDC and SCD Type 2
A plain transform test covers a function of its input. Order-dependent logic needs more, and the framework handles this directly. Consider the customers_history table, which resolves change events into a Type 2 history, and note that the order in which those events arrive determines the result. The test seeds three events for one customer, deliberately out of sequence, and asserts the current row reflects the latest by sequence.
One property of the Beta release is also worth noting. A run performs a selective refresh, and a full refresh is not available, so a test builds exactly the tables it names and then reads the results. Each call to run() does one selective refresh. A test can call run() more than once. For example, refresh the chain, add rows to the source table, and then refresh the pipeline again. Use this method to test incremental behavior, as we do support multiple selective refreshes in the test suite. Keep each test small, so that it is easy to read and completes quickly.
Here is an example of this selective refresh approach.
def test_cdc_scd2_current_state_is_latest_by_sequence(test_spark):
mock_customers_cdf(test_spark)
run_chain(test_spark, CDC_CHAIN)
current = (
test_spark.table(f"{FQ}.customers_history")
.filter("__END_AT IS NULL AND customer_id = 'C1'")
.first()
)
assert current["tier"] == "Gold"
assert current["city"] == "Manchester"
We have two rules that are applied to maintain the checks. You can reference every table by its fully qualified name, catalog, and schema included, because the fixture redirects fully qualified names only. You can also list every table in the chain from the mocked source down to the table under test, because the run does a selective refresh of exactly the tables you name and does not build their upstream tables for you. A view is inlined automatically, so it never appears in the list, which is why the CDC test lists only customers_history , while a test on the gold table lists the whole chain.
If you do not list a table in the chain, the run fails. The run_chain helper displays the table that the run failed to build.
orders_raw It is not in the chain, so the status check shows SOURCE_TABLE_NOT_MATERIALIZED
Summary
Because these tests run alongside the pipeline and use its compute, they fit naturally into the editing process. You can change a function, a rule, run the relevant tests in the editor, and see the results before running the full pipeline, all in one place, without switching context. You can test the entire file at once, and each run rebuilds only the tables you specify. That means a focused test takes seconds instead of waiting for the full pipeline.
A test you write today stops the same defect from returning. If the defect reappears, the suite catches it. Over time, the suite becomes a written record of the assumptions your pipeline relies on, checked on every run rather than left to memory. These tests do not replace running the pipeline on real data. They lock the logic and the declared contracts in place, so full runs focus on volume, freshness, and integration rather than on a transform that was wrong from the start. Testing handles logic correctness, monitoring handles data correctness, and developers who switch windows constantly lose the speed this in-UI experience gives them.
Get started today on the Native Unit testing flow for Lakeflow SDP, all available in the UI!
What is next?
Today, these tests run inside the web-based Lakeflow editor. What comes next is moving that same workflow out to the command line. You already author a pipeline as plain Python or SQL files, keep them in Git, and deploy them with the Databricks CLI and Asset Bundles, so the pipeline you tested here already lives as code you can version and review like any other part of your stack. As SDP development moves to the CLI, you will run these tests from your own IDE, where the loop is fast and local, and trigger them in your CI/CD process, so every change is automatically verified before it is promoted. That closes the distance between a test you run by hand and a gate that runs on its own. Your tests travel with the code, which means a pipeline you can test is a pipeline you can trust, in the editor today and across your own toolchain next.
Getting started
The fastest path is to clone the quickstart repo and run the setup notebook, which seeds the data and creates the pipeline for you. The official documentation covers native unit testing for SDP as the feature moves through Beta.
