Task failure callbacks in Apache Airflow - Mini Project: Build & Apply

Airflow Task Failure Callbacks

📖 Scenario: You are managing workflows in Apache Airflow. You want to be notified whenever a task fails in your workflow. This helps you quickly respond to issues.

🎯 Goal: Build an Airflow DAG that uses a task failure callback function to print a message when a task fails.

📋 What You'll Learn

Create a Python function called task_failure_alert that takes context as a parameter and prints 'Task failed!'

Create an Airflow DAG named failure_callback_dag with default arguments including on_failure_callback set to task_failure_alert

Add a task called failing_task that always fails by raising an exception

Run the DAG and observe the failure callback message printed

💡 Why This Matters

🌍 Real World

In real projects, task failure callbacks help alert teams immediately when something goes wrong in automated workflows.

💼 Career

Knowing how to use failure callbacks is important for DevOps engineers and data engineers managing Airflow pipelines to ensure reliability and quick issue response.

1

Create the failure callback function

Write a Python function called task_failure_alert that takes one parameter context and prints the text 'Task failed!' inside the function.

Apache Airflow

# Define the failure callback function
# Your code here

Hint

This function will be called by Airflow when a task fails. It must accept one argument called context.

2

Create the DAG with failure callback

Import DAG and datetime. Create a DAG named failure_callback_dag with start_date set to datetime(2024, 1, 1) and on_failure_callback set to the function task_failure_alert.

Apache Airflow

def task_failure_alert(context):
    print('Task failed!')

# Import DAG and datetime
# Create failure_callback_dag with on_failure_callback=task_failure_alert
# Your code here

Hint

Use schedule_interval='@once' to run the DAG once for testing.

3

Add a failing task to the DAG

Import PythonOperator from airflow.operators.python. Add a task called failing_task to failure_callback_dag using PythonOperator. The task should run a Python function that raises an exception with the message 'Failing on purpose'.

Apache Airflow

from airflow import DAG
from datetime import datetime

def task_failure_alert(context):
    print('Task failed!')

failure_callback_dag = DAG(
    'failure_callback_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@once',
    on_failure_callback=task_failure_alert
)

# Import PythonOperator
# Define a function that raises Exception('Failing on purpose')
# Create failing_task using PythonOperator with the function and DAG
# Your code here

Hint

The task must raise an exception to trigger the failure callback.

4

Run the DAG and observe the failure callback output

Run the DAG failure_callback_dag and observe the output. Write a print statement that outputs exactly 'Task failed!' to simulate the failure callback message.

Apache Airflow

from airflow import DAG
from datetime import datetime
from airflow.operators.python import PythonOperator

def task_failure_alert(context):
    print('Task failed!')

def fail_function():
    raise Exception('Failing on purpose')

failure_callback_dag = DAG(
    'failure_callback_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@once',
    on_failure_callback=task_failure_alert
)

failing_task = PythonOperator(
    task_id='failing_task',
    python_callable=fail_function,
    dag=failure_callback_dag
)

# Print the failure callback message
# Your code here

Hint

This simulates the output you would see when the failure callback runs.

Practice

1. What is the main purpose of on_failure_callback in an Airflow task?

easy

A. To run a function automatically when the task fails

B. To schedule the task to run at a specific time

C. To retry the task multiple times on failure

D. To log the task's successful completion

Solution

1. Step 1: Understand the role of on_failure_callback

   This parameter is designed to specify a function that runs automatically when a task fails.

2. Step 2: Differentiate from other task features

   Scheduling and retries are handled by other parameters, not on_failure_callback.

3. Final Answer:

   To run a function automatically when the task fails -> Option A

4. Quick Check:

   Failure callback = automatic function on failure [OK]

Hint: Failure callback runs code only when task fails [OK]

Common Mistakes:

• Confusing failure callback with scheduling
• Thinking it retries the task automatically
• Assuming it runs on task success

2. Which of the following is the correct way to assign a failure callback function named notify_failure to an Airflow task?

easy

A. task = PythonOperator(failure_callback=notify_failure, ...)

B. task = PythonOperator(on_failure_callback=notify_failure, ...)

C. task = PythonOperator(on_fail_callback=notify_failure, ...)

D. task = PythonOperator(callback_on_failure=notify_failure, ...)

Solution

1. Step 1: Recall the exact parameter name

   The correct parameter to assign a failure callback is on_failure_callback.

2. Step 2: Match the parameter with the options

   Only task = PythonOperator(on_failure_callback=notify_failure, ...) uses on_failure_callback correctly.

3. Final Answer:

   task = PythonOperator(on_failure_callback=notify_failure, ...) -> Option B

4. Quick Check:

   Correct parameter name = on_failure_callback [OK]

Hint: Use exact parameter name: on_failure_callback [OK]

Common Mistakes:

• Using incorrect parameter names like failure_callback
• Swapping words in the parameter name
• Misspelling the parameter

3. Given this code snippet, what will be printed if the task fails?

def alert(context):
    print(f"Task {context['task_instance'].task_id} failed")

task = PythonOperator(
    task_id='my_task',
    python_callable=lambda: 1/0,
    on_failure_callback=alert
)

medium

A. Task succeeded

B. ZeroDivisionError

C. No output, task succeeds

D. Task my_task failed

Solution

1. Step 1: Analyze the task's python_callable

   The lambda divides 1 by 0, causing a failure (ZeroDivisionError).

2. Step 2: Understand the failure callback behavior

   On failure, the alert function runs and prints the failure message with the task ID.

3. Final Answer:

   Task my_task failed -> Option D

4. Quick Check:

   Failure triggers alert print [OK]

Hint: Failure triggers callback print message [OK]

Common Mistakes:

• Expecting ZeroDivisionError to print directly
• Thinking task succeeds despite error
• Ignoring the callback print output

4. You wrote this failure callback function:

def fail_alert():
    print("Task failed")

task = PythonOperator(
    task_id='fail_task',
    python_callable=lambda: 1/0,
    on_failure_callback=fail_alert
)

Why does the callback not run when the task fails?

medium

A. The callback function must be decorated with @task

B. The callback function must return a value

C. The callback function must accept a context parameter

D. The callback function must be async

Solution

1. Step 1: Check callback function signature

   Failure callbacks receive a context dictionary argument; missing it causes errors.

2. Step 2: Understand Airflow callback requirements

   The callback must accept one parameter (context) to work properly.

3. Final Answer:

   The callback function must accept a context parameter -> Option C

4. Quick Check:

   Callback needs context argument [OK]

Hint: Callback must accept context argument [OK]

Common Mistakes:

• Defining callback without parameters
• Expecting return value from callback
• Making callback async unnecessarily

5. You want to send an email alert only when a task fails and the failure reason contains the word "timeout". How should you implement the on_failure_callback function?

hard

A. Define a function that checks context['exception'] for 'timeout' and sends email if true

B. Define a function that always sends email regardless of failure reason

C. Use on_failure_callback to retry the task on timeout

D. Set on_failure_callback to None and handle alerts elsewhere

Solution

1. Step 1: Understand the requirement

   Send email only if failure reason contains "timeout".

2. Step 2: Use context to check exception message

   The failure callback receives context with exception info; check if 'timeout' is in the exception string.

3. Step 3: Send email conditionally

   Send email alert only if the condition is met inside the callback function.

4. Final Answer:

   Define a function that checks context['exception'] for 'timeout' and sends email if true -> Option A

5. Quick Check:

   Check exception text in callback before alert [OK]

Hint: Check exception text in callback before alert [OK]

Common Mistakes:

• Sending email on all failures without filtering
• Trying to retry task in failure callback
• Ignoring exception details in context