SLA misses and notifications in Apache Airflow - Mini Project: Build & Apply

SLA Misses and Notifications in Airflow

📖 Scenario: You are managing data pipelines using Apache Airflow. You want to track when tasks miss their Service Level Agreement (SLA) deadlines and send notifications to the team.This helps ensure timely data processing and quick response to delays.

🎯 Goal: Build an Airflow DAG that defines tasks with SLAs and configures an SLA miss callback function to send notifications.You will create the DAG, set SLA times, define the notification function, and verify the SLA miss handling.

📋 What You'll Learn

Create an Airflow DAG with at least two tasks

Set SLA times for the tasks using sla parameter

Define a function called sla_miss_callback that accepts dag, task_list, blocking_task_list, slas, session parameters

Configure the DAG to use sla_miss_callback for SLA miss notifications

Print a message inside sla_miss_callback showing which tasks missed SLA

Trigger the DAG run and observe SLA miss notification output

💡 Why This Matters

🌍 Real World

In real data engineering teams, monitoring SLAs ensures pipelines run on time. SLA miss notifications help quickly fix delays to keep data fresh.

💼 Career

Understanding SLA misses and notifications is important for DevOps and data engineering roles managing workflow reliability and alerting.

1

Create the Airflow DAG and tasks

Create an Airflow DAG called sla_notification_dag with start_date set to datetime(2024, 1, 1) and schedule_interval set to timedelta(days=1). Inside the DAG, create two PythonOperator tasks named task_a and task_b that run simple Python functions printing 'Task A executed' and 'Task B executed' respectively.

Apache Airflow

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

# Create the DAG and tasks below
# Your code here

Hint

Use DAG to create the DAG and PythonOperator for tasks. Define simple functions for python_callable.

2

Add SLA times to the tasks

Add an SLA of 10 minutes to task_a and 15 minutes to task_b by setting the sla parameter with timedelta(minutes=10) and timedelta(minutes=15) respectively in their PythonOperator definitions.

Apache Airflow

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

def task_a_func():
    print('Task A executed')

def task_b_func():
    print('Task B executed')

sla_notification_dag = DAG(
    'sla_notification_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval=timedelta(days=1)
)

# Add sla parameter to task_a and task_b below
# Your code here

Hint

Add the sla argument inside the PythonOperator constructor for each task.

3

Define the SLA miss callback function

Define a function called sla_miss_callback that accepts parameters dag, task_list, blocking_task_list, slas, and session. Inside the function, print a message: "SLA missed for tasks: {task_list}" using an f-string.

Apache Airflow

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

# Define sla_miss_callback function below
# Your code here

Hint

Define the function with the exact name and parameters. Use an f-string to print the task list.

4

Configure the DAG to use the SLA miss callback and print output

Add the sla_miss_callback function to the sla_notification_dag by setting the sla_miss_callback parameter in the DAG constructor. Then, write a print statement outside the DAG that prints "SLA miss notification setup complete".

Apache Airflow

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

# Add sla_miss_callback to DAG and print confirmation below
# Your code here

Hint

Pass sla_miss_callback=sla_miss_callback when creating the DAG. Use print() to show the confirmation message.