16. Schema Evolution in PySpark

DfSchema Evolution

Schema evolution is the ability to modify a table's schema over time while maintaining backward and forward compatibility with existing data. It supports adding, removing, renaming, and modifying columns without rewriting historical data.

DfSchema Compatibility

Schema compatibility defines whether a new schema can read data written with an old schema. Types: backward-compatible (new reads old), forward-compatible (old reads new), full-compatible (both directions).

Compatible = \\begin{cases} \\text{Backward} & \\text{if } S_{new} \\text{ can read } S_{old} \\\\ \\text{Forward} & \\text{if } S_{old} \\text{ can read } S_{new} \\\\ \\text{Full} & \\text{if both directions work} \\end{cases}

Schema Evolution Cost

C_{evolve} = N_{files} \\times (C_{read} + C_{transform} + C_{write})

Here,

$C_{evolve}$ =Total cost of schema evolution operation
$N_{files}$ =Number of Parquet files affected
$C_{read}$ =Cost of reading existing files with old schema
$C_{transform}$ =Cost of applying schema transformation
$C_{write}$ =Cost of rewriting files with new schema

Delta Lake supports schema evolution via mergeSchema option. When enabled, new columns are added with null values for existing rows. Column renames and type widening are also supported.

Use overwriteSchema instead of mergeSchema when you want to completely replace the schema. mergeSchema is additive only — it cannot remove or rename columns without rewriting data.

ThParquet Schema Resolution

Theorem: Parquet stores schema as metadata in each file. When reading with a different schema, Parquet resolves by column name — missing columns return null, extra columns are ignored, and type mismatches cause errors unless compatible conversions are defined.

Schema evolution modifies table schema over time without rewriting historical data
Compatibility: backward (new reads old), forward (old reads new), full (both)
Delta Lake: mergeSchema (additive), overwriteSchema (replace)
Parquet: column-name-based resolution; each file stores its own schema

🏗️ Schema Evolution Architecture

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────┐
│                    SCHEMA EVOLUTION ARCHITECTURE                         │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    VERSION 1 (Initial Schema)                    │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Schema V1                                              │    │   │
│  │  │  ┌──────┬──────┬──────┬──────┐                         │    │   │
│  │  │  │ id   │ name │ email│ age  │                         │    │   │
│  │  │  │ INT  │ STR  │ STR  │ INT  │                         │    │   │
│  │  │  └──────┴──────┴──────┴──────┘                         │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                              │                                          │
│                              ▼                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    VERSION 2 (Schema Change)                     │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Schema V2 (Added column)                               │    │   │
│  │  │  ┌──────┬──────┬──────┬──────┬──────┐                  │    │   │
│  │  │  │ id   │ name │ email│ age  │salary│                  │    │   │
│  │  │  │ INT  │ STR  │ STR  │ INT  │DOUBLE│                  │    │   │
│  │  │  └──────┴──────┴──────┴──────┴──────┘                  │    │   │
│  │  │                                                         │    │   │
│  │  │  Backward Compatible: V1 readers can read V2 data      │    │   │
│  │  │  Forward Compatible: V2 readers can read V1 data       │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                              │                                          │
│                              ▼                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    VERSION 3 (Breaking Change)                   │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Schema V3 (Renamed column)                             │    │   │
│  │  │  ┌──────┬──────┬─────────┬──────┬──────┐               │    │   │
│  │  │  │ id   │ name │email_id │ age  │salary│               │    │   │
│  │  │  │ INT  │ STR  │ STR     │ INT  │DOUBLE│               │    │   │
│  │  │  └──────┴──────┴─────────┴──────┴──────┘               │    │   │
│  │  │                                                         │    │   │
│  │  │  Breaking Change: V1 readers cannot read V3 data       │    │   │
│  │  │  Requires migration or compatibility layer              │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    COMPATIBILITY MATRIX                          │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Change Type      │ Backward │ Forward │ Impact         │    │   │
│  │  │───────────────────┼──────────┼─────────┼────────────────│    │   │
│  │  │  Add column       │ Yes      │ Yes     │ Low            │    │   │
│  │  │  Remove column    │ Yes      │ No      │ Medium         │    │   │
│  │  │  Rename column    │ No       │ No      │ High           │    │   │
│  │  │  Change type      │ Depends  │ Depends │ High           │    │   │
│  │  │  Add nested field │ Yes      │ Yes     │ Low            │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────┘

