# 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.

- Source: https://akshaythakare.com/notes/pyspark/explicit-schemas
- Topic: PySpark
- Level: beginner
- Reading time: 2 min
- Tags: pyspark, schemas, data-quality
- Updated: 2026-10-11

## 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
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).

> **Warning: nullable=False is not validation**
>
> Spark does not reliably enforce `nullable=False` on read. Validate required fields explicitly and route failures to a quarantine dataset with a reason.

**Schema-first read:** Raw JSON (untrusted) → Explicit schema (StructType) → FAILFAST (malformed? stop) → Validate (required fields) → Typed DataFrame (safe to join)

## Validate, then quarantine

```python
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.