Schema evolution: a Spark & Data Eng guide for developers

Introduction

Data pipelines rarely stay frozen once they go to production. Producers add new fields, rename columns, or change data types—and when they do, your Spark jobs either break or silently fail. Schema evolution: a Spark & Data Eng guide for developers walks through the patterns we use to handle these inevitable changes without rewriting your entire ETL infrastructure.

Over 70 days of this series, we've built streaming pipelines, optimized queries, and wrangled messy data. Today we're solving one of the most practical problems in production data engineering: how to accommodate schema changes gracefully, maintain data quality, and alert the team when something unexpected arrives at the door.

Add columns as nullable

The simplest form of schema evolution is adding new columns to upstream data. In many cases, this shouldn't break your Spark job—but only if you handle nullability correctly.

When a producer adds a new field, they should add it as nullable by default. On the consumer side (your job), Spark's default behavior is to ignore unknown columns when reading structured data, so new fields won't cause parse errors. However, if you use strict schema validation or have downstream operations that assume specific columns exist, you'll need explicit handling.

Here's a realistic scenario: your payment service adds a new settlement_method column to events. Your existing Spark job reads the data fine, but any aggregation or report that assumes only certain columns exist may break.

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, LongType, BooleanType

spark = SparkSession.builder.appName("schema-evolution").getOrCreate()

# Original schema
original_schema = StructType([
    StructField("transaction_id", LongType(), False),
    StructField("amount", LongType(), False),
    StructField("timestamp", LongType(), False),
])

# New schema with nullable column added
evolved_schema = StructType([
    StructField("transaction_id", LongType(), False),
    StructField("amount", LongType(), False),
    StructField("timestamp", LongType(), False),
    StructField("settlement_method", StringType(), True),  # nullable=True
])

# Read with permissive mode to handle missing columns
df = spark.read \
    .option("mode", "PERMISSIVE") \
    .json("s3://data/payments/")

# Select only the columns you know about; new ones are ignored
df_processed = df.select("transaction_id", "amount", "timestamp")
print(df_processed.schema)

The key is nullable=True when defining new columns. If you're using auto-inferred schemas (which we generally avoid in production), be explicit about allowing new fields to be null. In Delta Lake or structured sources, set spark.sql.storeAssignmentPolicy to ANSI to enforce strict type safety and catch mismatches early.

mergeSchema in Delta

Delta Lake gives us a powerful tool for schema evolution: mergeSchema. This option automatically merges the schema of new data with the existing table schema, adding any new columns as nullable fields at the end.

This is invaluable for append-heavy workloads where producers iterate on their schema. Instead of manual schema management or breaking your pipeline, Delta handles it.

from pyspark.sql.types import StructType, StructField, StringType, LongType

# Simulate data arriving with extra column
new_data = spark.createDataFrame([
    {
        "user_id": 123,
        "email": "alice@example.com",
        "age": 28,
        "loyalty_tier": "gold"  # New column
    },
    {
        "user_id": 124,
        "email": "bob@example.com",
        "age": 35,
        "loyalty_tier": "silver"
    }
], schema="user_id long, email string, age int, loyalty_tier string")

# Write with mergeSchema enabled
new_data.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save("s3://data/users_delta/")

# Read back and inspect
result = spark.read.format("delta").load("s3://data/users_delta/")
result.printSchema()

When mergeSchema=true, Delta updates the table's metadata to include loyalty_tier as a new nullable column. Older rows will have NULL for this field. This approach works smoothly for additive changes, but doesn't handle deletions, renames, or type changes—those require different strategies.

For type changes (e.g., age from int to string), Delta requires explicit casting or a separate migration. For deletions, you need a schema contract (see below) to detect and alert rather than silently dropping data.

Contracts with producers

The best defense against schema chaos is contracts between producers and consumers. A contract is a documented agreement about what columns will always exist, their types, and which changes are allowed.

We typically document this as a JSON or YAML spec that both teams maintain. The producer commits to not removing certain core fields or changing their types without a deprecation window. The consumer builds validation into the pipeline to enforce the contract.

Here's a practical example using a contract file:

# contract.yaml (checked into git, shared between teams)
# version: 1.0
# required_columns:
#   - name: order_id
#     type: long
#     nullable: false
#   - name: customer_id
#     type: long
#     nullable: false
#   - name: total_amount
#     type: decimal(10, 2)
#     nullable: false
#   - name: created_at
#     type: timestamp
#     nullable: false
# allowed_new_columns: true
# allowed_type_changes: false
# allowed_column_removal: false

import yaml
from pyspark.sql.types import DataType, LongType, DecimalType, TimestampType

def load_contract(path):
    with open(path, "r") as f:
        return yaml.safe_load(f)

