Testing and Debugging Spark Data Pipelines

Reading note: the practical examples and exercises are optional. Focus on the concepts, architecture, assumptions, and design decisions; no execution or project submission is required for this reading path.

Start with an observable promise

A successful Spark job can still publish the wrong answer. Before writing a test, state the row grain, required columns and types, null policy, duplicate policy, time boundary, and expected destination effect. For a table of accepted measurements, zero may be valid while a missing or negative amount is rejected. Dropping zero rows would violate that rule even if the job completed.

Unit and integration tests address different boundaries. A focused transformation test supplies controlled DataFrames; a storage integration test actually writes and reads a format. Representative scale tests address resource pressure, execution variability, and recovery. These levels complement each other. Small examples can expose a hot-key distribution or a bad skew-handling rule, but cannot establish production latency or memory behavior.

Fixed examples are useful evidence, not a promise to catch most defects. Run quality checks on current input too, define whether a violation fails publication or enters a reviewed rejection path, and retain enough evidence to investigate. Source completeness, correctness of business definitions, and downstream side effects need checks beyond the transformation.

Build a fixture and an independent answer

A test fixture supplies controlled starting conditions: small input rows with an explicit schema, a Spark session with known settings, and temporary storage where needed. Use synthetic values that isolate a rule. Include the boundary itself and nearby cases, not only normal data. Empty and all-null input often need an explicit schema because inference has too little information.

A test oracle is the basis for deciding whether the observed result is correct. For a tiny fixture, write the expected rows from the stated rule rather than running the same transformation twice. Comparing a function with itself checks repeatability, not the intended rule. A manually specified answer can also be wrong, so explain why each row belongs or is rejected.

The lab has five rows: A with zero and B with ten are accepted; C has no amount, D is negative, and the last row has no ID. The expected answer is exactly A and B, with their values. Two deliberately wrong variants remove A or duplicate B. The test must reject both. This checks that the assertions are sensitive to these defects, not merely that the current implementation passes.

Compare meaning instead of incidental layout

A DataFrame has no guaranteed global row order without an ordering operation. Compare an unordered multiset when order is not part of the contract; a set loses duplicate counts. Sorting is useful only with a defined comparison and enough keys to resolve relevant ties. Floating-point sums may differ with execution order, so choose justified absolute and relative tolerances before comparing them.

Check schema separately, including column names, order, types, and the nullability or metadata that the contract requires. In current PySpark, assertDataFrameEqual has options for row order, numeric tolerance, and ignoring nullable differences. Do not assume its defaults express your contract. A schema permitting nulls is different from finding actual null values. Storage readers may widen nullability even when every stored value is present.

The lab uses Counter on collected tuples of strings and integers. It preserves duplicate multiplicity and compares exact values for this tiny fixture. It is not a general comparator for nested values, NaNs, or production-size data. For large comparisons, use a deliberate reconciliation strategy and explain what sampling or hashes can miss. Equal row counts and a grand total can hide changed keys.

A function accepting and returning DataFrames is easier to isolate when paths, clocks, sessions, and side effects are explicit. It is not automatically pure: nondeterministic expressions, mutable sources, UDF state, and session settings can change results. Check repeatability under relevant partition counts and settings without insisting on one physical plan across versions.

Exercise the real boundary you depend on

A Parquet round trip can test values, types, and empty output with the actual reader and writer. It does not test cloud permissions, a production catalog, a table format’s transaction protocol, or compatibility with a different consumer. Add those tests against representative integrations, with isolated paths and cleanup. Preserve failed-run evidence when cleanup would otherwise erase the reproduction.

For publication, test failure before commit, repeated attempts, existing-output preservation, and the reader’s view during replacement. Repeating an append and seeing doubled rows is a failure only if the intended operation was a retry of the same logical load. Define that identity first. A contract test can detect the incompatible change exercised by its fixtures; it cannot guarantee that every future producer change will be detected before release.

The lab writes accepted rows and an empty DataFrame to two temporary Parquet locations and reads them back. It checks column names and types and checks that accepted values contain no nulls. It does not require a non-nullable storage schema or simulate an atomic table replacement. Spark stops before the temporary directory is removed.

Reduce the failure without removing its cause

An error message and traceback show where a failure surfaced. Spark can report a driver-side analysis error before a job starts, an executor-side data error during an action, or a storage/commit failure after computation. Follow the relevant cause chain and structured error condition, not only the outer Py4J wrapper. Preserve the application and attempt ID, code and dependency versions, session settings, input snapshot, plan, task context, and affected output state.

Lazy evaluation means constructing a transformation is not sufficient to exercise all its behavior. Trigger an action that reads the affected expression. count() may allow Spark to prune an unused projection, so it is not always a test of a cast or UDF. The lab constructs an invalid integer cast and then collects that value under ANSI mode; the expected CAST_INVALID_INPUT condition confirms the failure actually occurred.

A minimal reproducible example preserves the failure while removing unrelated work. Reduce columns, rows, and dependencies one at a time and rerun. Preserve a problematic value, duplicate relationship, timestamp boundary, encoding, or execution condition when it matters. If the error disappears locally, record the difference instead of declaring the production incident fixed.

Distinguish data-dependent failures from transient infrastructure failures using evidence. Bounded retries can be appropriate for transient errors only when the destination effect is safe to repeat. Do not replace every invalid value with null, silently drop bad rows, or disable ANSI mode merely to obtain a green run. Choose a reviewed reject, correction, or fallback policy and test its output.

