pyspark-contracts

Lightweight schema and data quality validation for PySpark DataFrames.

Full documentation: pyspark-contracts.readthedocs.io

Validates at runtime — before your pipeline runs — with a structured JSON report queryable in CloudWatch Logs Insights, Datadog, or any log aggregator.

Install

pip install pyspark-contracts

PySpark is assumed to be on your classpath (Glue, Databricks, EMR). For local development: pip install pyspark-contracts[pyspark]

Tested on PySpark 3.3–4.x and Python 3.10–3.14, in both classic mode and Spark Connect.

Quick start

from pyspark.sql.types import DateType, FloatType, StringType
from pyspark_contracts import Contract, Field

class OdometerContract(Contract):
    vin           = Field(StringType(), nullable=False, min_length=17, max_length=17)
    odometerStart = Field(FloatType(),  nullable=False, min_value=0)
    odometerEnd   = Field(FloatType(),  nullable=False, min_value=0)
    readingDate   = Field(DateType())

# Hard fail (default) — logs ERROR + raises ContractViolationError
OdometerContract().validate(df)

# Soft fail — logs WARNING + returns ViolationReport
report = OdometerContract().validate(df, mode="soft")
if report:
    print(f"{len(report.violations)} violation(s) found")

Validation modes

Mode	On violation	Return
"hard" (default)	Logs ERROR (JSON), raises ContractViolationError	—
"soft"	Logs WARNING (JSON), continues	ViolationReport

Lazy vs fail-fast

By default, "hard" mode stops at the first violation found (fewer Spark actions, faster feedback), while "soft" mode collects every violation so you get the full picture in one run. Override either with lazy:

# Hard mode, but collect every violation before raising
OdometerContract().validate(df, lazy=True)

# Soft mode, but stop at the first violation
report = OdometerContract().validate(df, mode="soft", lazy=False)

Cross-column checks

Field constraints are column-local. For rules spanning more than one column, use condition (simple two-column comparisons) or @check (arbitrary logic):

from pyspark_contracts import Contract, Field, check
from pyspark.sql import DataFrame
from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType, TimestampType

class FinalDfContract(Contract):
    end_dt   = Field(TimestampType(), nullable=False)
    start_dt = Field(TimestampType(), nullable=False,
                     condition=lambda f: F.col(f) < F.col("end_dt"),
                     condition_description="start_dt must precede end_dt")

    delta_km = Field(DoubleType(), nullable=False)

    @check("delta_km must be non-negative (no regression)")
    def no_regression(self, df: DataFrame) -> DataFrame:
        return df.filter(F.col("delta_km") < 0)

    @check("delta_km must not exceed daily threshold")
    def no_abnormal_jump(self, df: DataFrame, threshold: float = 1000.0) -> DataFrame:
        return df.filter(F.col("delta_km") > threshold)

A @check method receives the full DataFrame and returns the failing rows — an empty result means it passed. Extra parameters (threshold) are supplied via validate() and routed to the checks that declare them:

report = FinalDfContract().validate(df, mode="soft", threshold=2000.0)

Both condition and @check are skipped entirely when the DataFrame has a missing column or type mismatch, since either can reference arbitrary columns.

Validating pipeline functions

@check_input and @check_output move contract validation from the function body onto its signature, so a refactor can’t accidentally drop it:

from pyspark_contracts import check_input, check_output

@check_input(AccumulatedDataContract, param="accumulated_df")
@check_output(FinalDfContract)
def build_final_df(accumulated_df: DataFrame, contracts_df: DataFrame) -> DataFrame:
    return accumulated_df.join(contracts_df, ...)

check_input validates the named parameter before the function runs — a bad step never executes. check_output validates the return value after the function runs, before it’s returned to the caller. Both accept mode and any extra kwargs, forwarded to validate() exactly as if you’d called it directly:

@check_output(FinalDfContract, mode="soft", threshold=2000.0)
def build_final_df(df: DataFrame) -> DataFrame:
    ...

A function needing more than one input validated stacks @check_input:

@check_input(AccumulatedDataContract, param="accumulated_df")
@check_input(ContractsContract, param="contracts_df")
def build_final_df(accumulated_df, contracts_df): ...

Validation depth

from pyspark_contracts import ValidationDepth

# Column names and types only — reads df.schema, triggers zero Spark actions
OdometerContract().validate(df, depth=ValidationDepth.SCHEMA_ONLY)

# Skip structural checks, run only nullable/range/length/regex/condition/@check
OdometerContract().validate(df, depth=ValidationDepth.DATA_ONLY)

# Default: both
OdometerContract().validate(df, depth=ValidationDepth.SCHEMA_AND_DATA)

SCHEMA_AND_DATA is the default and runs the full pipeline described everywhere else in this README — every schema and quality check, exactly as if depth were never passed.

SCHEMA_ONLY is particularly useful in unit tests where the DataFrame is built from a known factory: the schema is guaranteed by construction, so only the data constraints are worth checking — the reverse case, DATA_ONLY, skips column/type checks and assumes the schema is already right, without adding a safety net if it isn’t.

Dataset-level checks

min_rows / max_rows are declared as plain class attributes, not Field options — they assert something about the DataFrame as a whole rather than a single column:

