Creating a basic DAG file in Apache Airflow - Step-by-Step CLI Walkthrough

Introduction

A DAG file in Airflow defines a workflow as a set of tasks with dependencies. Creating a basic DAG file helps you automate and schedule tasks in a clear, repeatable way.

When you want to run a data processing job every day at a specific time.

When you need to automate sending reports weekly without manual effort.

When you want to chain tasks so one runs only after another finishes successfully.

When you want to monitor and retry failed tasks automatically.

When you want to keep track of task execution history and logs.

Config File - basic_dag.py

basic_dag.py

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'basic_dag',
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval=timedelta(days=1),
    catchup=False,
)

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

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

t1 >> t2

default_args: sets default parameters for tasks like owner, retries, and start date.

DAG: defines the workflow with an ID, schedule, and description.

BashOperator: runs shell commands as tasks.

t1 >> t2: sets task order so t2 runs after t1.

Commands

Lists all DAGs currently available in Airflow to verify the new DAG is recognized.

Terminal

airflow dags list

Expected Output

basic_dag

example_bash_operator

example_python_operator

Manually triggers the basic_dag to run immediately for testing.

Terminal

airflow dags trigger basic_dag

Expected Output

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

Lists all tasks defined in the basic_dag to confirm the tasks are loaded.

Terminal

airflow tasks list basic_dag

Expected Output

print_date

sleep

Runs the print_date task for the given date without scheduling, useful for debugging.

Terminal

airflow tasks test basic_dag print_date 2024-06-01

Expected Output

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

[2024-06-01 12:00:00,100] {bash.py:130} INFO - Output:

Thu Jun  1 12:00:00 UTC 2024

[2024-06-01 12:00:00,200] {taskinstance.py:1234} INFO - Task exited with return code 0

Key Concept

If you remember nothing else from this pattern, remember: a DAG file defines tasks and their order to automate workflows in Airflow.

Common Mistakes

Not setting a start_date or setting it in the future.

Airflow will not schedule the DAG runs if the start_date is missing or after the current date.

Always set start_date to a past or current date to enable scheduling.

Forgetting to set catchup=False when you don't want backdated runs.

Airflow will try to run all missed schedules from start_date to now, causing many unexpected runs.

Set catchup=False in the DAG to run only from now on.

Not defining task dependencies, so tasks run in parallel unintentionally.

Tasks may run out of order, causing errors if one depends on another's output.

Use >> or set_upstream/set_downstream to define task order.

Summary

Create a Python file defining a DAG with tasks and their dependencies.

Use airflow CLI commands to list, trigger, and test the DAG and its tasks.

Set start_date and catchup properly to control scheduling behavior.

Practice

1. What is the main purpose of a DAG file in Apache Airflow?

easy

A. To configure the Airflow webserver settings

B. To store logs generated by tasks

C. To install Python packages required by tasks

D. To define tasks and their schedule for workflow automation

Solution

1. Step 1: Understand what a DAG file represents

   A DAG file in Airflow is a Python script that defines a Directed Acyclic Graph, which represents tasks and their dependencies.

2. Step 2: Identify the role of the DAG file

   The DAG file schedules and organizes tasks to automate workflows, not for logging, configuration, or package installation.

3. Final Answer:

   To define tasks and their schedule for workflow automation -> Option D

4. Quick Check:

   DAG file purpose = Define tasks and schedule [OK]

Hint: DAG files define workflows, not configs or logs [OK]

Common Mistakes:

• Confusing DAG file with Airflow config files
• Thinking DAG files store logs
• Assuming DAG files install packages

2. Which of the following is the correct way to import the DAG class in an Airflow DAG file?

easy

A. import DAG from airflow.models

B. import airflow.DAG

C. from airflow.models import DAG

D. from airflow import DAG

Solution

1. Step 1: Recall the correct import syntax in Python

   Python imports use 'from module import class' syntax for specific classes.

2. Step 2: Match with Airflow's module structure

   The recommended import used in official Airflow documentation is 'from airflow.models import DAG'.

