Task failure callbacks in Apache Airflow - Commands & Configuration

Introduction

Sometimes tasks in a workflow fail. Task failure callbacks let you run special code automatically when a task fails. This helps you react quickly, like sending alerts or cleaning up resources.

When you want to send an email notification if a task fails in your workflow

When you need to log extra information for debugging after a task failure

When you want to trigger a cleanup process if a task does not complete successfully

When you want to restart or rerun certain tasks automatically after failure

When you want to update a dashboard or monitoring system about task failures

Config File - my_dag.py

my_dag.py

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from airflow.utils.email import send_email

def task_that_fails():
    raise Exception('This task fails on purpose')

def failure_callback(context):
    task_instance = context.get('task_instance')
    dag_run = context.get('dag_run')
    message = f"Task {task_instance.task_id} failed in DAG {dag_run.dag_id}"
    print(message)
    send_email(to='[e-mail]', subject='Task Failure Alert', html_content=message)

def success_task():
    print('This task runs if previous tasks succeed')

with DAG(dag_id='example_failure_callback', start_date=days_ago(1), schedule_interval='@daily', catchup=False) as dag:
    failing_task = PythonOperator(
        task_id='failing_task',
        python_callable=task_that_fails,
        on_failure_callback=failure_callback
    )

    success = PythonOperator(
        task_id='success_task',
        python_callable=success_task
    )

    failing_task >> success

This DAG defines two tasks: one that always fails and one that runs after it.

The failure_callback function runs automatically when failing_task fails. It prints a message and sends an email alert.

The on_failure_callback parameter links the callback function to the task.

Commands

List all available DAGs to confirm your DAG is recognized by Airflow.

Terminal

airflow dags list

Expected Output

example_failure_callback

Manually start the DAG run to test the failure callback in action.

Terminal

airflow dags trigger example_failure_callback

Expected Output

Created <DagRun example_failure_callback @ 2024-06-01T00:00:00+00:00, manual__2024-06-01T00:00:00+00:00, externally triggered: True>

List tasks in the DAG to verify task names and order.

Terminal

airflow tasks list example_failure_callback

Expected Output

failing_task

success_task

Run the failing task manually for the given date to see the failure and callback in action.

Terminal

airflow tasks test example_failure_callback failing_task 2024-06-01

Expected Output

Running: ['python', '-m', 'airflow', 'tasks', 'run', 'example_failure_callback', 'failing_task', '2024-06-01', '--local']

[2024-06-01 00:00:00,000] {python.py:114} INFO - Executing <Task(PythonOperator): failing_task> on 2024-06-01

[2024-06-01 00:00:00,100] {python.py:120} INFO - Running task_that_fails

[2024-06-01 00:00:00,200] {taskinstance.py:1234} ERROR - This task fails on purpose

[2024-06-01 00:00:00,300] {taskinstance.py:1250} INFO - Task failing_task failed

Task failing_task failed in DAG example_failure_callback

[2024-06-01 00:00:00,400] {email.py:56} INFO - Sending email to [e-mail]

Key Concept

If you remember nothing else from this pattern, remember: on_failure_callback lets you run custom code automatically whenever a task fails.

Common Mistakes

Not passing the context parameter to the failure callback function

Without context, the callback cannot access task details like task_id or dag_id, so it cannot report useful information.

Define the callback function to accept a single 'context' argument and use it to get task and DAG info.

Assigning on_failure_callback to the DAG instead of the individual task

The DAG-level callback runs on DAG failure, not on individual task failures, so task failures won't trigger it.

Set on_failure_callback on the specific task(s) that need failure handling.

Not importing or configuring email sending properly in the callback

The callback tries to send email but fails silently or errors out, so alerts are not sent.

Import airflow.utils.email.send_email and configure SMTP settings in Airflow to enable email sending.

Summary

Define a Python function that accepts a context argument to handle task failure events.

Assign this function to the on_failure_callback parameter of the task you want to monitor.

Trigger the DAG and observe the callback running automatically when the task fails.

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