Airflow basics: your first DAG, tasks, dependencies, schedules and how to test it locally

A DAG is a Python file that describes tasks and the order they run in. Airflow reads the file, schedules runs, and executes each task when its upstream tasks succeed. Most first-DAG problems are about scheduling, not code.

#A first DAG

dags/daily_sales.py:

from datetime import datetime, timedelta
from airflow.decorators import dag, task     # Airflow 3: from airflow.sdk import dag, task

@dag(
    schedule="0 6 * * *",                       # 06:00 every day
    start_date=datetime(2026, 10, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
    tags=["sales", "example"],
)
def daily_sales():

    @task
    def extract() -> list[dict]:
        return [{"id": 1, "amount": 120.5}, {"id": 2, "amount": 80.0}]

    @task
    def transform(rows: list[dict]) -> float:
        return sum(r["amount"] for r in rows)

    @task
    def load(total: float):
        print(f"Daily total: {total}")

    load(transform(extract()))

daily_sales()

The function calls define dependencies: extract runs first, then transform, then load.

#Schedules and dates that confuse everyone

• start_date must be a fixed date in the past. Never datetime.now(); it changes every time the file is parsed.
• A run covers a data interval. The run for the interval 2026-10-28 06:00 to 2026-10-29 06:00 starts at the end of that interval. This is why "my DAG did not run today" is often "it will run at the end of today's interval".
• catchup=False stops Airflow from creating a run for every missed interval since start_date. Leave it off unless you really want backfills.

Use the logical date in tasks instead of "today":

@task
def extract(**context):
    day = context["ds"]          # e.g. "2026-10-28"
    ...

#Passing data between tasks

TaskFlow passes return values through XCom. Keep them small (IDs, paths, counts). Large data belongs in storage; pass the path instead.

Warning

Returning a DataFrame or a big list from a task stores it in the metadata database. It slows Airflow down and can fail outright. Write data to object storage or a table and return the location.

#Classic operators

You can mix TaskFlow with operators:

from airflow.operators.bash import BashOperator   # Airflow 3: airflow.providers.standard.operators.bash

cleanup = BashOperator(task_id="cleanup", bash_command="echo cleaning {{ ds }}")
load_task = load(transform(extract()))
load_task >> cleanup

>> means "runs before". Jinja templates like {{ ds }} are rendered at run time.

#Test before you deploy

Check the file imports without errors:

python dags/daily_sales.py
airflow dags list-import-errors

Run the whole DAG once, in-process, without the scheduler:

airflow dags test daily_sales 2026-10-28

Run a single task:

airflow tasks test daily_sales transform 2026-10-28

Both print task logs to your terminal and do not need a running scheduler.

#Trigger and watch

airflow dags unpause daily_sales
airflow dags trigger daily_sales
airflow dags list-runs -d daily_sales

In the UI, the Grid view shows each run and task colour-coded; click a task to read its log.

#Good habits from day one

@task(execution_timeout=timedelta(minutes=30))
def long_job(): ...

#Prevent it next time

• Add a CI step that imports every DAG file and fails on import errors.
• Always set catchup explicitly.
• Write tasks so they can run twice for the same date without duplicating data.