This route decides what a green pytest run is evidence of. Three things make that answer worth something: a session whose defaults match production, test data that shows only what the test is about, and an honest account of where local Spark and Databricks Spark disagree. The third one is the reason this route exists — the disagreement is not always an error, and when it is not, it is a different number.
Stand 29.07.2026
The harness
Everything that decides whether a green suite is evidence lives in one file and five settings. Get them wrong and the suite is a ritual that runs in eleven seconds and proves nothing.
The whole harness is one file, and it decides exactly two things: which Spark the tests run against, and which defaults are pinned. SPARK_TEST_MODE is the switch — local for the pull-request suite, connect for the nightly parity run. No test knows which one it got.
pythontests/conftest.py — the entire harness
import os
import pytest
MODE = os.environ.get("SPARK_TEST_MODE", "local")
@pytest.fixture(scope="session")
def spark():
"""Session-scoped: the JVM starts once per worker, not once per test."""
if MODE == "connect":
from databricks.connect import DatabricksSession
session = DatabricksSession.builder.serverless(True).getOrCreate()
elif MODE == "local":
from delta import configure_spark_with_delta_pip
from pyspark.sql import SparkSession
builder = (
SparkSession.builder.master("local[2]")
.appName("dwh-tests")
# pinned because production pins them
.config("spark.sql.ansi.enabled", "true")
.config("spark.sql.session.timeZone", "UTC")
# pinned because 200 shuffle tasks over six rows is absurd
.config("spark.sql.shuffle.partitions", "1")
# STATIC configs — applied only when the session is created
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config(
"spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.delta.catalog.DeltaCatalog",
)
.config("spark.ui.enabled", "false")
)
# puts the io.delta JARs matching the installed delta-spark wheel on the
# classpath; the two configs above name classes that are absent without it
session = configure_spark_with_delta_pip(builder).getOrCreate()
else:
raise ValueError(f"SPARK_TEST_MODE must be local or connect, got {MODE!r}")
yield session
if MODE == "local":
session.stop()
@pytest.fixture(scope="session")
def rates(spark):
"""silver.fx_rate, seeded once: EUR at 1, CHF at 0.94 on the factory's date."""
from tests.factories import FX_RATE_SCHEMA, fx_rate_rows
return spark.createDataFrame(fx_rate_rows(), FX_RATE_SCHEMA)
@pytest.fixture(autouse=True)
def _isolate(spark):
"""One session for the whole run means one test can poison the next."""
yield
for table in spark.catalog.listTables():
if table.isTemporary:
spark.catalog.dropTempView(table.name)
spark.catalog.clearCache()
The default path needs no workspace, no token and no grant. A new joiner clones the repo, runs pytest, and has a green suite before anyone has provisioned them anything — the one dependency is a first run that resolves the Delta JARs, which the CI image should have already cached.
Why here
Engineers run this suite forty times a day. A harness that takes ninety seconds to reach the first assertion gets run once, just before pushing — at which point it is not a test suite, it is a slow linter.
The sharper reason: a session built by a bare SparkSession.builder.getOrCreate() is not the engine this warehouse runs on. Every setting nobody pinned is a place where the test and production disagree, and the disagreement surfaces at 03:00 in silver.sales_order_line rather than in a pull request.
Doing it
Keep conftest.py to one screen — under eighty lines. A harness longer than that has bugs of its own and no tests of its own.
Pin every config that can change a result. Leave performance tuning out, with the single exception of shuffle partitions.
Pin delta-spark to the Delta version shipped with the runtime production uses, not to whatever pip resolves today.
Fail loudly on an unknown SPARK_TEST_MODE. A typo that silently falls back to local is a parity run that stops running and keeps reporting green.
Embedding it
Time a cold clone-to-green run with a new joiner watching. That number is the harness's real quality score, and the team owns it once they have seen it.
When someone proposes a helper in conftest.py, ask which test it makes readable. Harness code grows until it needs its own test suite and nobody notices the moment it did.
Name an owner for conftest.py at handover. Shared ownership of the harness means nobody upgrades PySpark.
Defaults
Session-scoped Spark, function-scoped cleanup, nothing in between.
SPARK_TEST_MODE defaults to local; pull-request CI never talks to a workspace and needs no credentials.
Every config the warehouse depends on is explicit, never inherited from a library default.
Gotchas
getOrCreate returns an EXISTING session if one is already up in the process, and static configs are not applied to it. Runtime SQL settings are, so ANSI and the timezone look fine while the Delta extension is silently missing — and the symptom is a parse or analysis error on MERGE INTO, which reads like a syntax bug. It happens when any imported module builds a session at import time.
Setting spark.sql.extensions without putting the Delta JARs on the classpath. The two config lines name Java classes; pip-installing delta-spark ships only the Python side, so the session build dies with ClassNotFoundException on DeltaSparkSessionExtension. configure_spark_with_delta_pip resolves the matching JARs — but it resolves them from Maven, so on an offline or egress-restricted CI runner the first session hangs and then fails inside Ivy with a resolution timeout that never mentions Delta. Bake the JARs into the image or pre-populate the Ivy cache.
Local PySpark on Windows needs winutils.exe, hadoop.dll and HADOOP_HOME before it can write a Delta table. Without them the failure is NativeIO$Windows.access0 deep in a stack trace. Standardise on a devcontainer or WSL and say so in the README; a JAVA_HOME mismatch has the same shape, green on every machine but one.
Session-scoped Spark without the cleanup fixture: a temp view created in one test is visible to the next. The suite passes, then fails when someone adds a test above it, and the bisect blames an innocent commit.
depends on your pyspark pin — off on 3.5, on from 4.0
on: serverless compute and SQL warehouses run ANSI
a bad cast returns NULL in the test and raises in production, or the reverse
spark.sql.session.timeZone
the JVM zone — Europe/Zurich on a laptop, UTC in the container
UTC
to_date(order_ts) lands on a different day; date_sk differs by one between laptop and CI
spark.sql.shuffle.partitions
200
200, with AQE coalescing at real volumes
199 empty tasks per shuffle; a six-second suite becomes six minutes
ANSI is the one that changes numbers rather than timings. The webshop sends quantities as strings, and German decimal commas arrive in them:
pythonThe same cast, two behaviours
from pyspark.sql.functions import col
bad = spark.createDataFrame([("2,5",)], "quantity STRING")
bad.select(col("quantity").cast("decimal(12,3)")).collect()
# ansi.enabled = false -> [Row(quantity=None)]
# ansi.enabled = true -> [CAST_INVALID_INPUT] The value '2,5' of the type
# "STRING" cannot be cast to "DECIMAL(12,3)"
The NULL is the dangerous outcome, not the exception. It propagates: quantity NULL makes amount_eur NULL, the line contributes nothing to revenue, and no job fails. Business rule 5 — amount_eur must be greater than 0 for non-cancelled lines — is what eventually catches it, one layer and several hours downstream.
textpyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
# src layout: dwh.* and tests.* both import without an editable install
pythonpath = ["src", "."]
addopts = "-n 4 --dist loadfile -q"
markers = ["classic_only: known Spark Classic behaviour; expires 2026-10-01"]
The other two levers are structural. Session scope on the spark fixture is worth roughly five seconds per test, because that is what starting a JVM costs: three hundred tests function-scoped spend twenty-five minutes doing nothing but starting Spark, and session-scoped they pay it once. And --dist loadfile sends every test in a file to the same xdist worker, so module-scoped fixtures are built once and a file that creates a temp view cannot collide with another worker doing the same. The default, --dist load, scatters tests individually and gives you both problems.
Why here
These settings are the difference between a suite that models production and a suite that models your laptop. The failure they prevent is the expensive kind: not a red build, but a green one that was never asking the right question.
Runtime belongs in the same topic because it is a correctness lever, not a comfort one. A suite that takes eleven minutes is run less often, then marked @pytest.mark.skip during a deadline, then deleted in the cleanup sprint.
Doing it
Set ansi, timezone and shuffle partitions in the builder, never in an environment variable a CI runner might not export.
Cap xdist explicitly. -n 4 on a laptop, -n 4 in CI, tuned only if you have measured it.
Keep money in DECIMAL end to end so no tolerance is ever needed in a comparison.
Write one test that asserts the session settings, and let the parity run tell you when a serverless default moves.
Embedding it
Flip ansi.enabled to false in front of the team and re-run the suite. Whatever still passes is a test that was not testing what its name says.
Ask whoever adds a config line to write the test that fails without it. A config with no failing test is a superstition the next upgrade will keep alive.
Give the team the suite-runtime number in every retro. It only degrades, and only they can defend it.
Configs live in conftest.py, not in a Makefile and not in CI YAML.
Never assert a non-ANSI behaviour, such as a bad cast returning NULL. That is encoding a bug as a requirement.
Measure the suite. Over three minutes and it is on its way out of the team's habits.
Gotchas
shuffle.partitions = 1 hides every partition-dependent bug. monotonically_increasing_id looks like a dense sequence from zero, so a surrogate-key test passes; in production the id jumps by 2 to the 33rd at each partition boundary. One partition also makes output order look stable, so assert df.collect()[0] passes locally and is nondeterministic at scale.
spark.conf.set inside a test leaks to every test that runs afterwards, because the session is shared. Which tests those are depends on collection order, so adding an unrelated test file moves the failure. This is what the autouse cleanup fixture is for.
-n auto on a 64-core CI runner starts 64 JVMs. The container is OOM-killed, pytest reports that a worker crashed, and it does not name the test that was running — so it reads as flakiness rather than as a memory limit.
A missing timezone pin does not fail. It shifts order_ts across a midnight boundary, so exactly the tests seeded near 00:00 fail, only in CI, and only for developers in a non-UTC zone.
Test data and comparison
Two rules. The test body contains only the thing under test, and the comparison is done by a function that understands DataFrames.
A factory returns a valid, boring row. Each argument is a delta from valid. That makes the test body a one-line statement of what is unusual about this case, which is also the only documentation the test will ever have.
pythontests/factories.py
from datetime import date, datetime
from decimal import Decimal
from typing import Any
from pyspark.sql.types import (
DateType, DecimalType, IntegerType, StringType, StructField, StructType,
TimestampType,
)
ORDER_LINE_SCHEMA = StructType([
StructField("order_id", StringType(), False),
StructField("line_number", IntegerType(), False),
StructField("customer_id", StringType(), False),
StructField("product_id", StringType(), False),
StructField("order_ts", TimestampType(), False),
StructField("quantity", DecimalType(12, 3), False),
StructField("unit_price", DecimalType(18, 2), False),
StructField("currency", StringType(), False),
StructField("status", StringType(), False),
])
# A boring line that satisfies all seven business rules.
_ORDER_LINE: dict[str, Any] = {
"order_id": "SO-1000",
"line_number": 10,
"customer_id": "C-0001",
"product_id": "P-0001",
"order_ts": datetime(2026, 3, 2, 9, 15),
"quantity": Decimal("2.000"),
"unit_price": Decimal("49.90"),
"currency": "EUR",
"status": "OPEN",
}
def order_line_rows(*overrides: dict[str, Any]) -> list[dict[str, Any]]:
"""One row per argument; each argument is the delta from a valid line.
order_line_rows() -> one valid line
order_line_rows({"status": "CANCELLED"}) -> one cancelled line
"""
return [{**_ORDER_LINE, **o} for o in (overrides or ({},))]
FX_RATE_SCHEMA = StructType([
StructField("currency", StringType(), False),
StructField("rate_date", DateType(), False),
StructField("fx_rate", DecimalType(18, 6), False),
])
# silver.fx_rate on the order date above. EUR is stored at 1 so the join in
# build_fct_order_line needs no special case for the reporting currency.
_FX_RATES: list[dict[str, Any]] = [
{"currency": "EUR", "rate_date": date(2026, 3, 2), "fx_rate": Decimal("1.000000")},
{"currency": "CHF", "rate_date": date(2026, 3, 2), "fx_rate": Decimal("0.940000")},
]
def fx_rate_rows(*overrides: dict[str, Any]) -> list[dict[str, Any]]:
"""No arguments: both currencies. With arguments: one row per delta from CHF."""
return [{**_FX_RATES[1], **o} for o in overrides] or [dict(r) for r in _FX_RATES]
pythontests/gold/test_fct_order_line.py — the test body is the delta
def test_cancelled_lines_are_kept_in_the_fact(spark, rates):
"""Rule 1 excludes cancellations from REVENUE, so the fact must keep the row."""
lines = spark.createDataFrame(
order_line_rows(
{"order_id": "SO-1", "status": "SHIPPED"},
{"order_id": "SO-2", "status": "CANCELLED"},
),
ORDER_LINE_SCHEMA,
)
fct = build_fct_order_line(lines, rates)
assertDataFrameEqual(
fct.select("order_id", "status"),
[Row(order_id="SO-1", status="SHIPPED"),
Row(order_id="SO-2", status="CANCELLED")],
)
Note the explicit schema on createDataFrame. It is not politeness — it is the difference between testing DECIMAL arithmetic and testing floating point.
Why here
The alternative is a checked-in CSV or JSON fixture carrying every column of silver.sales_order_line. A reviewer cannot see which column mattered, because all forty are present and all forty look equally deliberate.
Then the model changes. Adding region_sk means editing thirty fixture files, and the ones you miss do not fail — the column is absent, arrives as NULL, and the test keeps passing while asserting less than it used to. A factory has one place to change.
Doing it
One factory per canonical grain: order_line_rows, customer_rows, fx_rate_rows. Nothing else.
Always pass the explicit schema to createDataFrame. Inference is how DECIMAL becomes DOUBLE.
Keep defaults valid against all seven business rules, so any invalid seed is visible in the test body.
Import the same factories from the SQL-assertion tests, so both kinds of test seed identical data.
Embedding it
The first time someone checks in a CSV fixture, ask them to rewrite one test with the factory in the review. The diff makes the argument better than you can.
Have the team write the factory for the next table before the table exists. It forces the grain conversation while it is still free.
In review, ask which line of the test is the thing under test. If it takes more than a second to point at, the setup is too loud.
Defaults
Defaults are boring and valid; overrides are the interesting bit.
Money and quantities are Decimal, never float.
Factories return plain dicts, so they serve pytest, SQL seeding and a pipeline fixture table alike.
Deterministic values only. A test that depends on the wall clock fails once, at night, and is never reproduced.
Gotchas
createDataFrame inference turns 2.5 into DOUBLE and 2 into LONG. Production has DECIMAL(12,3) and INT, so the suite exercises binary floating point while the warehouse does decimal arithmetic with HALF_UP rounding. A one-cent error in rule 2 is then invisible in every test and visible in the first reconciliation against SAP.
A factory that computes derived values, for example filling amount_eur itself, makes the expected value come from the same logic as the actual. The test then proves that a function equals itself, and it stays green through the bug it was written to catch.
Returning a shared dict rather than a copy lets one test mutate the data of another. The spread in the comprehension is what prevents it; a factory that returns a module-level list does not.
datetime.now in a default. The test passes for months and fails the night an order_ts crosses a day boundary or an SLA window, and nobody can reproduce it in the morning.
#assertDataFrameEqual and what it actually compares
What
PySpark ships the comparison functions; there is no reason to hand-roll one. assertDataFrameEqual for data, assertSchemaEqual for contracts, and they are not interchangeable.
pythontests/gold/test_fct_order_line.py
from decimal import Decimal
from pyspark.sql import Row
from pyspark.testing import assertDataFrameEqual, assertSchemaEqual
from dwh.gold.fct_order_line import build_fct_order_line
from tests.factories import ORDER_LINE_SCHEMA, order_line_rows
from tests.schemas import FCT_ORDER_LINE_SCHEMA
def test_amount_eur_is_rounded_to_the_cent(spark, rates):
"""Rule 2: amount_eur = ROUND(quantity * unit_price * fx_rate, 2)."""
lines = spark.createDataFrame(
order_line_rows({
"quantity": Decimal("3.000"),
"unit_price": Decimal("19.99"),
"currency": "CHF", # rates fixture has CHF -> EUR at 0.940000
}),
ORDER_LINE_SCHEMA,
)
actual = build_fct_order_line(lines, rates).select("order_id", "amount_eur")
# 3.000 * 19.99 * 0.94 = 56.3718 -> 56.37
assertDataFrameEqual(actual, [Row(order_id="SO-1000", amount_eur=Decimal("56.37"))])
def test_fct_order_line_schema_is_the_published_contract(spark, rates):
"""gold.fct_order_line is read by Power BI. Its schema is an interface."""
actual = build_fct_order_line(
spark.createDataFrame(order_line_rows(), ORDER_LINE_SCHEMA), rates
)
assertSchemaEqual(actual.schema, FCT_ORDER_LINE_SCHEMA)
expected may be a DataFrame or a list of Row. For a handful of rows the list is better: it fits on screen and needs no second createDataFrame whose schema you then have to keep in step. Row order is not compared by default, which is correct — Spark promises no ordering without an ORDER BY. Set checkRowOrder=True only when order is part of the contract, which in a warehouse is almost never.
Why here
A hand-written assert actual.collect() == expected fails on row order, fails on Row versus tuple, and prints two walls of text that a reader has to diff by eye. assertDataFrameEqual prints the differing rows and only those — a two-minute fix instead of a fifteen-minute squint.
The schema assertion earns its place separately. gold.fct_order_line is read by Power BI and by gold.mv_sales, so a column rename is a breaking change to a published interface. You want it as a red test in a pull request, not as a broken report on a Monday.
Doing it
assertDataFrameEqual for values, assertSchemaEqual for interfaces. Do not let one stand in for the other.
Declare the expected gold schemas once, in tests/schemas.py, and assert against them from every test that touches the table. One file to diff when the DDL moves.
Select the columns you are asserting about. Comparing all thirty makes every unrelated change a failure in this test.
Embedding it
Break one value on purpose and let the team read the failure output. They will stop writing collect-and-compare within the week.
Ask for the assertion first in review, before the transformation. A test whose assertion is hard to write is usually a function doing two jobs.
Have the team add the schema-contract test for the table BI already reads, then rename a column and watch it fire. That demonstration is what buys the practice.
Defaults
One behaviour per test, named in the test name, with the business rule number in the docstring.
Exact comparison on DECIMAL; no tolerance on money, ever.
Schema contracts asserted for every gold table, in one file, so the interface is visible in one place.
Gotchas
assertDataFrameEqual compares the schema too, nullability included. A frame from createDataFrame carries nullable=False on the fields you declared; the same data after a join comes back nullable=True. The test then fails over a difference that changes no number and no report, and the usual reaction is to weaken the assertion rather than to build expected the same way.
rtol and atol relax FLOAT and DOUBLE comparison. If amount_eur is a double, a tolerance makes a wrong cent pass. Keeping money in DECIMAL removes the temptation, because decimals compare exactly.
Both sides are materialised to the driver. Pointing this at a real gold table under a Connect run collects millions of rows into the client process and either hangs or dies. Use it on seeded data; use the zero-rows SQL pattern for production-scale checks.
Comparing the output of a function against a value computed by importing the same function. It is common, it is always green, and it is worth nothing. The expected value is a constant, written by hand, or it is not an expected value.
Classic, Connect, and the difference that returns wrong numbers
Local pytest runs Spark Classic in one JVM. Serverless compute, standard-access clusters and databricks-connect all run Spark Connect. Same API, different evaluation model — and two of the differences change a number instead of raising.
In Classic, a DataFrame holds an analysed plan built in the same process as the engine. In Connect, it holds an unresolved protobuf plan that travels to a remote server and is resolved there, when an action runs. Almost everything follows from that one sentence.
The four that matter
Spark Classic — local pytest
Spark Connect — serverless, standard access
How it reaches you
Analysis
eager: df.select('nope') raises at that line
lazy: the plan leaves unresolved; the error appears at the action
a pytest.raises around a select passes locally and fails at night
Temp views
the plan captures the view as analysed at that moment
the view NAME is resolved at execution time
replacing a view changes an existing DataFrame's result, silently
Python UDFs
the closure is pickled when the query executes
the closure is pickled when the UDF is declared
a value changed between those two lines gives different numbers
RDD API
spark.sparkContext available
absent
a test helper using parallelize dies with a clear error — the easy case
The temp-view row is the one to internalise, because it returns a number rather than an error:
pythonSame code, same data, two different answers
spark.createDataFrame(
order_line_rows({"status": "SHIPPED"}), ORDER_LINE_SCHEMA
).createOrReplaceTempView("lines")
shipped = spark.table("lines").where("status = 'SHIPPED'")
# ... 200 lines later, another helper reuses the name
spark.createDataFrame(
order_line_rows({"status": "CANCELLED"}, {"status": "CANCELLED"}),
ORDER_LINE_SCHEMA,
).createOrReplaceTempView("lines")
shipped.count()
# Spark Classic -> 1 the view was resolved when `shipped` was built
# Spark Connect -> 0 the name is resolved now, and nothing is SHIPPED
No exception, no warning, no log line. A test that seeds through temp views is green locally and wrong on the platform.
The UDF row has the same shape. The closure is pickled at a different moment, so a value that changes between declaration and execution is read differently by each engine:
pythonUDF closures are captured at different moments
from decimal import Decimal
from pyspark.sql.functions import udf
from pyspark.sql.types import DecimalType
rate = Decimal("1.00") # EUR
to_eur = udf(lambda amount: amount * rate, DecimalType(18, 2))
rate = Decimal("0.94") # CHF, loaded afterwards
df.select(to_eur("unit_price")).collect()
# Spark Classic -> uses 0.94 the closure is pickled when the query runs
# Spark Connect -> uses 1.00 the closure is pickled at declaration
Where exactly the pickle happens under Connect — at udf(...) or at the column expression it builds — is an implementation detail that has moved between client versions. That is the point rather than a caveat to it: the value a UDF closes over is captured at a moment neither engine promises, so any transform whose answer depends on that moment is already wrong. Reproduce it on the client version you actually pin before quoting the mechanism to a client.
Both behaviours are defensible; neither is what the author meant. The fix is not a better UDF — rule 2 is a join to silver.fx_rate on currency and date, which is deterministic, testable in SQL, and free.
Why here
The dangerous class is not fails on Databricks. CI catches that on the first deployment, it costs an afternoon, and everyone learns something. The dangerous class is passes in both places and returns different numbers, because nothing in the process is looking for it.
The exposure here is concrete: currency conversion, SCD2 clock injection and the surrogate-key build are all places where a captured-versus-resolved-later difference changes a value rather than raising. A discrepancy found by a business user six weeks after go-live costs more credibility than the whole suite bought.
Doing it
Seed with DataFrames from factories, not with temp views. The difference disappears if the name never exists.
Avoid Python UDFs in transforms. Built-in functions and joins have identical semantics on both engines and are an order of magnitude faster.
Import exceptions from pyspark.errors, never from pyspark.sql.utils, so an except clause matches under both engines.
Mark the handful of genuinely Classic-only tests with a marker whose reason carries an expiry date.
Embedding it
Run the temp-view example in front of the team, locally and then on the workspace. Two numbers on one screen ends the discussion permanently.
Ask in every review where a test seeds its data. Temp views in a test are the tell; the team learns to spot it faster than you can.
When a Connect difference bites, make the team write the regression test before you explain the cause.
Defaults
Factories and DataFrames for seeding; temp views only where the view itself is under test.
No Python UDFs in the warehouse path. If one is unavoidable, isolate it and test it under both engines.
Exceptions imported from pyspark.errors; client and runtime versions pinned and upgraded together.
Gotchas
A Python UDF that imports a module from workspace files works on the driver and fails inside the UDF on an old runtime. From Databricks Runtime 13.3 LTS, directories added to sys.path, or structured as Python packages, are distributed to all executors automatically; on 12.2 LTS and below they must be installed on executors explicitly. Locally the driver and the executor are the same process, so the suite never sees it.
Testing for a raised AnalysisException around a select. Under Connect nothing is raised until an action, so the pytest.raises block exits without an exception and the test fails with a message about the exception not being raised — which reads like the code got better.
spark.sparkContext in a test helper. It is a clean, loud failure under Connect, which makes it the best of these problems, but it is usually buried in a shared helper that a hundred tests import.
A databricks-connect client whose version does not match the runtime major.minor it points at produces protocol and serialization errors that read like bugs in your query plan. Engineers debug the transformation for hours before checking pip show databricks-connect.
One suite, two engines, two schedules. Pull requests run local Spark and finish in minutes. A scheduled job runs the identical tests against the workspace overnight, and its only job is to disagree with the local run.
bashThe two invocations
# every pull request — local Spark, no workspace, minutes
SPARK_TEST_MODE=local pytest -n 4 --dist loadfile -q
# nightly — the same tests, against serverless.
# RUN_ID comes from the job parameter below; there is no ambient env var for it.
SPARK_TEST_MODE=connect pytest -n 0 -q \
--junitxml=/Volumes/dwh_test/ci/reports/parity-${RUN_ID}.xml
-n 0 disables xdist for the parity run deliberately. Each worker would open its own session to the workspace, multiplying serverless cost by the worker count and running into concurrency limits that surface as flaky, unrepeatable failures. It overrides the -n 4 in addopts because pytest prepends addopts to the command line, so the later flag wins — worth knowing, because the same precedence is what lets a stray addopts entry silently undo a flag you thought you set.
yamlresources/parity.job.yml — defined in the test target only
resources:
jobs:
parity:
name: "parity: pytest under Spark Connect"
schedule:
quartz_cron_expression: "0 30 2 * * ?"
timezone_id: Europe/Zurich
pause_status: UNPAUSED
email_notifications:
on_failure: [dwh-oncall@example.com] # a rota, never a person
tasks:
- task_key: pytest_connect
notebook_task:
notebook_path: ../tests/run_parity
base_parameters:
spark_test_mode: connect
run_id: "{{job.run_id}}"
target_schema: "run_{{job.run_id}}"
The JUnit XML lands in the ci Volume of dwh_test, so a failure leaves an artefact somebody can open six weeks later without re-running anything.
Why here
The divergence class in the previous topic is invisible to a local suite by construction. No amount of local test writing finds it, because the local engine is the thing being wrong about. The only mitigation is to run the same assertions on the real engine.
That converts we find out at go-live into we find out within twenty-four hours, for one serverless run a night. Cheapest insurance in the build phase, and the first thing teams skip.
Doing it
Run the identical suite. Resist a separate connect-only test set: the moment the two diverge, the parity run stops being a parity run.
Schedule it on main, not on a long-lived feature branch, or it reports on code nobody is shipping.
Give it a per-run schema in dwh_test named from the job run id, on the same pattern CI uses, and drop it in a step that runs even when the tests failed.
Route failures to the same rota as production jobs, with the same expectations.
Embedding it
In week two, have the team add the temp-view example as a real test. It goes green locally and red at night. That single demonstration is worth the whole route.
Put the last green parity date on the handover checklist, next to the named owners. A stale date is the honest status of the practice.
When it fails, let the team triage it without you once. Watching them do it is the acceptance test for the coaching.
Defaults
One suite, two modes, one switch. Never two suites.
Nightly, on main, against the compute production uses.
Failures page a rota. A nightly nobody reads is a cron job producing a tick.
Never against dwh_prod. The parity run writes.
Gotchas
A parity run that emails a shared mailbox. It is read for three weeks and then filtered, and the first anyone hears of the divergence is from a business user.
Running parity per pull request because it feels more rigorous. It is ten to fifty times slower, developers start pushing without waiting, and within a month the team has learned to ignore a red build.
Pointing the parity run at dwh_prod for realistic data. It creates tables, and eventually one of them shares a name with something real. Realistic data is a fixtures problem, not a catalog problem.
Letting the connect suite quietly skip. If credentials expire, pytest collects zero tests and exits 0, which is indistinguishable from success in the job UI. Assert a minimum collected-test count in the run script.
Straight to the documentation
The generic platform material lives in the official docs and is not restated here. These are the pages worth having open while you work through this route.
Databricks Connect for Python — the install and version matrix — the client must match the runtime major.minor you point it at
Databricks Connect limitations — read before assuming a local pass means anything: sparkContext, the RDD API and several APIs are simply absent
ANSI compliance in Databricks SQL — the exact list of expressions whose behaviour changes, and the error classes you will start seeing