Schema validation in Apache Spark - Step-by-Step Execution

Concept Flow - Schema validation

Define Schema

↓

Load Data

↓

Apply Schema to Data

↓

Validate Data Against Schema

↓

If Valid

→Process Data

↓

If Invalid

→Raise Error or Handle

↓

End

The flow starts by defining a schema, then loading data, applying the schema, validating the data, and either processing valid data or handling errors for invalid data.

Execution Sample

Apache Spark

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

spark = SparkSession.builder.getOrCreate()

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

data = [('Alice', 30), ('Bob', 'not_a_number'), ('Charlie', 25)]

# Create DataFrame with schema
try:
    df = spark.createDataFrame(data, schema=schema)
    df.show()
except Exception as e:
    print(f"Schema validation error: {e}")

This code defines a schema for name and age, tries to create a DataFrame with data where one age is invalid, and shows the DataFrame or error.

Execution Table

Step	Action	Data Sample	Schema Check	Result
1	Define schema with fields 'name' (string) and 'age' (integer)	N/A	N/A	Schema ready
2	Load data [('Alice', 30), ('Bob', 'not_a_number'), ('Charlie', 25)]	('Alice', 30)	N/A	Data loaded
3	Create DataFrame applying schema	('Alice', 30)	name: string, age: integer	Row accepted
4	Validate row ('Bob', 'not_a_number')	('Bob', 'not_a_number')	age should be integer	Validation error raised
5	Catch exception and print error	N/A	N/A	Schema validation error printed
6	End process	N/A	N/A	Process stopped due to error

💡 Data contains invalid type for 'age' in second row, causing schema validation error and stopping process.

Variable Tracker

Variable	Start	After Step 2	After Step 3	After Step 4	Final
schema	None	Defined StructType with name and age fields	Same	Same	Same
data	None	[('Alice', 30), ('Bob', 'not_a_number'), ('Charlie', 25)]	Same	Same	Same
df	None	None	Created DataFrame with first row accepted	Error on second row, df not fully created	None due to error

Key Moments - 2 Insights

Why does the DataFrame creation fail when 'age' is 'not_a_number'?

Because the schema expects 'age' to be an integer, but the value 'not_a_number' is a string, causing a type mismatch error as shown in execution_table step 4.

Can Spark partially create a DataFrame if some rows fail schema validation?

No, Spark raises an error and stops creating the DataFrame when it encounters invalid data, as seen in step 5 where the exception is caught.

Visual Quiz - 3 Questions

Look at the execution table, what happens at step 4 when validating the row ('Bob', 'not_a_number')?

A Validation error is raised due to type mismatch

B Row is accepted without error

C Row is skipped silently

D Schema is changed to accept string

💡 Hint

Refer to execution_table row 4 under 'Result' column

At which step is the schema first defined?

A Step 3

B Step 1

C Step 2

D Step 4

💡 Hint

Check execution_table row 1 under 'Action'

If the invalid age value 'not_a_number' was changed to 40, what would happen at step 4?

A DataFrame creation would fail at step 3

B Validation error still occurs

C Row would be accepted successfully

D Schema would need to be redefined

💡 Hint

Look at variable_tracker for 'df' changes and execution_table step 4 result

Concept Snapshot

Schema validation in Spark:
- Define a StructType schema with field names and types
- Load data as list or RDD
- Create DataFrame applying the schema
- Spark checks each row matches schema types
- If mismatch, Spark raises error and stops
- Valid data proceeds for processing

Full Transcript

Schema validation in Apache Spark involves defining a schema that specifies the expected data types for each column. When loading data, Spark applies this schema to ensure each row matches the expected types. If a row contains a value that does not match the schema, such as a string where an integer is expected, Spark raises a validation error and stops creating the DataFrame. This prevents invalid data from entering the processing pipeline. The example code shows defining a schema with 'name' as string and 'age' as integer, then attempting to create a DataFrame with one invalid age value. The execution trace shows the error occurs at the invalid row, and the process stops with an error message.

Practice

1. What is the main purpose of schema validation in Apache Spark?

easy

A. To check if the data matches the expected format before processing

B. To speed up the Spark job execution

C. To automatically clean the data

D. To visualize data distributions

Solution

1. Step 1: Understand schema validation role

   Schema validation ensures data matches the expected structure and types before analysis.

2. Step 2: Identify the main goal

   This helps catch errors early and prevents processing wrong data formats.

3. Final Answer:

   To check if the data matches the expected format before processing -> Option A

4. Quick Check:

   Schema validation = format check [OK]

Hint: Schema validation means checking data format first [OK]

Common Mistakes:

• Thinking schema speeds up execution directly
• Assuming schema cleans data automatically
• Confusing schema validation with visualization

2. Which of the following is the correct way to define a schema with two fields: 'name' as string and 'age' as integer in Spark?

easy

A. Schema(['name': 'string', 'age': 'int'])