🔄 Schema Compatibility Modes

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────┐
│                    SCHEMA COMPATIBILITY MODES                            │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    BACKWARD COMPATIBILITY                        │   │
│  │                                                                   │   │
│  │  Schema V1 (Old) ─────────────────────────────────────────▶    │   │
│  │  ┌──────┬──────┬──────┬──────┐                                 │   │
│  │  │ id   │ name │ email│ age  │                                 │   │
│  │  └──────┴──────┴──────┴──────┘                                 │   │
│  │                                                                   │   │
│  │  Schema V2 (New) ─────────────────────────────────────────▶    │   │
│  │  ┌──────┬──────┬──────┬──────┬──────┐                         │   │
│  │  │ id   │ name │ email│ age  │salary│                         │   │
│  │  └──────┴──────┴──────┴──────┴──────┘                         │   │
│  │                                                                   │   │
│  │  V1 readers can read V2 data (new column has default value)    │   │
│  │  V2 readers can read V1 data (missing column uses null)        │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    FORWARD COMPATIBILITY                         │   │
│  │                                                                   │   │
│  │  Schema V1 (New) ─────────────────────────────────────────▶    │   │
│  │  ┌──────┬──────┬──────┐                                        │   │
│  │  │ id   │ name │ email│                                        │   │
│  │  └──────┴──────┴──────┘                                        │   │
│  │                                                                   │   │
│  │  Schema V2 (Old) ─────────────────────────────────────────▶    │   │
│  │  ┌──────┬──────┬──────┬──────┬──────┐                         │   │
│  │  │ id   │ name │ email│ age  │salary│                         │   │
│  │  └──────┴──────┴──────┴──────┴──────┘                         │   │
│  │                                                                   │   │
│  │  V2 readers can read V1 data (ignores extra columns)           │   │
│  │  V1 readers cannot read V2 data (unknown columns)              │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    FULL COMPATIBILITY                            │   │
│  │                                                                   │   │
│  │  Both backward AND forward compatible                           │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  • Add nullable columns (with defaults)                 │    │   │
│  │  │  • Remove columns with defaults                         │    │   │
│  │  │  • Never rename columns                                 │    │   │
│  │  │  • Never change column types (except widening)          │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  │                                                                   │   │
│  │  Ideal for: Multi-version readers, data lakes                  │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    NONE COMPATIBILITY                            │   │
│  │                                                                   │   │
│  │  No compatibility guarantees                                    │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  • Any schema change allowed                            │    │   │
│  │  │  • Readers must match exact schema                      │    │   │
│  │  │  • Use only when all readers are updated simultaneously │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  │                                                                   │   │
│  │  Risk: Readers may fail if schema doesn't match                │   │
│  └─────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────┘

