Schema evolution and enforcement in Apache Spark - Mini Project: Build & Apply

Schema Evolution and Enforcement in Apache Spark

📖 Scenario: You work at a company that collects user data daily. The data format changes slightly over time, adding new fields. You need to handle these changes safely using schema evolution and enforcement in Apache Spark.

🎯 Goal: Build a Spark program that reads initial user data with a defined schema, updates the schema to include a new field, and enforces the schema when reading new data files.

📋 What You'll Learn

Create an initial Spark schema for user data with fields: user_id (integer), name (string), and age (integer).

Create a DataFrame with initial user data using the schema.

Define an updated schema that adds a new field email (string) to the user data.

Read new user data with the updated schema enforcing the schema during read.

Show the final DataFrame with all fields including the new email field.

💡 Why This Matters

🌍 Real World

In real companies, data formats change over time. Schema evolution helps keep data pipelines working smoothly without errors.

💼 Career

Data engineers and data scientists use schema enforcement and evolution to manage changing data structures in big data systems like Apache Spark.

1

Create initial user data with schema

Create a Spark schema called initial_schema with fields user_id (IntegerType), name (StringType), and age (IntegerType). Then create a DataFrame called df_initial with data [(1, 'Alice', 30), (2, 'Bob', 25)] using initial_schema.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

spark = SparkSession.builder.appName('SchemaEvolution').getOrCreate()

# Create initial_schema with user_id, name, age
# Create df_initial with data [(1, 'Alice', 30), (2, 'Bob', 25)] using initial_schema
# Your code here

Hint

Use StructType and StructField to define the schema. Then use spark.createDataFrame with the data and schema.

2

Define updated schema with new field

Create a new schema called updated_schema that adds a new field email (StringType) to the existing fields user_id, name, and age.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

spark = SparkSession.builder.appName('SchemaEvolution').getOrCreate()

initial_schema = StructType([
    StructField('user_id', IntegerType(), True),
    StructField('name', StringType(), True),
    StructField('age', IntegerType(), True)
])

data_initial = [(1, 'Alice', 30), (2, 'Bob', 25)]

df_initial = spark.createDataFrame(data_initial, schema=initial_schema)

# Create updated_schema adding email field
# Your code here

Hint

Copy the initial schema fields and add a new StructField for email with StringType().

3

Read new user data with updated schema enforcing schema

Create a list called data_new with tuples (3, 'Charlie', 28, '[e-mail]') and (4, 'Diana', 22, '[e-mail]'). Then create a DataFrame called df_new by reading data_new using updated_schema and enforcing the schema.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

spark = SparkSession.builder.appName('SchemaEvolution').getOrCreate()

initial_schema = StructType([
    StructField('user_id', IntegerType(), True),
    StructField('name', StringType(), True),
    StructField('age', IntegerType(), True)
])

data_initial = [(1, 'Alice', 30), (2, 'Bob', 25)]

df_initial = spark.createDataFrame(data_initial, schema=initial_schema)

updated_schema = StructType([
    StructField('user_id', IntegerType(), True),
    StructField('name', StringType(), True),
    StructField('age', IntegerType(), True),
    StructField('email', StringType(), True)
])

# Create data_new with new user data including email
# Create df_new DataFrame using updated_schema
# Your code here

Hint

Create a list of tuples with the new data including the email field. Then use spark.createDataFrame with updated_schema.

4

Show final DataFrame with schema enforcement

Use df_new.show() to display the new user data with the enforced schema including the email field.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

spark = SparkSession.builder.appName('SchemaEvolution').getOrCreate()

initial_schema = StructType([
    StructField('user_id', IntegerType(), True),
    StructField('name', StringType(), True),
    StructField('age', IntegerType(), True)
])

data_initial = [(1, 'Alice', 30), (2, 'Bob', 25)]

df_initial = spark.createDataFrame(data_initial, schema=initial_schema)

updated_schema = StructType([
    StructField('user_id', IntegerType(), True),
    StructField('name', StringType(), True),
    StructField('age', IntegerType(), True),
    StructField('email', StringType(), True)
])

data_new = [
    (3, 'Charlie', 28, '[e-mail]'),
    (4, 'Diana', 22, '[e-mail]')
]

df_new = spark.createDataFrame(data_new, schema=updated_schema)

# Show df_new
# Your code here

Hint

Use df_new.show() to display the DataFrame with the new schema.

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