Handling schema changes in data pipelines in Apache Airflow - Mini Project: Build & Apply

Handling schema changes in data pipelines

📖 Scenario: You work as a data engineer managing data pipelines using Apache Airflow. Sometimes, the structure of incoming data changes, like new columns added or removed. You want to build a simple Airflow DAG that reads data, checks for schema changes, and handles them gracefully to avoid pipeline failures.

🎯 Goal: Create an Airflow DAG that loads a sample dataset, defines the expected schema, compares the incoming data schema with the expected one, and logs a message if there is a schema change.

📋 What You'll Learn

Create a sample dataset as a Python dictionary with fixed columns

Define an expected schema as a list of column names

Write a Python function to compare the dataset columns with the expected schema

Create an Airflow DAG with tasks to load data, check schema, and log results

💡 Why This Matters

🌍 Real World

Data pipelines often break when data formats change unexpectedly. Detecting schema changes early helps keep pipelines stable and data accurate.

💼 Career

Data engineers and DevOps professionals use Airflow to automate workflows and handle schema changes to prevent pipeline failures.

1

Create sample dataset

Create a Python dictionary called sample_data with these exact keys and values: 'id': [1, 2, 3], 'name': ['Alice', 'Bob', 'Charlie'], and 'age': [25, 30, 35].

Apache Airflow

# Create the sample_data dictionary with keys 'id', 'name', and 'age'
# Your code here

Hint

Use a dictionary with keys 'id', 'name', and 'age'. Each key should have a list of values.

2

Define expected schema

Create a list called expected_schema containing the exact strings: 'id', 'name', and 'age'.

Apache Airflow

sample_data = {
    'id': [1, 2, 3],
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [25, 30, 35]
}
# Define the expected_schema list with 'id', 'name', and 'age'
# Your code here

Hint

Use a list with the exact column names as strings.

3

Write schema check function

Write a function called check_schema_change that takes data and expected as parameters. Inside, get the list of keys from data as data_schema. Return True if data_schema is not equal to expected, otherwise return False.

Apache Airflow

sample_data = {
    'id': [1, 2, 3],
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [25, 30, 35]
}

expected_schema = ['id', 'name', 'age']

# Define function check_schema_change(data, expected) below
# Your code here

Hint

Use list(data.keys()) to get the current schema and compare it with expected.

4

Create Airflow DAG and log schema check

Create an Airflow DAG named schema_change_dag with default args including start_date as days_ago(1). Inside the DAG, define a PythonOperator task called check_schema_task that calls a function schema_check. Define schema_check to use check_schema_change(sample_data, expected_schema) and print 'Schema changed!' if it returns True, else print 'Schema is as expected.'. Finally, set the task in the DAG context.

Apache Airflow

sample_data = {
    'id': [1, 2, 3],
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [25, 30, 35]
}

expected_schema = ['id', 'name', 'age']

def check_schema_change(data, expected):
    data_schema = list(data.keys())
    return data_schema != expected

# Import required Airflow modules
# Define schema_check function
# Create DAG named schema_change_dag with start_date days_ago(1)
# Define PythonOperator check_schema_task calling schema_check
# Set task in DAG context
# Your code here

Hint

Use PythonOperator to run the schema_check function inside the DAG context.

Practice

1. What is the main reason to handle schema changes in Airflow data pipelines?

easy

A. To prevent pipeline failures due to unexpected data format changes

B. To speed up the execution of tasks

C. To reduce the number of DAGs in Airflow

D. To increase the number of parallel tasks

Solution

1. Step 1: Understand schema changes impact

   Schema changes can cause data format mismatches that break pipelines.

2. Step 2: Identify the main goal of handling schema changes

   Handling schema changes aims to avoid pipeline failures by validating schemas.

3. Final Answer:

   To prevent pipeline failures due to unexpected data format changes -> Option A

4. Quick Check:

   Prevent failures = C [OK]

Hint: Schema changes break pipelines; handle to avoid failure [OK]

Common Mistakes:

• Thinking schema changes speed up tasks
• Confusing schema handling with DAG count
• Assuming schema changes increase parallelism

2. Which Airflow operator is best suited to check and validate data schema before processing?

easy

A. PythonOperator

B. BashOperator

C. DummyOperator

D. EmailOperator

Solution

1. Step 1: Identify operator for custom Python code

   PythonOperator runs Python functions, ideal for schema checks.

2. Step 2: Compare with other operators

   BashOperator runs shell commands, DummyOperator does nothing, EmailOperator sends emails.

3. Final Answer:

   PythonOperator -> Option A

4. Quick Check:

   Python code for schema check = A [OK]

Hint: Use PythonOperator for custom schema validation code [OK]

Common Mistakes:

• Choosing BashOperator for Python code
• Using DummyOperator which does nothing
• Selecting EmailOperator unrelated to schema checks

3. Given this PythonOperator task in Airflow, what happens if the schema check fails by raising an Exception?

def check_schema():
    if not schema_is_valid():
        raise ValueError('Schema mismatch detected')

check_schema_task = PythonOperator(
    task_id='check_schema',
    python_callable=check_schema,
    dag=dag
)

medium

A. The task completes successfully ignoring the error

B. The task fails and the DAG run is marked as failed

C. The task retries automatically without failing

D. The task is skipped and DAG continues

Solution

1. Step 1: Understand exception effect in PythonOperator

   Raising an Exception causes the task to fail in Airflow.

2. Step 2: Check DAG run status on task failure

   When a task fails without retries, the DAG run is marked failed.

3. Final Answer:

   The task fails and the DAG run is marked as failed -> Option B

4. Quick Check:

   Exception raised = task fail = D [OK]

Hint: Exception in PythonOperator causes task failure [OK]

Common Mistakes:

• Assuming task skips on exception
• Thinking task retries automatically without retry config
• Believing task completes despite error

4. You wrote this schema validation function in Airflow but the DAG fails to detect schema mismatches properly. What is the error?

def validate_schema():
    if schema_is_valid():
        raise ValueError('Schema mismatch detected')

validate_task = PythonOperator(
    task_id='validate_schema',
    python_callable=validate_schema,
    dag=dag
)

medium

A. The PythonOperator is missing retries parameter

B. The task_id is invalid and causes failure

C. The condition is reversed; it raises error when schema is valid

D. The function does not return any value

Solution

1. Step 1: Analyze the condition logic

   The function raises error if schema_is_valid() is True, which is wrong.

2. Step 2: Identify correct logic for schema validation

   Error should raise when schema_is_valid() is False, indicating mismatch.

3. Final Answer:

   The condition is reversed; it raises error when schema is valid -> Option C

4. Quick Check:

   Raise error on invalid schema, not valid = B [OK]

Hint: Check if condition logic matches error raising intent [OK]

Common Mistakes:

• Reversing condition logic
• Ignoring task_id naming rules
• Expecting return value in PythonOperator callable

5. You want to handle schema changes gracefully in Airflow by branching: if schema matches, continue processing; if not, send alert and stop. Which Airflow feature helps implement this logic?

hard

A. LatestOnlyOperator to run tasks only on latest DAG run

B. SubDagOperator to run schema check in a sub-DAG

C. TriggerDagRunOperator to start a new DAG on schema mismatch

D. BranchPythonOperator to choose next task based on schema check

Solution

1. Step 1: Understand branching in Airflow

   BranchPythonOperator allows choosing next task based on condition.

2. Step 2: Match branching to schema check use case

   Use BranchPythonOperator to continue or alert based on schema validity.

3. Final Answer:

   BranchPythonOperator to choose next task based on schema check -> Option D

4. Quick Check:

   Branching logic = BranchPythonOperator = A [OK]

Hint: Use BranchPythonOperator for conditional task flow [OK]

Common Mistakes:

• Using SubDagOperator which is for nested DAGs
• TriggerDagRunOperator starts new DAG, not branching
• LatestOnlyOperator controls run timing, not branching