Skip to content
PHP2Shell: The PHP Bug
Akshay Thakare

Always Use Explicit Schemas in PySpark

Schema inference is convenient in a notebook and dangerous in a pipeline. Here is how to define, enforce and evolve schemas deliberately.

PySparkBeginner2 min readUpdated 11 Oct 2026

Why inference hurts pipelines

spark.read.json(path) with inference has to scan the data to guess types, and the guess depends on whichever files happen to be present:

  • A column that is always an integer today becomes a string the day one record has "N/A"
  • Money inferred as double introduces floating-point rounding
  • An empty batch produces a schema with no columns, breaking downstream joins

Define the contract in code

Python.py
from pyspark.sql.types import (StructType, StructField, StringType,
                               IntegerType, DecimalType, TimestampType, BooleanType)
 
ORDER_ITEMS = StructType([
    StructField("order_id",        StringType(),       nullable=False),
    StructField("line_id",         StringType(),       nullable=False),
    StructField("product_id",      StringType(),       nullable=False),
    StructField("quantity",        IntegerType(),      nullable=False),
    StructField("unit_price",      DecimalType(18, 2), nullable=False),
    StructField("discount_amount", DecimalType(18, 2), nullable=False),
    StructField("updated_at",      TimestampType(),    nullable=False),
    StructField("is_deleted",      BooleanType(),      nullable=False),
])
 
df = (spark.read
        .schema(ORDER_ITEMS)
        .option("mode", "FAILFAST")      # stop on malformed records
        .json("s3://bucket/raw/.../dataset=order_items/"))

mode options: PERMISSIVE (default — nulls plus a corrupt-record column), DROPMALFORMED (silently drops — rarely what you want), FAILFAST (raise).

Schema-first read
Raw JSONuntrusted
Explicit schemaStructType
FAILFASTmalformed? stop
Validaterequired fields
Typed DataFramesafe to join

Validate, then quarantine

Python.py
required = ["order_id", "line_id", "product_id", "quantity"]
bad = df.filter(" OR ".join(f"{c} IS NULL" for c in required) + " OR quantity < 0")
good = df.subtract(bad)

Evolving schemas

Version the schema alongside the manifest (schema_version). New optional fields can be added compatibly; removed or retyped required fields should fail the batch until the contract is updated on purpose.

Finished this note?

Related notes

Share

LinkedInXWhatsApp