📊 Schema Evolution Strategies

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────┐
│                    SCHEMA EVOLUTION STRATEGIES                           │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    STRATEGY 1: ADDITIVE EVOLUTION                │   │
│  │                                                                   │   │
│  │  Only add new columns (never remove or rename)                  │   │
│  │                                                                   │   │
│  │  V1: [id, name, email]                                          │   │
│  │  V2: [id, name, email, salary]  ← Added salary                  │   │
│  │  V3: [id, name, email, salary, department]  ← Added department  │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Pros:                                                  │    │   │
│  │  │  • Simple to implement                                  │    │   │
│  │  │  • Backward compatible by default                       │    │   │
│  │  │  • No data migration needed                             │    │   │
│  │  │                                                         │    │   │
│  │  │  Cons:                                                  │    │   │
│  │  │  • Schema grows indefinitely                            │    │   │
│  │  │  • May contain unused columns                           │    │   │
│  │  │  • Naming conflicts possible                            │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    STRATEGY 2: VERSIONED SCHEMA                  │   │
│  │                                                                   │   │
│  │  Maintain multiple schema versions with mapping                 │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Schema Registry                                        │    │   │
│  │  │  ┌──────┬─────────────────────────────────────────┐    │    │   │
│  │  │  │ V1   │ {id: int, name: string, email: string}  │    │    │   │
│  │  │  │ V2   │ {id: int, name: string, email: string,  │    │    │   │
│  │  │  │      │  salary: double}                         │    │    │   │
│  │  │  │ V3   │ {id: int, name: string, email: string,  │    │    │   │
│  │  │  │      │  salary: double, dept: string}           │    │    │   │
│  │  │  └──────┴─────────────────────────────────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  │                                                                   │   │
│  │  Mapping:                                                       │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  V1 → V2: Add salary with default 0.0                  │    │   │
│  │  │  V2 → V3: Add dept with default "Unknown"              │    │   │
│  │  │  V3 → V2: Drop dept column                             │    │   │
│  │  │  V2 → V1: Drop salary column                           │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    STRATEGY 3: SCHEMA-on-READ                    │   │
│  │                                                                   │   │
│  │  Apply schema at read time, not write time                      │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Raw Data (schema-free)                                 │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  {"id": 1, "name": "Alice", "extra": "data"}    │    │    │   │
│  │  │  │  {"id": 2, "name": "Bob", "salary": 50000}     │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  │                                                         │    │   │
│  │  │  Reader 1 (Schema V1): [id, name]                       │    │   │
│  │  │  Reader 2 (Schema V2): [id, name, salary]               │    │   │
│  │  │  Reader 3 (Schema V3): [id, name, email, age]           │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  │                                                                   │   │
│  │  Best for: Data lakes, schema-flexible systems                  │   │
│  └─────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────┘

📚 Detailed Explanation

Schema evolution is a critical concept in data management that enables systems to handle changes in data structure over time without breaking existing applications or losing data. In PySpark, schema evolution is particularly important for data lakes where data is stored for long periods and accessed by multiple consumers with different schema requirements.

The fundamental challenge of schema evolution is maintaining compatibility between different versions of data. Backward compatibility ensures that older readers can read newer data, while forward compatibility ensures that newer readers can read older data. Understanding these compatibility modes is essential for designing robust data systems.

Additive evolution is the simplest and most common strategy. By only adding new columns and never removing or renaming existing ones, systems maintain backward compatibility automatically. New columns should have default values to handle cases where older readers encounter newer data. This strategy works well for many use cases but can lead to schema bloat over time.

Versioned schema management provides more flexibility by maintaining explicit mappings between schema versions. This approach allows for more complex changes like column renames and type changes, but requires a schema registry or similar mechanism to manage the mappings. It's ideal for systems that need to support multiple schema versions simultaneously.

Schema-on-read applies the schema at read time rather than write time. This approach stores data in a schema-free format (like JSON or Avro) and allows readers to apply their own schema when reading. It provides maximum flexibility but can impact performance and data quality since validation happens at read time.

The choice of format significantly impacts schema evolution capabilities. Parquet and Avro both support schema evolution but have different characteristics. Parquet is columnar and optimized for analytical queries, while Avro is row-based and better for schema evolution. Both support embedding schema in the data file, which helps with self-describing data.

Type evolution is a complex aspect of schema evolution. Widening changes (like int to long) are generally safe and backward compatible. Narrowing changes (like long to int) can cause data loss and are not backward compatible. Type changes that change semantics (like string to date) require careful handling and often need explicit mapping logic.

Best practices for schema evolution include: using additive evolution when possible, maintaining a schema registry for complex changes, testing compatibility thoroughly, documenting schema changes, and implementing schema validation at both write and read times. It's also important to monitor schema usage and clean up unused columns periodically.