class MyContract(Contract):
    min_rows = 100
    max_rows = 1_000_000
    vin = Field(StringType())

Checked right after the row count is known, so they apply under DATA_ONLY and SCHEMA_AND_DATA depth (not SCHEMA_ONLY, which never computes a row count). Unlike @check/condition, min_rows/max_rows are plain data, so they round-trip through to_dict()/ContractSchema with no “can’t reconstruct” warning.

Disabling validation

PYSPARK_CONTRACTS_ENABLED=false  # skips all validation, no code changes needed

Useful when performance is critical and the schema is already trusted — e.g. the read right after a validated write.

Bringing your own logger

import logging
logger = logging.getLogger("my-etl-job")
OdometerContract().validate(df, logger=logger)

If no logger is passed, the library emits JSON to stdout.

Field options

Option	Type	Description
nullable	bool	Default True. Set False to fail on null values.
min_value	numeric / date	Minimum allowed value.
max_value	numeric / date	Maximum allowed value.
min_length	int	Minimum string length.
max_length	int	Maximum string length.
regex	str	Regex pattern the entire value must match — anchored for you, so "[0-9]+" rejects "abc123".
allowed_values	list	Allowlist of valid values.
unique	bool	Default False. Set True to fail when the column has duplicate (non-null) values.

Documenting a contract

description and metadata on Field are pure documentation — they never affect validation:

class AssetHdrTelemetryReadingContract(Contract):
    asset_hdr_id = Field(LongType(), nullable=False,
                         description="AF internal vehicle identifier")
    meter_reading_value = Field(DecimalType(26, 12), nullable=False, min_value=0,
                                description="Odometer reading in km",
                                metadata={"unit": "km", "source": "asset_hdr_telemetry_reading"})

Export the whole contract as data — types, constraints, descriptions, and the name/description of every @check (never the executable code, which can’t be serialized):

schema_dict = AssetHdrTelemetryReadingContract.to_dict()
json_str = AssetHdrTelemetryReadingContract.to_json()

Or print a quick human-readable summary:

print(AssetHdrTelemetryReadingContract.describe())

Sharing a schema without the Python class

Reload an exported schema and validate with it, without importing the original Contract subclass:

from pyspark_contracts import ContractSchema

schema_dict = AssetHdrTelemetryReadingContract.to_dict()
ContractSchema.from_dict(schema_dict).validate(df)

# Or straight from JSON
ContractSchema.from_json(json_str).validate(df)

ContractSchema reconstructs everything to_dict() captured as data — types, nullability, and every quality constraint. It cannot reconstruct @check methods or condition lambdas (arbitrary Python code was never serialized) — from_dict()/from_json() emit a warning naming any such rules the original contract had, so you know they won’t be enforced.

Log output

Both modes emit the same structured JSON before acting:

{
  "level": "ERROR",
  "message": "contract violation — job aborted",
  "contract": "OdometerContract",
  "mode": "hard",
  "depth": "schema_and_data",
  "violations": [
    {"kind": "missing_column", "column": "odometerStart", "expected_type": "float"},
    {
      "kind": "value_out_of_range",
      "column": "odometerEnd",
      "constraint": "min_value=0",
      "row_pct": 2.3,
      "failure_count": 42,
      "sample_values": [-1.0, -5.5, -12.0]
    }
  ],
  "row_count": 1842
}

CloudWatch Logs Insights

-- All contract violations in the last 7 days
filter message = "contract violation — job aborted"
| stats count(*) as failures by contract

-- Violations by column across all runs
filter ispresent(violations)
| flatten violations
| stats count(*) by violations.kind, violations.column

-- Failing @check methods across all runs
filter ispresent(violations)
| flatten violations
| filter violations.kind = "check_failed"
| stats count(*) by contract, violations.check

Violation kinds

Each violation’s kind is a ViolationKind member. It’s a str enum, so comparing it to the plain string still works:

from pyspark_contracts import ViolationKind

nulls = [v for v in report.violations if v.kind == ViolationKind.NULL_VIOLATION]

Kind	Trigger
missing_column	Column declared in contract is absent from the DataFrame
type_mismatch	Column exists but type differs from declared type (reported as Spark type strings, e.g. decimal(26,12))
null_violation	nullable=False column contains null values
value_out_of_range	Value below min_value or above max_value
length_out_of_range	String length below min_length or above max_length
regex_mismatch	Value does not match regex pattern
value_not_allowed	Value not in allowed_values list
condition_failed	Field(condition=...) expression evaluated to false
check_failed	@check-decorated method returned non-empty failing rows (named in check, not column)
duplicate_value	Field(unique=True) column has repeated (non-null) values
row_count_out_of_range	DataFrame row count below min_rows or above max_rows

Performance note

The row count and every column constraint (nullable, ranges, lengths, regex, allowed_values, condition) are checked together in a single Spark action, however many the contract declares. On top of that:

Adds one Spark action each	Why
unique field	Needs a groupBy, which can’t share the single pass
@check method	Counts the rows it returns
Violation found	Fetches its sample_values

So a contract without unique or @check reads a passing DataFrame once. If yours has several of them, or you expect violations, and the DataFrame is expensive to recompute, cache it before validating:

df = df.cache()
OdometerContract().validate(df)

See BENCHMARKS.md for measured overhead.