DAG performance tracking in Apache Airflow - Mini Project: Build & Apply

DAG Performance Tracking in Airflow

📖 Scenario: You are managing workflows using Apache Airflow. You want to track the performance of your DAG runs by recording the duration of each run.This helps you understand how long your workflows take and identify any delays.

🎯 Goal: Build a simple Airflow DAG that tracks the duration of each DAG run and stores this information in a Python dictionary.You will create the data structure, add a configuration variable for minimum duration threshold, calculate durations, and finally print the durations.

📋 What You'll Learn

Create a dictionary to store DAG run durations

Add a minimum duration threshold variable

Calculate the duration of each DAG run using start and end times

Print the dictionary with DAG run durations

💡 Why This Matters

🌍 Real World

Tracking DAG run durations helps teams monitor workflow performance and detect delays or failures early.

💼 Career

Understanding how to track and analyze DAG performance is important for DevOps engineers and data engineers working with workflow automation.

1

Create a dictionary to store DAG run durations

Create a dictionary called dag_run_durations with these exact entries: 'run_1': 0, 'run_2': 0, 'run_3': 0.

Apache Airflow

# Create a dictionary called dag_run_durations with keys 'run_1', 'run_2', 'run_3' and values 0
# Your code here

Hint

Use curly braces to create a dictionary with keys and values.

2

Add a minimum duration threshold variable

Add a variable called min_duration_threshold and set it to 5 (minutes).

Apache Airflow

dag_run_durations = {'run_1': 0, 'run_2': 0, 'run_3': 0}
# Add a variable min_duration_threshold set to 5
# Your code here

Hint

Just assign the number 5 to the variable min_duration_threshold.

3

Calculate the duration of each DAG run

Given these start and end times in minutes for each run: start_times = {'run_1': 0, 'run_2': 10, 'run_3': 20} and end_times = {'run_1': 7, 'run_2': 15, 'run_3': 30}, use a for loop with variables run and start to iterate over start_times.items(). Calculate the duration as end_times[run] - start and update dag_run_durations[run] with this value.

Apache Airflow

dag_run_durations = {'run_1': 0, 'run_2': 0, 'run_3': 0}
min_duration_threshold = 5
start_times = {'run_1': 0, 'run_2': 10, 'run_3': 20}
end_times = {'run_1': 7, 'run_2': 15, 'run_3': 30}
# Use a for loop with run and start to calculate durations and update dag_run_durations
# Your code here

Hint

Use a for loop to go through start_times and calculate duration using end_times.

4

Print the dictionary with DAG run durations

Write a print statement to display the dag_run_durations dictionary.

Apache Airflow

dag_run_durations = {'run_1': 0, 'run_2': 0, 'run_3': 0}
min_duration_threshold = 5
start_times = {'run_1': 0, 'run_2': 10, 'run_3': 20}
end_times = {'run_1': 7, 'run_2': 15, 'run_3': 30}
for run, start in start_times.items():
    duration = end_times[run] - start
    dag_run_durations[run] = duration
# Print the dag_run_durations dictionary
# Your code here

Hint

Use print(dag_run_durations) to show the durations.

Practice

1. What is the main purpose of tracking DAG performance in Airflow?

easy

A. To measure how long workflows and tasks take to run

B. To change the order of tasks in a DAG

C. To create new DAGs automatically

D. To delete old task logs

Solution

1. Step 1: Understand DAG performance tracking

   DAG performance tracking focuses on measuring execution times of workflows and tasks.

2. Step 2: Identify the main goal

   The goal is to find how long tasks and DAG runs take to help improve pipeline speed.

3. Final Answer:

   To measure how long workflows and tasks take to run -> Option A

4. Quick Check:

   DAG performance tracking = measuring run times [OK]

Hint: Think: performance means measuring speed or time [OK]

Common Mistakes:

• Confusing performance tracking with task ordering
• Thinking it creates or deletes DAGs
• Mixing logs management with performance tracking

2. Which Airflow model stores information about individual task execution times?

easy

A. DagRun

B. Variable

C. TaskInstance

D. Connection

Solution