Advanced techniques include schema inference (automatically detecting schema from data), schema classification (categorizing schema changes by impact), and schema evolution automation (automatically applying compatible changes). These techniques help manage schema evolution at scale.

📊 Key Concepts Table

Concept	Description	Use Case
Backward Compatibility	Old readers can read new data	Multi-version support
Forward Compatibility	New readers can read old data	Legacy system support
Full Compatibility	Both backward and forward	Data lakes, shared datasets
Schema Registry	Centralized schema management	Complex schema evolution
Schema-on-Read	Apply schema at read time	Flexible data storage
Type Evolution	Changing column data types	Data model changes

💻 Code Examples

Basic Schema Evolution with Parquet

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder \
    .appName("SchemaEvolution") \
    .config("spark.sql.parquet.mergeSchema", "true") \
    .getOrCreate()

# Write initial data (Schema V1)
df_v1 = spark.createDataFrame([
    (1, "Alice", "alice@example.com", 25),
    (2, "Bob", "bob@example.com", 30)
], ["id", "name", "email", "age"])

df_v1.write.mode("overwrite").parquet("/path/to/data")

# Read data (should work with Schema V1)
df_read_v1 = spark.read.parquet("/path/to/data")
df_read_v1.show()

# Write new data with additional column (Schema V2)
df_v2 = spark.createDataFrame([
    (3, "Charlie", "charlie@example.com", 35, 75000.0),
    (4, "Diana", "diana@example.com", 28, 85000.0)
], ["id", "name", "email", "age", "salary"])

# Append with schema evolution
df_v2.write.mode("append").option("mergeSchema", "true").parquet("/path/to/data")

# Read all data (should include both schemas)
df_read_all = spark.read.parquet("/path/to/data")
df_read_all.show()

# Verify schema evolution
print("Schema after evolution:")
df_read_all.printSchema()

Schema Evolution with Delta Lake

from pyspark.sql import SparkSession
from delta.tables import DeltaTable
from pyspark.sql.functions import *

spark = SparkSession.builder \
    .appName("DeltaSchemaEvolution") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Write initial data (Schema V1)
df_v1 = spark.createDataFrame([
    (1, "Alice", "alice@example.com", 25),
    (2, "Bob", "bob@example.com", 30)
], ["id", "name", "email", "age"])

df_v1.write.format("delta").mode("overwrite").save("/path/to/delta/data")

# Read data
spark.read.format("delta").load("/path/to/delta/data").show()

# Write new data with additional column (Schema V2)
df_v2 = spark.createDataFrame([
    (3, "Charlie", "charlie@example.com", 35, 75000.0),
    (4, "Diana", "diana@example.com", 28, 85000.0)
], ["id", "name", "email", "age", "salary"])

# Append with schema evolution
df_v2.write.format("delta").mode("append").option("mergeSchema", "true").save("/path/to/delta/data")

# Read all data
spark.read.format("delta").load("/path/to/delta/data").show()

# Verify schema evolution
print("Delta schema after evolution:")
spark.read.format("delta").load("/path/to/delta/data").printSchema()

Schema Evolution with Avro

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder \
    .appName("AvroSchemaEvolution") \
    .config("spark.sql.avro.compression.codec", "snappy") \
    .getOrCreate()

# Define Schema V1
schema_v1 = """
{
    "type": "record",
    "name": "User",
    "fields": [
        {"name": "id", "type": "int"},
        {"name": "name", "type": "string"},
        {"name": "email", "type": "string"},
        {"name": "age", "type": "int"}
    ]
}
"""

# Write initial data with Schema V1
df_v1 = spark.createDataFrame([
    (1, "Alice", "alice@example.com", 25),
    (2, "Bob", "bob@example.com", 30)
], ["id", "name", "email", "age"])

df_v1.write.mode("overwrite").format("avro").option("avroSchema", schema_v1).save("/path/to/avro/data")

