Schema evolution and enforcement in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Output of schema enforcement with missing column

What will be the output when trying to read a DataFrame with a schema that requires a non-nullable column, but the data is missing that column?

Apache Spark

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

spark = SparkSession.builder.master('local').appName('SchemaTest').getOrCreate()

schema = StructType([
    StructField('name', StringType(), nullable=False),
    StructField('age', IntegerType(), nullable=False)
])

data = [('Alice',), ('Bob',)]
rdd = spark.sparkContext.parallelize(data)
df = spark.createDataFrame(rdd, ['name'])

try:
    df_with_schema = spark.createDataFrame(df.rdd, schema)
    df_with_schema.show()
except Exception as e:
    print(e)

A It raises an error because the 'age' column is missing and is non-nullable.

B It ignores the schema and shows only the 'name' column.

C It shows the DataFrame with 'age' column filled with nulls.

D It fills the missing 'age' column with zeros automatically.

💡 Hint

Spark pads missing columns with nulls, even if marked non-nullable.

When a schema has more columns than the data, Spark pads the missing columns with nulls. The non-nullable flag is an optimization hint and does not prevent nulls during DataFrame creation.

❓ data_output

intermediate

Result of schema evolution by adding a new nullable column

Given a DataFrame with schema having columns 'id' and 'name', what will be the result of reading new data that adds a nullable column 'email' without changing the original schema?

Apache Spark

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

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

original_schema = StructType([
    StructField('id', IntegerType(), nullable=False),
    StructField('name', StringType(), nullable=True)
])

# Original data
data1 = [(1, 'Alice'), (2, 'Bob')]
df1 = spark.createDataFrame(data1, schema=original_schema)
df1.show()

# New data with extra nullable column 'email'
data2 = [(3, 'Charlie', '[e-mail]'), (4, 'David', None)]
df2 = spark.createDataFrame(data2, ['id', 'name', 'email'])

# Read new data with original schema
try:
    df2_with_original_schema = spark.createDataFrame(df2.rdd, original_schema)
    df2_with_original_schema.show()
except Exception as e:
    print(e)

A It raises an error because the new 'email' column is not in the original schema.

B It reads only the 'id' and 'name' columns, ignoring 'email' without error.

C It reads all three columns including 'email' despite the original schema.

D It fills the missing 'email' column with nulls and shows all columns.

💡 Hint

Think about how Spark handles extra columns when applying a schema with fewer columns.

When applying a schema with fewer columns than the data, Spark reads only the columns defined in the schema and ignores extra columns without error.

🔧 Debug

advanced

Identify the cause of schema mismatch error

Why does the following code raise a schema mismatch error when reading JSON data with a schema expecting an integer, but the data has a string?

Apache Spark

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

spark = SparkSession.builder.master('local').appName('SchemaMismatch').getOrCreate()

schema = StructType([
    StructField('id', IntegerType(), nullable=False)
])

json_data = '{"id": "abc"}'
df = spark.read.option("mode", "FAILFAST").schema(schema).json(spark.sparkContext.parallelize([json_data]))
df.show()

A The error occurs because Spark cannot read JSON from an RDD.

B The error occurs because the schema is missing a nullable field.

C The error occurs because the JSON data is not properly formatted.

D The error occurs because the JSON string 'abc' cannot be cast to integer as required by the schema.

💡 Hint

Check the data type of the value in JSON and the expected schema type.

Spark tries to cast the JSON string 'abc' to integer but fails, causing a schema mismatch error.

❓ visualization

advanced

Visualizing schema evolution with added and removed columns

You have two DataFrames: one with columns ['id', 'name', 'age'] and another with ['id', 'name', 'email']. You want to visualize the schema differences. Which visualization best shows added and removed columns clearly?

A A Venn diagram showing overlap and unique columns between the two schemas.

B A bar chart showing counts of rows in each DataFrame.

C A scatter plot of 'age' vs 'email' columns.

D A pie chart showing percentage of null values in each column.

💡 Hint

Think about how to show common and different columns between two sets.

A Venn diagram clearly shows which columns are shared and which are unique to each schema, making it ideal for visualizing schema evolution.

🚀 Application

expert

Handling schema evolution in streaming data ingestion

You have a streaming pipeline ingesting JSON data with evolving schema: new nullable columns are added over time. How should you configure Spark Structured Streaming to handle schema evolution without failing on new columns?

A Use a schema with all possible columns marked nullable and enable schema inference to allow new columns without failure.

B Manually update the schema to include new columns and restart the streaming query each time schema changes.

C Set 'spark.sql.streaming.schemaInference' to true and use a schema with only the original columns.

D Disable schema enforcement so Spark accepts any schema changes automatically.

💡 Hint

Consider how to allow new columns to appear without breaking the stream.

Using a schema that includes all expected columns as nullable and enabling schema inference allows Spark to handle new columns gracefully during streaming ingestion.

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