B. StructType('name' StringType, 'age' IntegerType)

C. StructType([StructField('name', StringType()), StructField('age', IntegerType())])

D. StructField('name', 'string'), StructField('age', 'int')

Solution

1. Step 1: Recall correct schema syntax

   In Spark, schema is defined using StructType with a list of StructField objects specifying name and type.

2. Step 2: Match syntax to options

   StructType([StructField('name', StringType()), StructField('age', IntegerType())]) correctly uses StructType with StructField and proper types.

3. Final Answer:

   StructType([StructField('name', StringType()), StructField('age', IntegerType())]) -> Option C

4. Quick Check:

   Schema = StructType + StructField [OK]

Hint: Use StructType with list of StructField for schema [OK]

Common Mistakes:

• Omitting StructType wrapper
• Using wrong syntax without brackets
• Confusing schema with dictionary or list

3. Given this code snippet:

from pyspark.sql.types import StructType, StructField, StringType, IntegerType
schema = StructType([
  StructField('name', StringType(), True),
  StructField('age', IntegerType(), True)
])
data = [('Alice', 30), ('Bob', 'thirty')]
df = spark.createDataFrame(data, schema)
df.show()

What will happen when this code runs?

medium

A. It will show both rows with 'age' as string for Bob

B. It will raise an error due to type mismatch in 'age' for Bob

C. It will convert 'thirty' to null silently

D. It will ignore the second row and show only the first

Solution

1. Step 1: Analyze schema and data types

   The schema expects 'age' as IntegerType, but Bob's age is a string 'thirty'.

2. Step 2: Understand Spark behavior on type mismatch

   Spark will raise a runtime error when data does not match schema types during DataFrame creation.

3. Final Answer:

   It will raise an error due to type mismatch in 'age' for Bob -> Option B

4. Quick Check:

   Type mismatch causes error [OK]

Hint: Mismatch in data type causes error, not silent conversion [OK]

Common Mistakes:

• Assuming Spark converts wrong types silently
• Thinking Spark ignores bad rows by default
• Believing all rows display regardless of errors

4. You wrote this schema and code:

schema = StructType([
  StructField('id', IntegerType()),
  StructField('value', StringType())
])
data = [(1, 'a'), (2, 'b')]
df = spark.createDataFrame(data, schema)
df.printSchema()

But you get an error: TypeError: StructField() missing 1 required positional argument: 'nullable'. What is the fix?

medium

A. Add the 'nullable' argument to each StructField, e.g. StructField('id', IntegerType(), True)

B. Remove the schema and let Spark infer it

C. Change StructType to SchemaType

D. Use a list of tuples instead of StructField

Solution

1. Step 1: Understand StructField parameters

   StructField requires three arguments: name, data type, and nullable (True/False).

2. Step 2: Fix the code by adding nullable

   Add the third argument 'nullable=True' to each StructField to fix the error.

3. Final Answer:

   Add the 'nullable' argument to each StructField, e.g. StructField('id', IntegerType(), True) -> Option A

4. Quick Check:

   StructField needs nullable argument [OK]

Hint: Always include nullable in StructField [OK]

Common Mistakes:

• Forgetting nullable argument
• Trying to rename StructType
• Ignoring error and running code

5. You have a CSV file with columns: 'user', 'score', and 'date'. You want to read it with Spark and ensure 'score' is an integer and 'date' is a date type. Which schema definition and read method is correct to validate this?

hard

A. schema = StructType([StructField('user', StringType(), False), StructField('score', IntegerType(), False), StructField('date', DateType(), False)])
df = spark.read.option('header', 'true').csv('file.csv', schema=schema)

B. schema = StructType([StructField('user', StringType()), StructField('score', StringType()), StructField('date', StringType())])
df = spark.read.csv('file.csv', schema=schema)

C. df = spark.read.csv('file.csv', inferSchema=True, header=True)

D. schema = StructType([StructField('user', StringType(), True), StructField('score', IntegerType(), True), StructField('date', DateType(), True)])
df = spark.read.csv('file.csv', schema=schema, header=True)

Solution

1. Step 1: Define schema with correct types and nullable

   Use StructType with StructField for 'user' as string, 'score' as integer, and 'date' as DateType, all nullable True.

2. Step 2: Read CSV with schema and header option

   Use spark.read.csv with schema argument and header=True to apply schema validation on read.

3. Final Answer:

   schema = StructType([StructField('user', StringType(), True), StructField('score', IntegerType(), True), StructField('date', DateType(), True)]) df = spark.read.csv('file.csv', schema=schema, header=True) -> Option D

4. Quick Check:

   Schema + header=True validates CSV read [OK]

Hint: Define schema with types, use header=True when reading CSV [OK]

Common Mistakes:

• Using StringType for all columns ignoring types
• Not setting header=True for CSV with headers
• Setting nullable=False unnecessarily causing errors