DAG performance tracking in Apache Airflow - Commands & Configuration

Introduction

Tracking the performance of your Airflow DAGs helps you understand how long tasks take and if any are failing. This lets you improve your workflows and fix problems quickly.

When you want to see how long each task in your workflow takes to run.

When you need to find out if any tasks are failing or retrying often.

When you want to monitor the overall health and efficiency of your data pipelines.

When you want to compare performance before and after changes to your DAGs.

When you want to alert your team if tasks take too long or fail.

Config File - my_dag.py

my_dag.py

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from datetime import timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'performance_tracking_example',
    default_args=default_args,
    description='A simple DAG to demonstrate performance tracking',
    schedule_interval='@daily',
    start_date=days_ago(2),
    catchup=False,
    tags=['example'],
)

task1 = BashOperator(
    task_id='print_date',
    bash_command='date',
    dag=dag,
)

task2 = BashOperator(
    task_id='sleep',
    bash_command='sleep 5',
    dag=dag,
)

task1 >> task2

This DAG runs two simple tasks: printing the date and sleeping for 5 seconds.

Airflow automatically tracks the start time, end time, duration, and status of each task run.

The default_args set retry behavior and owner info.

The DAG runs daily starting two days ago, without catching up missed runs.

Commands

List all DAGs currently available in Airflow to confirm your DAG is loaded.

Terminal

airflow dags list

Expected Output

dag_id               | description                         | schedule_interval | file

performance_tracking_example | A simple DAG to demonstrate performance tracking | @daily           | /usr/local/airflow/dags/my_dag.py

Manually trigger the DAG to run now so we can track its performance.

Terminal

airflow dags trigger performance_tracking_example

Expected Output

Created <DagRun performance_tracking_example @ 2024-06-01 12:00:00+00:00: manual__2024-06-01T12:00:00+00:00, externally triggered: True>

List all tasks in the DAG to know what tasks we will track.

Terminal

airflow tasks list performance_tracking_example

Expected Output

print_date

sleep

Check the state of the 'print_date' task for the run on 2024-06-01 to see if it succeeded.

Terminal

airflow tasks state performance_tracking_example print_date 2024-06-01

Expected Output

success

View the logs of the 'print_date' task to see its output and timing details.

Terminal

airflow tasks logs performance_tracking_example print_date 2024-06-01

Expected Output

[2024-06-01 12:00:01,000] {bash_operator.py:123} INFO - Running command: date

Thu Jun  1 12:00:01 UTC 2024

[2024-06-01 12:00:02,000] {bash_operator.py:130} INFO - Command exited with return code 0

Key Concept

If you remember nothing else from DAG performance tracking, remember: Airflow automatically records task start, end, duration, and status, which you can view via CLI or UI to monitor your workflows.

Common Mistakes

Not triggering the DAG manually after adding it, so no runs exist to track.

Without a DAG run, there is no performance data to view or analyze.

Use 'airflow dags trigger <dag_id>' to start a run and generate performance data.

Checking task state or logs for the wrong execution date.

Airflow tracks tasks by execution date; wrong date means no data or errors.

Always specify the correct execution date when querying task states or logs.

Ignoring retries and failure logs when analyzing performance.

Retries and failures affect total runtime and reliability but may be missed if not checked.

Review retry counts and failure logs to get a full picture of task performance.

Summary

Create a DAG with simple tasks to generate performance data.

Trigger the DAG manually to start a run and collect metrics.

Use Airflow CLI commands to list DAGs, tasks, check task states, and view logs for performance insights.

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