Keep the reduced failing case as a regression test, show that the old or deliberately broken behavior fails it, and rerun the affected integration and representative scale checks. Report what was actually executed. Passing local tests supports those cases; it does not establish production throughput, cluster recovery, or complete data capture.

Lab: run the tests in local Spark

Use Python 3.12, pyspark==4.2.0, and Java 17 with JAVA_HOME set. Driver and Python workers use the same Python minor version. Run the setup and then each example in a process without another Spark session. UTC, ANSI mode, and AQE disabled are explicit test conditions. Each example stops its session and removes temporary files. The invalid-cast test deliberately produces Spark ERROR logs before catching and checking the expected exception; startup warnings and those logs are separate from the recorded stdout. This lab does not implement a persistent rejected-row store.

import os, sys, tempfile
from pathlib import Path
from contextlib import contextmanager, redirect_stdout
from io import StringIO
from decimal import Decimal
from pyspark.sql import SparkSession, functions as F, types as T

os.environ["PYSPARK_PYTHON"] = sys.executable
os.environ["SPARK_LOCAL_IP"] = "127.0.0.1"

@contextmanager
def local_spark():
    with tempfile.TemporaryDirectory(prefix="de025-") as work:
        spark = (SparkSession.builder.master("local[2]").appName("testing-lab")
                 .config("spark.driver.host", "127.0.0.1")
                 .config("spark.driver.bindAddress", "127.0.0.1")
                 .config("spark.ui.enabled", "false")
                 .config("spark.ui.showConsoleProgress", "false")
                 .config("spark.sql.warehouse.dir", work)
                 .config("spark.sql.session.timeZone", "UTC")
                 .config("spark.sql.timestampType", "TIMESTAMP_LTZ")
                 .config("spark.sql.ansi.enabled", "true")
                 .config("spark.sql.shuffle.partitions", "8")
                 .config("spark.sql.autoBroadcastJoinThreshold", "-1")
                 .config("spark.sql.adaptive.enabled", "false")
                 .getOrCreate())
        spark.sparkContext.setLogLevel("ERROR")
        try:
            yield spark, Path(work)
        finally:
            spark.stop()

from collections import Counter
from pyspark.errors import NumberFormatException

schema = T.StructType([T.StructField("id", T.StringType(), True),
                       T.StructField("amount", T.LongType(), True)])

def fixture(spark):
    return spark.createDataFrame([("A",0),("B",10),("C",None),("D",-1),(None,5)], schema)

def accepted(df):
    return df.filter(F.col("id").isNotNull() & F.col("amount").isNotNull() & (F.col("amount") >= 0))

def bag(df):
    return Counter(tuple(r) for r in df.select("id", "amount").collect())

Check exact rows, schema, empty input, and two incorrect variants.

with local_spark() as (spark, work):
    source = fixture(spark)
    good = accepted(source)
    expected = Counter([("A",0),("B",10)])
    assert good.schema == schema
    assert bag(good) == expected
    wrong = good.filter("amount > 0")
    assert bag(wrong) != expected
    duplicated = good.unionByName(good.filter("id = 'B'"))
    assert set(bag(duplicated)) == set(expected) and bag(duplicated) != expected
    empty = spark.createDataFrame([], schema)
    assert accepted(empty).schema == schema and bag(accepted(empty)) == Counter()
    assert source.count() - good.count() == 3
    print("expected rows and schema match:", True)
    print("zero-dropping and duplicate mutants detected:", True)
    print("empty input supported; rejected rows:", 3)
# expected rows and schema match: True
# zero-dropping and duplicate mutants detected: True
# empty input supported; rejected rows: 3

Write and read accepted and empty Parquet output.

with local_spark() as (spark, work):
    result = accepted(fixture(spark))
    path = work / "result"
    result.write.parquet(str(path))
    restored = spark.read.parquet(str(path))
    assert restored.dtypes == [("id","string"),("amount","bigint")]
    assert bag(restored) == Counter([("A",0),("B",10)])
    assert restored.filter("id IS NULL OR amount IS NULL").count() == 0
    empty_path = work / "empty"
    spark.createDataFrame([], schema).write.parquet(str(empty_path))
    restored_empty = spark.read.parquet(str(empty_path))
    assert restored_empty.dtypes == restored.dtypes and restored_empty.count() == 0
    print("Parquet round trip preserves rows and types:", True)
    print("empty Parquet output remains readable:", True)
# Parquet round trip preserves rows and types: True
# empty Parquet output remains readable: True

Reproduce the invalid cast at an action and vary input partition count.

with local_spark() as (spark, work):
    raw = spark.createDataFrame([("not-an-integer",)], "raw string")
    expression = raw.select(F.col("raw").cast("long").alias("number"))
    print("invalid-value expression constructed:", True)
    try:
        expression.collect()
    except NumberFormatException as error:
        assert error.getCondition() == "CAST_INVALID_INPUT"
        print("action exposes condition:", error.getCondition())
    else:
        raise AssertionError("invalid cast unexpectedly succeeded")
    expected = Counter([("A",0),("B",10)])
    for partitions in (1,3):
        assert bag(accepted(fixture(spark).repartition(partitions))) == expected
    print("result preserved across one and three input partitions:", True)
# invalid-value expression constructed: True
# action exposes condition: CAST_INVALID_INPUT
# result preserved across one and three input partitions: True

References: Testing PySpark, assertDataFrameEqual options, Spark ANSI behavior.