Home / Tutorials / pytest

PYSPARK TESTING

Test a small PySpark DataFrame with pytest

Example tested with PySpark 4.2.0 and pytest

Start one local Spark session for the test run. Feed the transformation a tiny DataFrame and assert the business result, not Spark's internal plan.

Create a session fixture

import pytest
from pyspark.sql import SparkSession

@pytest.fixture(scope="session")
def spark():
    session = (
        SparkSession.builder
        .master("local[2]")
        .appName("pipeline-tests")
        .config("spark.sql.shuffle.partitions", "2")
        .getOrCreate()
    )
    session.sparkContext.setLogLevel("ERROR")
    yield session
    session.stop()

Session scope avoids paying Spark's startup cost for every test. Two local threads are enough for these small examples.

Test rows that describe the rule

from src.pipeline import transform

def test_transform_keeps_unmatched_sale(spark):
    sales = spark.createDataFrame(
        [(101, "Alice", 250.50, 25.05),
         (102, "David", 120.00, 6.00)],
        ["transaction_id", "customer_name", "net_amount", "tax_amount"],
    )
    customers = spark.createDataFrame(
        [("Alice", "London")],
        ["customer_name", "city"],
    )

    rows = transform(sales, customers).collect()

    assert [row.transaction_id for row in rows] == [101, 102]
    assert rows[0].gross_amount == pytest.approx(275.55)
    assert rows[1].city is None

pytest.approx() avoids brittle equality checks for floating-point calculations. Collecting is appropriate here because the fixture has only two rows. Do not collect production-sized data in a test.

Run the test

python -m pytest -q

1 passed