DAG performance tracking in Apache Airflow - Step-by-Step Execution

Process Flow - DAG performance tracking

DAG starts execution

↓

Record start time

↓

Run each task in DAG

↓

Record task start and end times

↓

Calculate task durations

↓

Aggregate task durations

↓

Calculate total DAG run time

↓

Store performance metrics

↓

Visualize or alert on performance

This flow shows how Airflow tracks the start and end times of each task and the entire DAG to measure performance.

Execution Sample

Apache Airflow

from airflow.models import DagRun
from airflow.utils.state import State

# Fetch DAG run
dag_run = DagRun.find(dag_id='example_dag')[0]

# Calculate duration
run_duration = dag_run.end_date - dag_run.start_date

This code fetches a DAG run and calculates its total duration by subtracting start time from end time.

Process Table

Step	Action	Task	Start Time	End Time	Duration	DAG Total Duration
1	DAG run started	N/A	10:00:00
2	Task A started	Task A	10:00:05
3	Task A ended	Task A		10:00:15	10 seconds
4	Task B started	Task B	10:00:16
5	Task B ended	Task B		10:00:30	14 seconds
6	Task C started	Task C	10:00:31
7	Task C ended	Task C		10:00:45	14 seconds
8	DAG run ended	N/A		10:00:50		50 seconds
9	Performance metrics stored	N/A				50 seconds

💡 DAG run ends at 10:00:50, total duration 50 seconds calculated and stored.

Status Tracker

Variable	Start	After Step 3	After Step 5	After Step 7	Final
Task A Duration	0	10 seconds	10 seconds	10 seconds	10 seconds
Task B Duration	0	0	14 seconds	14 seconds	14 seconds
Task C Duration	0	0	0	14 seconds	14 seconds
DAG Total Duration	0	0	0	0	50 seconds

Key Moments - 3 Insights

Why does the DAG total duration include gaps between tasks?

The DAG total duration measures from the first task start to the last task end, including any waiting or scheduling delays between tasks, as shown in steps 3 to 8.

How is each task's duration calculated?

Each task's duration is the difference between its start and end times, for example, Task A duration is 10 seconds from step 2 to 3.

Why do we store performance metrics after the DAG run ends?

Storing metrics after the DAG completes (step 9) allows tracking historical performance and detecting slowdowns or failures over time.

Visual Quiz - 3 Questions

Look at the execution table, what is the duration of Task B at step 5?

A 14 seconds

B 10 seconds

C 16 seconds

D 30 seconds

💡 Hint

Check the 'Duration' column for Task B at step 5 in the execution table.

At which step does the DAG run end according to the execution table?

A Step 7

B Step 9

C Step 8

D Step 6

💡 Hint

Look for the row with action 'DAG run ended' in the execution table.

If Task C took 20 seconds instead of 14, how would the DAG total duration change?

A It would remain 50 seconds

B It would increase to 56 seconds plus any gaps

C It would increase to 56 seconds

D It would decrease

💡 Hint

Consider that DAG total duration includes all task durations plus any waiting time between tasks.

Concept Snapshot

DAG Performance Tracking in Airflow:
- Track start and end times of each task
- Calculate task durations (end - start)
- Aggregate durations for total DAG run time
- Store metrics for monitoring
- Use Airflow metadata DB or logs
- Helps identify slow tasks and bottlenecks

Full Transcript

DAG performance tracking in Airflow involves recording the start and end times of each task and the entire DAG run. This allows calculation of individual task durations and the total time the DAG took to complete. The process starts when the DAG run begins, recording the start time. Each task's start and end times are tracked to compute durations. After all tasks finish, the total DAG duration is calculated from the first task start to the last task end. These performance metrics are stored for monitoring and analysis. This helps detect slow tasks and optimize workflows.

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