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.