def validate_against_contract(df, contract):
    """Validate DataFrame against schema contract."""
    schema_map = {field.name: field.dataType for field in df.schema.fields}
    errors = []
    
    for col_spec in contract.get("required_columns", []):
        col_name = col_spec["name"]
        expected_type = col_spec["type"]
        nullable = col_spec.get("nullable", True)
        
        if col_name not in schema_map:
            errors.append(f"Required column '{col_name}' is missing")
            continue
        
        field = next(f for f in df.schema.fields if f.name == col_name)
        if field.nullable != nullable:
            errors.append(
                f"Column '{col_name}' nullability mismatch: "
                f"expected {nullable}, got {field.nullable}"
            )
    
    return errors

# Usage
contract = load_contract("contract.yaml")
df = spark.read.format("delta").load("s3://data/orders/")

violations = validate_against_contract(df, contract)
if violations:
    for v in violations:
        print(f"Contract violation: {v}")
    raise ValueError("Schema does not meet contract requirements")

print("✓ Schema validation passed")

This contract-first approach shifts responsibility: producers know exactly which columns are safe to modify, and consumers have automated checks to catch surprises. It also serves as documentation for both teams and helps during onboarding.

Alert on unexpected changes

Even with contracts and good communication, unexpected schema changes happen. A junior engineer on the producer team merges a schema-breaking change, or a third-party data source changes its API. Your pipeline needs to detect and alert before silently producing wrong results.

We typically add schema inspection and alerting as an early step in the pipeline, before any transformations. This catches problems immediately and prevents corrupt data from flowing downstream.

import hashlib
from pyspark.sql.types import StructType
from datetime import datetime

def schema_hash(schema):
    """Compute a deterministic hash of a schema."""
    schema_str = schema.json()
    return hashlib.md5(schema_str.encode()).hexdigest()

def check_schema_changes(df, schema_file_path, alert_func=None):
    """
    Check if current schema matches last known schema.
    If changed, log and optionally alert.
    """
    current_hash = schema_hash(df.schema)
    
    try:
        with open(schema_file_path, "r") as f:
            previous_hash = f.read().strip()
    except FileNotFoundError:
        # First run, save schema
        with open(schema_file_path, "w") as f:
            f.write(current_hash)
        print(f"[{datetime.now()}] Schema recorded: {current_hash}")
        return {"status": "baseline", "previous": None, "current": current_hash}
    
    if current_hash == previous_hash:
        return {"status": "unchanged", "hash": current_hash}
    
    # Schema changed
    message = (
        f"⚠ Schema change detected!\n"
        f"Previous hash: {previous_hash}\n"
        f"Current hash: {current_hash}\n"
        f"Time: {datetime.now()}\n"
        f"Schema fields:\n"
    )
    for field in df.schema.fields:
        message += f"  - {field.name}: {field.dataType}\n"
    
    print(message)
    
    if alert_func:
        alert_func(message)
    
    # Update schema file only if you want to auto-accept
    # Otherwise, require manual review and update
    
    return {
        "status": "changed",
        "previous": previous_hash,
        "current": current_hash,
        "fields": [f.name for f in df.schema.fields]
    }

# Usage in a pipeline
def send_slack_alert(message):
    """Send alert to Slack webhook or CloudWatch."""
    import requests
    webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK"
    requests.post(webhook_url, json={"text": message})

df = spark.read.format("delta").load("s3://data/orders/")
result = check_schema_changes(df, "/tmp/last_schema.txt", alert_func=send_slack_alert)

if result["status"] == "changed":
    # Decide: halt, review manually, or proceed with logging
    print(f"New fields: {result['fields']}")
    # Option: fail the job to force manual review
    # raise ValueError("Unexpected schema change; manual review required")

This pattern captures the schema on first run, then watches for changes. When a change is detected, it logs the details and sends an alert (Slack, email, CloudWatch, etc.). The team can then decide whether to accept the change, adjust the contract, or roll back the producer change.

Store the schema hash in version control or an external metadata service, not locally, so it's accessible across job runs and workers.

Common mistakes

• Ignoring nullable columns: Adding a new column as non-nullable (nullable=false) will cause reads to fail if older data lacks that column. Always default to nullable=true for new fields.
• Over-relying on mergeSchema without validation: mergeSchema=true is convenient but silent. Missing schema contracts means you won't know when a breaking change occurs until data quality metrics fail downstream.
• No producer-consumer communication: Schema evolution works only if both sides know the rules. Document contracts explicitly and review them during code review.
• Alerting too late: Detecting schema changes after transformations or aggregations means bad data has already been produced. Validate schemas early, before processing.
• Type coercion without logging: Spark can often cast data between types, but a column changing from long to string silently may indicate a producer bug. Log and alert on type changes even if casting succeeds.

Key takeaways

• New columns should always be nullable; Spark's permissive mode handles them gracefully.
• Delta's mergeSchema=true automates additive schema changes but requires contracts and monitoring.
• Explicit contracts between producers and consumers prevent surprises and document expectations clearly.
• Schema validation at the pipeline entry point catches problems before they propagate downstream.
• Alert mechanisms (Slack, email, CloudWatch) keep teams informed of unexpected changes in real time.
• Store schema baselines in version control or metadata services for auditability and reproducibility.