Schema evolution and enforcement in Apache Spark

Introduction

Schema evolution and enforcement help keep data organized and consistent when it changes over time.

When new columns are added to a dataset but old data still exists.

When you want to make sure data types stay the same to avoid errors.

When reading data from files that may have different versions of the schema.

When updating a table structure without losing existing data.

When you want to prevent unexpected changes in data format.

Syntax

Apache Spark

spark.read.option("mergeSchema", "true").parquet("path/to/data")

// To enforce schema
import org.apache.spark.sql.types._
val schema = new StructType().add("name", StringType).add("age", IntegerType)
spark.read.schema(schema).json("path/to/json")

Use mergeSchema=true to allow Spark to combine different schema versions.

Use .schema() to enforce a fixed schema when reading data.

Examples

This reads parquet files with different schemas and merges them into one.

Apache Spark

val df = spark.read.option("mergeSchema", "true").parquet("/data/events")
df.printSchema()

This enforces a fixed schema when reading JSON data to avoid surprises.

Apache Spark

import org.apache.spark.sql.types._

val fixedSchema = new StructType()
  .add("id", IntegerType)
  .add("name", StringType)
  .add("score", DoubleType)

val df = spark.read.schema(fixedSchema).json("/data/scores.json")
df.printSchema()

Sample Program

This example shows how to save data with one schema, add new data with an extra column, and then read both together by merging schemas.

Apache Spark

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._

val spark = SparkSession.builder.appName("SchemaEvolutionExample").master("local").getOrCreate()

// Create first DataFrame with schema v1
val dataV1 = Seq((1, "Alice"), (2, "Bob"))
val schemaV1 = new StructType().add("id", IntegerType).add("name", StringType)
val dfV1 = spark.createDataFrame(spark.sparkContext.parallelize(dataV1), schemaV1)

// Save as parquet
val path = "./temp_parquet"
dfV1.write.mode("overwrite").parquet(path)

// Create second DataFrame with schema v2 (new column 'age')
val dataV2 = Seq((3, "Charlie", 30), (4, "David", 40))
val schemaV2 = new StructType().add("id", IntegerType).add("name", StringType).add("age", IntegerType)
val dfV2 = spark.createDataFrame(spark.sparkContext.parallelize(dataV2), schemaV2)

// Append new data with new schema
dfV2.write.mode("append").parquet(path)

// Read with mergeSchema option to combine both schemas
val mergedDF = spark.read.option("mergeSchema", "true").parquet(path)
mergedDF.printSchema()
mergedDF.show()

spark.stop()

Important Notes

Schema evolution works well with file formats like Parquet and Delta Lake.

Enforcing schema helps catch errors early if data does not match expected types.

Always test schema changes on a small dataset before applying to big data.

Summary

Schema evolution lets you add or change columns without breaking old data.

Schema enforcement ensures data matches the expected format.

Use mergeSchema option to combine different schema versions when reading.

Practice

1. What does schema evolution in Apache Spark allow you to do?

easy

A. Add or change columns without breaking old data

B. Automatically delete old data files

C. Enforce strict data types on all columns

D. Compress data files for faster reading

Solution

1. Step 1: Understand schema evolution concept

   Schema evolution means you can update the schema by adding or changing columns without losing compatibility with existing data.

2. Step 2: Compare options with definition

   Only Add or change columns without breaking old data correctly describes this ability. Other options describe unrelated features.

3. Final Answer:

   Add or change columns without breaking old data -> Option A

4. Quick Check:

   Schema evolution = Add/change columns safely [OK]

Hint: Schema evolution means changing schema without breaking data [OK]

Common Mistakes:

• Confusing schema evolution with schema enforcement
• Thinking schema evolution deletes old data
• Assuming schema evolution compresses data

2. Which option correctly enables schema merging when reading Parquet files in Spark?

easy

A. spark.read.option("schemaEvolution", "enabled").parquet(path)

