Handling schema changes in data pipelines in Apache Airflow - Step-by-Step Execution

Process Flow - Handling schema changes in data pipelines

Detect Schema Change

↓

Validate New Schema

↓

Update Pipeline Code

↓

Test Pipeline with New Schema

↓

Deploy Updated Pipeline

↓

Monitor Pipeline Runs

↓

Handle Errors or Rollback

This flow shows how a data pipeline detects and adapts to schema changes step-by-step to keep data processing smooth.

Execution Sample

Apache Airflow

def check_schema_change(old_schema, new_schema):
    return old_schema != new_schema

if check_schema_change(current_schema, incoming_schema):
    update_pipeline()

This code checks if the schema has changed and triggers a pipeline update if needed.

Process Table

Step	Action	Input/Condition	Result/Decision	Next Step
1	Detect schema change	old_schema vs new_schema	Schemas differ	Proceed to validate new schema
2	Validate new schema	Check required fields and types	Validation passed	Update pipeline code
3	Update pipeline code	Modify DAG or tasks to handle new schema	Code updated	Test pipeline with new schema
4	Test pipeline	Run pipeline with new schema data	Pipeline runs successfully	Deploy updated pipeline
5	Deploy pipeline	Push changes to Airflow environment	Pipeline deployed	Monitor pipeline runs
6	Monitor runs	Check logs and metrics	No errors detected	End
7	Exit	If errors occur	Rollback or fix issues	Repeat testing or deployment

💡 Execution stops when pipeline runs successfully with new schema or rollback is done after errors.

Status Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	After Step 4	After Step 5	Final
old_schema	{'id': int, 'name': str}	{'id': int, 'name': str}	{'id': int, 'name': str}	{'id': int, 'name': str}	{'id': int, 'name': str}	{'id': int, 'name': str}	{'id': int, 'name': str}
new_schema	{'id': int, 'name': str}	{'id': int, 'name': str, 'email': str}	{'id': int, 'name': str, 'email': str}	{'id': int, 'name': str, 'email': str}	{'id': int, 'name': str, 'email': str}	{'id': int, 'name': str, 'email': str}	{'id': int, 'name': str, 'email': str}
pipeline_code	Original code	Original code	Original code	Updated code	Updated code	Deployed code	Deployed code
pipeline_status	Idle	Detected schema change	Validated schema	Code updated	Tested successfully	Deployed	Stable

Key Moments - 3 Insights

Why do we need to validate the new schema before updating the pipeline?

Validating the new schema ensures it has all required fields and correct types. This prevents pipeline failures later, as shown in step 2 of the execution table.

What happens if the pipeline test fails after updating for the new schema?

If testing fails (step 4), the pipeline should not be deployed. Instead, fix the code or rollback as described in step 7 to avoid breaking data processing.

How does monitoring help after deploying the updated pipeline?

Monitoring checks logs and metrics to catch errors early (step 6). This helps maintain data quality and lets you react quickly if something goes wrong.

Visual Quiz - 3 Questions

Look at the execution table, at which step is the pipeline code updated to handle the new schema?

A Step 2

B Step 3

C Step 4

D Step 5

💡 Hint

Check the 'Action' column for where code modification happens.

According to the variable tracker, what is the value of 'pipeline_status' after step 4?

A Tested successfully

B Detected schema change

C Idle

D Deployed

💡 Hint

Look at the 'pipeline_status' row under 'After Step 4' column.

If the new schema did not include the 'email' field, how would the 'new_schema' variable change in the tracker?

A It would be empty

B It would include 'email' field

C It would remain the same as old_schema

D It would cause an error immediately

💡 Hint

Compare 'new_schema' and 'old_schema' values in the variable tracker.

Concept Snapshot

Handling schema changes in data pipelines:
1. Detect schema differences between old and new data.
2. Validate new schema fields and types.
3. Update pipeline code (DAG/tasks) to handle changes.
4. Test pipeline with new schema data.
5. Deploy updated pipeline and monitor runs.
6. Rollback if errors occur to keep data safe.

Full Transcript

This visual execution shows how to handle schema changes in data pipelines using Airflow. First, the pipeline detects if the incoming data schema differs from the current one. Then it validates the new schema to ensure all required fields and types are correct. Next, the pipeline code is updated to handle the new schema, such as adding new fields. After updating, the pipeline is tested with the new schema data to confirm it runs without errors. Once tests pass, the updated pipeline is deployed to the Airflow environment. Finally, the pipeline runs are monitored for errors or issues. If any errors occur, the pipeline can be rolled back or fixed before redeploying. This step-by-step approach helps keep data pipelines reliable and adaptable to changes in data structure.

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