pyspark-contracts

Lightweight schema and data quality validation for PySpark DataFrames.

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]

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

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)

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

Log output

Both modes emit the same structured JSON before acting:

{
  "level": "ERROR",
  "message": "contract violation — job aborted",
  "contract": "OdometerContract",
  "mode": "hard",
  "violations": [
    {"kind": "missing_column", "column": "odometerStart", "expected_type": "FloatType"},
    {
      "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

Violation kinds

Performance note

Each quality constraint triggers a Spark action (filter + count). For DataFrames with many constraints, consider caching before validation:

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