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.