Handling schema changes in data pipelines in Apache Airflow - Practice Problems & Coding Challenges

Challenge - 5 Problems

🧠 Conceptual

intermediate

Why use schema validation in Airflow data pipelines?

In data pipelines managed by Airflow, why is it important to validate schema changes before processing data?

A To prevent pipeline failures caused by unexpected data format changes

B To speed up the execution of Airflow DAGs by skipping data checks

C To automatically fix data errors without human intervention

D To reduce the number of tasks in the DAG by merging schema checks with data loading

💡 Hint

Think about what happens if the data format changes unexpectedly.

Validating schema helps catch unexpected changes early, preventing pipeline crashes and data corruption.

💻 Command Output

intermediate

Output of Airflow task with schema mismatch

What will be the output when an Airflow PythonOperator task runs a function that raises a ValueError due to schema mismatch?

Apache Airflow

def check_schema(data):
    if 'id' not in data:
        raise ValueError('Missing id field')

check_schema({'name': 'Alice'})

A Task retries automatically without error

B Task succeeds and logs 'Schema valid'

C Task skips execution and marks as success

D Task fails with ValueError: Missing id field

💡 Hint

What happens when a Python function raises an exception inside an Airflow task?

Raising ValueError causes the task to fail and Airflow logs the error.

🔀 Workflow

advanced

Best workflow to handle schema changes in Airflow

Which Airflow workflow best handles schema changes in a data pipeline to minimize downtime?

A Add a schema validation task before data processing and alert on failures

B Skip schema validation and rely on downstream tasks to handle errors

C Disable all schema checks to speed up pipeline execution

D Run data processing tasks first, then validate schema after completion

💡 Hint

Consider when to catch schema issues to avoid wasting resources.

Validating schema before processing prevents wasted work and allows quick alerts for fixes.

❓ Troubleshoot

advanced

Troubleshooting schema mismatch errors in Airflow logs

You see this error in Airflow logs: KeyError: 'user_id'. What is the most likely cause?

A The Airflow webserver is not running, so logs are incomplete

B The input data is missing the 'user_id' field expected by the task

C The DAG file has syntax errors causing task failures

D The Airflow scheduler is down and cannot assign tasks

💡 Hint

KeyError usually means a missing dictionary key in Python.

The error indicates the code tried to access a missing field in the input data.

✅ Best Practice

expert

Handling evolving schemas in Airflow pipelines

What is the best practice to handle evolving data schemas in Airflow pipelines to ensure smooth updates?

A Ignore schema changes and fix errors as they appear during pipeline runs

B Hardcode schema fields in tasks and update DAGs only when schema changes

C Implement schema versioning and use conditional branching in DAGs to process different versions

D Disable schema validation to avoid pipeline failures during schema evolution

💡 Hint

Think about how to support multiple schema versions without breaking pipelines.

Schema versioning with conditional DAG logic allows pipelines to adapt to changes smoothly.

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