pytest-pyspark-utils

A pytest plugin that provides a reusable spark session fixture and automated Delta table caching for PySpark testing. Eliminates boilerplate Spark session setup and speeds up tests by caching CSV/JSONL-to-Delta conversions.

Features

• Session-scoped spark fixture — one Spark session per test run, shared across all tests
• Automatic Delta JAR detection — resolves the correct Delta Lake Maven coordinates from the installed PySpark version
• Optional manual Delta Lake configuration via Maven JAR coordinates
• Delta table caching — CSV/JSONL files are converted to Delta once and cached between runs
• Per-test isolation — each test gets its own copy of the Delta tables via the delta_tables fixture
• PySpark version-agnostic — works with PySpark 3.x and 4.x
• Configurable via pytest.ini, pyproject.toml, or CLI flags

Installation

Install the plugin with your chosen PySpark version:

# PySpark 4.x
pip install "pytest-pyspark-utils[pyspark4]"

# PySpark 3.x
pip install "pytest-pyspark-utils[pyspark3]"

# Or pin an exact version alongside the plugin
pip install pytest-pyspark-utils pyspark==4.0.2

Usage of fixtures

Spark fixture

Once installed, the spark fixture is automatically available in all your tests — no import or conftest wiring needed.

from pyspark.sql import SparkSession

def test_something(spark: SparkSession):
    df = spark.createDataFrame([(1, "a"), (2, "b")], ["id", "value"])
    assert df.count() == 2

Delta_tables fixture

The plugin can automatically convert your test data files (CSV or JSONL) into cached Delta tables and register them as Spark SQL tables. Each test gets an isolated copy.

1. Organize your test data

tests/test_my_feature/
├── conftest.py
├── input/
│   ├── users.csv
│   └── orders.csv
├── expected/
│   └── results.csv
└── test_my_feature.py

2. Define your table config in conftest.py

# tests/test_my_feature/conftest.py
import pytest
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from pytest_pyspark_utils import TableConfig

users_schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
])

orders_schema = StructType([
    StructField("order_id", IntegerType(), True),
    StructField("user_id", IntegerType(), True),
    StructField("amount", IntegerType(), True),
])

@pytest.fixture(scope="module")
def delta_tables_config():
    return {
        "users": TableConfig(
            source="input",
            schema=users_schema,
            partition_by=["id"],
        ),
        "orders": TableConfig(
            source="input",
            schema=orders_schema,
        ),
        "expected_results": TableConfig(
            source="expected",
            schema=orders_schema,
        ),
    }

3. Use delta_tables in your tests

# tests/test_my_feature/test_my_feature.py
def test_user_orders(spark, delta_tables):
    users = spark.table("users")
    orders = spark.table("orders")

    result = users.join(orders, users.id == orders.user_id)
    assert result.count() > 0

def test_another_scenario(spark, delta_tables):
    # Each test gets a fresh, isolated copy of all tables
    spark.sql("DELETE FROM users WHERE id = 1")
    assert spark.table("users").count() == 4  # won't affect other tests

How delta_tables works

The fixture chain operates in two layers:

1. Module-level caching (runs once per test file): Reads CSV/JSONL files, converts them to Delta format, and caches the result in <test_dir>/_delta_cache/. On subsequent runs, if the source file and schema haven't changed, the cached Delta is reused instantly.

2. Function-level isolation (runs per test): Copies the cached Delta tables to a temporary directory, drops any existing Hive tables, and re-registers fresh tables pointing to the isolated copy.

This means the first run pays the conversion cost, but subsequent runs are fast — and every test is guaranteed a clean slate.

TableConfig reference

@dataclass
class TableConfig:
    source: str = "input"           # "input", "expected", or an absolute path
    schema: Optional[StructType] = None  # PySpark schema (recommended for consistency)
    table_name: Optional[str] = None     # Defaults to the dict key
    partition_by: Optional[List[str]] = None
    liquid_clustering: bool = False

Configuration

Delta Lake (automatic)

By default, the plugin automatically detects the installed PySpark version and resolves the correct Delta Lake Maven JAR coordinates — no configuration required.

# Just use the spark fixture — Delta Lake works out of the box
def test_with_delta(spark, delta_tables):
    spark.table("my_table").show()

If auto-detection fails (e.g. an unsupported PySpark version), the fixture falls back to a plain Spark session without Delta extensions and prints a warning.

Delta Lake (manual override)

You can override the auto-detected JAR by setting delta_jar explicitly in pyproject.toml:

[tool.pytest.ini_options]
delta_jar = "io.delta:delta-spark_2.13:4.0.1"
spark_app_name = "my-project-tests"
spark_driver_memory = "4g"
delta_cache_dir = "_delta_cache"

Or in pytest.ini:

[pytest]
delta_jar = io.delta:delta-spark_2.13:4.0.1
spark_app_name = my-project-tests
spark_driver_memory = 4g
delta_cache_dir = _delta_cache

Or pass it directly on the command line:

pytest --delta-jar=io.delta:delta-spark_2.13:4.0.1

Delta JAR coordinates by PySpark version

The following versions are auto-detected. Use this table if you need to set delta_jar manually:

Programmatic access

The determine_delta_jar() function is available as a public API:

from pytest_pyspark_utils import determine_delta_jar

# Auto-detect from installed PySpark
jar = determine_delta_jar()

# Or pass a specific version
jar = determine_delta_jar("3.5.3")
# => "io.delta:delta-spark_2.12:3.3.2"

Available options

How it works

The plugin registers several fixtures via the pytest entry point:

License

GNU GPL v3.0+