Handling schema changes in data pipelines in Apache Airflow - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to define a DAG with a daily schedule.

Apache Airflow

with DAG('data_pipeline', schedule_interval=[1], start_date=datetime(2024, 1, 1)) as dag:

Drag options to blanks, or click blank then click option'

A'@monthly'

B'@daily'

C'@weekly'

D'@hourly'

Attempts:

3 left

💡 Hint

Common Mistakes

Using '@hourly' when daily runs are needed.

Forgetting to put quotes around the schedule string.

The schedule_interval set to '@daily' runs the DAG once every day.

2fill in blank

medium

Complete the code to create a task that runs a Python function.

Apache Airflow

task = PythonOperator(task_id='process_data', python_callable=[1], dag=dag)

Drag options to blanks, or click blank then click option'

Aprocess_data_function

B'process_data_function'

Cprocess_data_function()

Dprocess_data_function()()

Attempts:

3 left

💡 Hint

Common Mistakes

Adding parentheses after the function name.

Passing the function name as a string.

The python_callable argument expects a function reference without parentheses.

3fill in blank

hard

Fix the error in the task dependency definition.

Apache Airflow

task1 [1] task2

Drag options to blanks, or click blank then click option'

A->

B><

C=>

D>>

Attempts:

3 left

💡 Hint

Common Mistakes

Using '=>' which is not valid in Airflow.

Using '->' which is not recognized.

The >> operator sets task1 to run before task2 in Airflow.

4fill in blank

hard

Fill both blanks to create a task that checks for a schema file before processing.

Apache Airflow

check_schema = [1](task_id='check_schema_file', filepath='schema.json', dag=dag)
process = [2](task_id='process_data', python_callable=process_data, dag=dag)

Drag options to blanks, or click blank then click option'

AFileSensor

BPythonOperator

CBashOperator

DDummyOperator

Attempts:

3 left

💡 Hint

Common Mistakes

Using BashOperator to check for files instead of FileSensor.

Using DummyOperator which does nothing.

FileSensor waits for the schema file, then PythonOperator runs the processing function.

5fill in blank

hard

Fill both blanks to create a dictionary comprehension that filters columns based on schema.

Apache Airflow

filtered_data = {col:val for col, val in data.items() if col [1] schema and val [2] None}

Drag options to blanks, or click blank then click option'

A:

Bin

Cis not

D==

Attempts:

3 left

💡 Hint

Common Mistakes

Using '=' instead of ':' in dict comprehension.

Using '==' instead of 'is not' for None check.

The comprehension creates a dict with columns and values where the column is in the schema and value is not None.

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