Designing a PySpark Pipeline That Survives Schema Drift

Every pipeline that ingests third-party data eventually breaks because someone upstream renamed a column. The interesting question is not how to prevent that — you cannot — but where in your pipeline you find out.

The default answer is “somewhere downstream, weeks later, when a number looks wrong.” That is the expensive answer.

The failure mode

A vendor adds a field. Your SELECT * ingestion picks it up. Nothing breaks. Three months later they rename customer_id to cust_id, your join silently produces zero matches, and a monthly aggregate quietly reports a smaller population than it should.

Spark will not help you here. It is perfectly happy to join on a column that matches nothing and hand you an empty DataFrame.

Declare the schema you expect

The single highest-value change is refusing to infer schemas on ingestion.

from pyspark.sql.types import (
    StructType, StructField, StringType, TimestampType, DoubleType,
)

CLAIMS_SCHEMA = StructType([
    StructField("claim_id",    StringType(),    nullable=False),
    StructField("customer_id", StringType(),    nullable=False),
    StructField("opened_at",   TimestampType(), nullable=False),
    StructField("reserve",     DoubleType(),    nullable=True),
])

df = (
    spark.read
    .schema(CLAIMS_SCHEMA)
    .option("mode", "FAILFAST")
    .parquet(source_path)
)

FAILFAST is the important half. The default PERMISSIVE mode nulls out whatever it cannot parse and carries on, which is precisely the behaviour that turns a schema change into a silent data-quality incident.

Then assert on what arrived

A declared schema catches type and name changes. It does not catch a column that is still present, still a string, and now empty.

from pyspark.sql import functions as F

def assert_populated(df, columns, threshold=0.99):
    """Fail if any required column is null more often than expected."""
    total = df.count()
    if total == 0:
        raise ValueError("Source produced zero rows")

    stats = df.agg(*[
        (F.count(F.col(c)) / F.lit(total)).alias(c) for c in columns
    ]).collect()[0].asDict()

    bad = {c: round(r, 4) for c, r in stats.items() if r < threshold}
    if bad:
        raise ValueError(f"Columns below {threshold} populated: {bad}")

Run it at the boundary, before the first join. The cost is one extra pass over the source; the alternative is discovering the problem in a dashboard.

A pipeline that fails at 02:00 is an operational problem. A pipeline that succeeds with wrong numbers is a credibility problem, and it costs far more to recover from.

Keep the contract next to the data, not in the code

Once you have several sources, the schemas want to live somewhere they can be reviewed independently of the transformation logic:

• One schema definition per source, versioned
• The expected population thresholds alongside it
• A changelog entry whenever a vendor changes something

This is not glamorous, but it turns “the pipeline broke” into “vendor X changed field Y on date Z,” which is a conversation you can actually have with the vendor.

What this buys you

None of this makes the pipeline more capable. It makes it honest about when it cannot do its job, which for anything feeding a model is the more valuable property.