B. spark.read.option("enforceSchema", "true").parquet(path)

C. spark.read.option("merge", "false").parquet(path)

D. spark.read.option("mergeSchema", "true").parquet(path)

Solution

1. Step 1: Recall correct option name for schema merging

   The correct option to merge schemas is "mergeSchema" set to "true".

2. Step 2: Verify syntax correctness

   spark.read.option("mergeSchema", "true").parquet(path) uses the exact syntax spark.read.option("mergeSchema", "true").parquet(path), which is valid.

3. Final Answer:

   spark.read.option("mergeSchema", "true").parquet(path) -> Option D

4. Quick Check:

   Use mergeSchema=true to merge schemas [OK]

Hint: Use mergeSchema option set to true for schema merging [OK]

Common Mistakes:

• Using wrong option names like enforceSchema or schemaEvolution
• Setting mergeSchema to false disables merging
• Confusing option names with other Spark settings

3. Given two Parquet files with schemas:
file1: (id INT, name STRING)
file2: (id INT, name STRING, age INT)
What will be the schema of the DataFrame when reading both files with mergeSchema=true?

medium

A. (id INT, name STRING, age INT)

B. (id INT, name STRING)

C. (id INT, age INT)

D. (name STRING, age INT)

Solution

1. Step 1: Understand schema merging behavior

   When mergeSchema=true, Spark combines all columns from all files, including new columns.

2. Step 2: Combine columns from both schemas

   file1 has (id, name), file2 adds (age). Merged schema includes all three columns.

3. Final Answer:

   (id INT, name STRING, age INT) -> Option A

4. Quick Check:

   Merged schema includes all columns from files [OK]

Hint: Merged schema includes all columns from all files [OK]

Common Mistakes:

• Ignoring new columns in second file
• Assuming schema stays as first file only
• Dropping columns not in all files

4. You try to read Parquet files with different schemas but get an error about schema mismatch. What is the likely cause?

medium

A. The files are corrupted

B. You did not set mergeSchema option to true

C. You used the wrong file path

D. Spark does not support schema evolution

Solution

1. Step 1: Identify cause of schema mismatch error

   Schema mismatch errors happen when Spark tries to read files with different schemas without merging enabled.

2. Step 2: Check mergeSchema option usage

   Not setting mergeSchema=true causes Spark to expect identical schemas, leading to errors.

3. Final Answer:

   You did not set mergeSchema option to true -> Option B

4. Quick Check:

   Enable mergeSchema=true to avoid schema mismatch errors [OK]

Hint: Set mergeSchema=true to fix schema mismatch errors [OK]

Common Mistakes:

• Blaming file corruption without checking options
• Assuming Spark cannot evolve schemas
• Ignoring mergeSchema option

5. You have a Delta Lake table with schema (id INT, name STRING). You want to add a new column age INT and keep old data accessible. Which approach correctly applies schema evolution and enforcement?

hard

A. Overwrite the table schema manually without options

B. Delete old data and write new data with new schema

C. Use ALTER TABLE ADD COLUMN age INT and write new data with mergeSchema=true

D. Read old data without options and append new data with new schema

Solution

1. Step 1: Add new column using ALTER TABLE

   ALTER TABLE ADD COLUMN adds the new column to the Delta table schema safely.

2. Step 2: Write new data with mergeSchema=true

   Writing with mergeSchema=true allows Spark to merge new schema with old data without errors.

3. Final Answer:

   Use ALTER TABLE ADD COLUMN age INT and write new data with mergeSchema=true -> Option C

4. Quick Check:

   ALTER TABLE + mergeSchema=true enables safe schema evolution [OK]

Hint: Use ALTER TABLE and mergeSchema=true to evolve Delta schema safely [OK]

Common Mistakes:

• Overwriting schema without mergeSchema causes data loss
• Deleting old data loses history
• Appending without mergeSchema causes errors