# Define Schema V2 (added salary column)
schema_v2 = """
{
    "type": "record",
    "name": "User",
    "fields": [
        {"name": "id", "type": "int"},
        {"name": "name", "type": "string"},
        {"name": "email", "type": "string"},
        {"name": "age", "type": "int"},
        {"name": "salary", "type": ["null", "double"], "default": null}
    ]
}
"""

# Write new data with Schema V2
df_v2 = spark.createDataFrame([
    (3, "Charlie", "charlie@example.com", 35, 75000.0),
    (4, "Diana", "diana@example.com", 28, 85000.0)
], ["id", "name", "email", "age", "salary"])

df_v2.write.mode("append").format("avro").option("avroSchema", schema_v2).save("/path/to/avro/data")

# Read all data with Schema V2
spark.read.format("avro").load("/path/to/avro/data").show()

# Verify schema evolution
print("Avro schema after evolution:")
spark.read.format("avro").load("/path/to/avro/data").printSchema()

Schema Evolution with Type Changes

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("TypeEvolution") \
    .config("spark.sql.parquet.mergeSchema", "true") \
    .getOrCreate()

# Write initial data with IntegerType
df_v1 = spark.createDataFrame([
    (1, "Alice", 25),
    (2, "Bob", 30)
], ["id", "name", "age"])

df_v1.write.mode("overwrite").parquet("/path/to/type_evolution/data")

# Read and verify initial schema
print("Initial schema:")
spark.read.parquet("/path/to/type_evolution/data").printSchema()

# Write new data with LongType (widening change)
df_v2 = spark.createDataFrame([
    (3, "Charlie", 35),
    (4, "Diana", 28)
], ["id", "name", "age"])

# Cast age to LongType for widening
df_v2 = df_v2.withColumn("age", col("age").cast(LongType()))

# Append with schema evolution
df_v2.write.mode("append").option("mergeSchema", "true").parquet("/path/to/type_evolution/data")

# Read all data
print("Schema after type evolution:")
spark.read.parquet("/path/to/type_evolution/data").printSchema()

# Show data
spark.read.parquet("/path/to/type_evolution/data").show()

📈 Performance Metrics

Metric	Impact	Mitigation
Schema Merge Time	Increases with schema complexity	Use schema registry, limit changes
File Size	May increase with schema evolution	Regular compaction, optimize file sizes
Read Performance	May degrade with complex schemas	Use columnar formats, projection pushdown
Write Performance	Minimal impact for additive changes	Batch writes, use appropriate formats
Storage Overhead	Increases with schema versions	Clean up old versions, use compression

🏆 Best Practices

Use additive evolution when possible - Only add columns, never remove or rename
Maintain a schema registry - Track schema versions and compatibility
Use appropriate formats - Parquet for analytics, Avro for schema evolution
Test compatibility thoroughly - Validate backward and forward compatibility
Document schema changes - Maintain clear documentation of evolution
Implement schema validation - Validate data against schema at write time
Monitor schema usage - Track which columns are actually used
Clean up unused columns - Periodically remove columns that are no longer needed
Use default values - Provide defaults for new columns to maintain compatibility
Plan for type evolution - Consider future type changes when designing schemas

🔗 Related Topics

15-data-quality.mdx: Schema validation and data quality
14-merge-upsert.mdx: Schema evolution during merge operations
20-monitoring-metrics.mdx: Monitoring schema changes
17-cluster-management.mdx: Cluster configuration for schema operations

16. Schema Evolution in PySpark

16. Schema Evolution in PySpark

DfSchema Evolution

DfSchema Compatibility

Schema Evolution Cost

ThParquet Schema Resolution

🏗️ Schema Evolution Architecture

🔄 Schema Compatibility Modes

📊 Schema Evolution Strategies

📚 Detailed Explanation

📊 Key Concepts Table

💻 Code Examples

Basic Schema Evolution with Parquet

Schema Evolution with Delta Lake

Schema Evolution with Avro

Schema Evolution with Type Changes

📈 Performance Metrics

🏆 Best Practices

🔗 Related Topics

See Also

Need Expert PySpark Help?