3. Final Answer:

   from airflow.models import DAG -> Option C

4. Quick Check:

   Correct import syntax = from airflow.models import DAG [OK]

Hint: Use 'from airflow.models import DAG' to import DAG class [OK]

Common Mistakes:

• Using incorrect import syntax
• Trying to import DAG directly from airflow
• Using invalid Python import statements

3. Given the following DAG snippet, what will be the task_id of the BashOperator task?

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

default_args = {'start_date': datetime(2024, 1, 1)}
with DAG('example_dag', default_args=default_args, schedule_interval='@daily') as dag:
    task = BashOperator(task_id='print_date', bash_command='date')

medium

A. print_date

B. task

C. example_dag

D. bash_command

Solution

1. Step 1: Identify the task_id parameter in BashOperator

   The BashOperator is created with task_id='print_date', which names the task.

2. Step 2: Understand task_id usage

   task_id is the unique identifier for the task in the DAG, so the task's id is 'print_date'.

3. Final Answer:

   print_date -> Option A

4. Quick Check:

   task_id value = print_date [OK]

Hint: Look for task_id argument in operator constructor [OK]

Common Mistakes:

• Confusing variable name with task_id
• Using DAG id as task_id
• Mistaking bash_command as task_id

4. Identify the error in this DAG file snippet:

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG('my_dag', start_date=datetime(2024, 1, 1), schedule_interval='@daily') as dag:
    task1 = BashOperator(task_id='task1', bash_command='echo Hello')
    task2 = BashOperator(task_id='task2', bash_command='echo World')
    task1 >> task2

print('DAG created')

medium

A. No error, DAG is correctly defined

B. Missing default_args dictionary for start_date

C. Incorrect import of BashOperator

D. Using print statement inside DAG context

Solution

1. Step 1: Check imports and DAG definition

   Imports are correct, DAG is created with start_date and schedule_interval directly in constructor.

2. Step 2: Verify task creation and dependencies

   Tasks are created with valid task_ids and bash commands, and dependency is set with 'task1 >> task2'.

3. Step 3: Confirm print statement usage

   Print outside DAG context is allowed and does not affect DAG creation.

4. Final Answer:

   No error, DAG is correctly defined -> Option A

5. Quick Check:

   DAG syntax correct = No error [OK]

Hint: Check if DAG and tasks follow Airflow syntax rules [OK]

Common Mistakes:

• Assuming default_args is mandatory
• Thinking print inside DAG causes error
• Confusing import paths for operators

5. You want to create a DAG that runs a task every Monday at 7 AM starting from March 1, 2024. Which schedule_interval and start_date combination is correct in the DAG definition?

hard

A. schedule_interval='@weekly', start_date=datetime(2024, 3, 4)

B. schedule_interval='0 7 * * 1', start_date=datetime(2024, 3, 1)

C. schedule_interval='@daily', start_date=datetime(2024, 3, 1, 7, 0)

D. schedule_interval='0 7 * * 0', start_date=datetime(2024, 3, 1)

Solution

1. Step 1: Determine the cron expression for Monday 7 AM

   Cron '0 7 * * 1' runs every Monday at 7:00 AM (1=Monday).

2. Step 2: Validate start_date

   March 1, 2024 (Friday) is before the first Monday (March 4), so first run on March 4, 7 AM.

3. Step 3: Eliminate other options

   A: '@weekly'='0 0 * * 0' (Sunday midnight), wrong day/time. C: daily. D: Sunday (0).

4. Final Answer:

   schedule_interval='0 7 * * 1', start_date=datetime(2024, 3, 1) -> Option B

5. Quick Check:

   Monday 7AM cron = '0 7 * * 1' [OK]

Hint: Use cron '0 7 * * 1' for Mon 7AM [OK]

Common Mistakes:

• Setting start_date on wrong weekday
• Confusing cron day numbers (0=Sunday, 1=Monday)
• Using daily schedule for weekly task