1. Step 1: Identify models related to execution times

   DagRun stores info about the whole DAG run, TaskInstance stores info about each task run.

2. Step 2: Match model to task execution time

   TaskInstance tracks start and end times for each task, so it holds execution time data.

3. Final Answer:

   TaskInstance -> Option C

4. Quick Check:

   TaskInstance = task execution info [OK]

Hint: TaskInstance = single task run details [OK]

Common Mistakes:

• Choosing DagRun which tracks whole DAG, not tasks
• Confusing Variables or Connections with timing data
• Assuming all info is in one model

3. Given this code snippet, what will be printed?

from airflow.models import TaskInstance
from datetime import datetime, timedelta

start = datetime(2024, 6, 1, 10, 0, 0)
end = datetime(2024, 6, 1, 10, 5, 30)
ti = TaskInstance(dag_id='example_dag', task_id='task1', execution_date=start)
ti.start_date = start
ti.end_date = end
print((ti.end_date - ti.start_date).total_seconds())

medium

A. 5.5

B. 330

C. SyntaxError

D. 330.0

Solution

1. Step 1: Identify variable name mismatch

   The code assigns TaskInstance to 'ti' but prints using 'ti', which is correct. However, the code snippet uses 'ti' consistently, so no mismatch here.

2. Step 2: Check for syntax errors

   The code is syntactically correct, but the TaskInstance constructor requires a DAG object or context in recent Airflow versions, so this code may raise a runtime error.

3. Step 3: Consider the actual output

   Assuming the code runs, the output would be 330.0 seconds.

4. Final Answer:

   330.0 -> Option D

5. Quick Check:

   5 min 30 sec = 330.0 seconds [OK]

Hint: Subtract datetimes, use total_seconds() for float seconds [OK]

Common Mistakes:

• Confusing minutes with seconds
• Expecting integer instead of float
• Syntax errors from missing imports (not here)

4. You wrote this code to get task duration but it raises an error:

ti = TaskInstance(dag_id='example_dag', task_id='task1', execution_date='2024-06-01T10:00:00')
ti.start_date = datetime(2024, 6, 1, 10, 0, 0)
ti.end_date = datetime(2024, 6, 1, 10, 5, 0)
duration = (ti.end_date - ti.start_date).total_seconds()
print(duration)

What is the likely cause of the error?

medium

A. execution_date should be a datetime, not a string

B. start_date and end_date must be strings

C. TaskInstance requires a DAG object on creation

D. total_seconds() is not a valid method

Solution

1. Step 1: Check execution_date type

   execution_date must be a datetime object, not a string, to avoid errors.

2. Step 2: Confirm other parts are correct

   start_date and end_date are datetime objects, total_seconds() is valid method.

3. Final Answer:

   execution_date should be a datetime, not a string -> Option A

4. Quick Check:

   execution_date type error = execution_date should be a datetime, not a string [OK]

Hint: Pass datetime objects, not strings, for execution_date [OK]

Common Mistakes:

• Passing string instead of datetime for execution_date
• Thinking start_date/end_date must be strings
• Believing total_seconds() is invalid

5. You want to track the average duration of all tasks in a DAG run using Airflow's ORM models. Which approach is best?

hard

A. Calculate average duration by counting TaskInstance rows only

B. Query TaskInstance for all tasks of the DAG run, calculate average of (end_date - start_date)

C. Use Airflow Variables to store durations manually

D. Query DagRun for duration and divide by number of tasks

Solution

1. Step 1: Understand data sources for task durations

   TaskInstance has start_date and end_date for each task, so durations come from there.

2. Step 2: Calculate average duration correctly

   Query all TaskInstances for the DAG run, compute (end_date - start_date) for each, then average.

3. Final Answer:

   Query TaskInstance for all tasks of the DAG run, calculate average of (end_date - start_date) -> Option B

4. Quick Check:

   Average task duration = average of TaskInstance times [OK]

Hint: Average durations come from TaskInstance times, not DagRun [OK]

Common Mistakes:

• Using DagRun duration divided by task count (wrong average)
• Storing durations manually in Variables (extra work)
• Counting rows without timing info