Writing Your First DAG

A DAG is a Python file that defines a workflow. Airflow's scheduler parses every .py file in the dags/ folder, and any file that creates a DAG object becomes a workflow in the UI.

This article writes a minimal one from scratch, one piece at a time.

Where the file goes#

project/  dags/    orders_pipeline.py     <- the scheduler parses this  transform/    raw_to_staging.py      <- the function the DAG calls

In the Docker stack the dags/ directory is bind-mounted into the scheduler and webserver, so saving a file makes the scheduler pick it up within seconds — no rebuild, no restart.

The default_args dictionary#

default_args is a dictionary of settings applied to every task in the DAG, unless a task overrides them:

from datetime import datetime, timedelta
default_args = {    "owner": "data-team",    "retries": 2,    "retry_delay": timedelta(minutes=5),}

• owner — a label shown in the UI; useful for filtering when many teams share an Airflow.
• retries — how many times a failed task is retried before it is marked failed. A pipeline with no retry policy fails on the first transient error (a network blip, a locked table). 2 is a sensible default.
• retry_delay — how long to wait between attempts. timedelta(minutes=5) gives a transient problem time to clear.

Other common keys: email_on_failure, execution_timeout, depends_on_past.

The with DAG(...) block#

The DAG itself is created with a context manager. Every task defined inside the with block is automatically attached to that DAG:

from airflow import DAG
with DAG(    dag_id="orders_pipeline",    default_args=default_args,    description="Load raw orders and build the staging table",    start_date=datetime(2026, 1, 1),    schedule="@hourly",    catchup=False,    max_active_runs=1,    tags=["orders", "etl"],) as dag:    ...

• dag_id — the unique name. It is what you see in the UI and use on the CLI, so make it descriptive and stable; renaming it creates a new DAG and orphans the old history.
• start_date — the first data interval the DAG covers. Use a fixed past date, never datetime.now() (which would move on every parse and confuse the scheduler).
• schedule — how often a run is created. @hourly here; the full set of options is in Scheduling and Backfill.
• catchup=False — when the DAG is first enabled (or re-enabled after being paused), do not create a run for every interval between start_date and now. With catchup=True and a start_date months ago, enabling the DAG triggers hundreds of runs at once. Almost always set this to False and backfill deliberately when you need to.
• max_active_runs=1 — do not let two runs of this DAG execute at the same time. Important when runs share a target table.
• tags — UI filtering labels.

The first task: PythonOperator#

An operator is a template for a task. PythonOperator runs a Python function:

from airflow.operators.python import PythonOperator
from transform.raw_to_staging import load_raw_orders_to_staging
with DAG(...) as dag:    load_staging = PythonOperator(        task_id="load_raw_orders_to_staging",        python_callable=load_raw_orders_to_staging,    )

• task_id — the task's name within the DAG. Unique per DAG, stable over time, shown in the graph.
• python_callable — the function to run. It is imported from another module; it contains no Airflow code and can be run and tested on its own. This separation is the point (see Orchestration and Airflow Fundamentals).

A complete minimal DAG#

from datetime import datetime, timedelta
from airflow import DAGfrom airflow.operators.python import PythonOperator
from transform.raw_to_staging import load_raw_orders_to_staging
default_args = {    "owner": "data-team",    "retries": 2,    "retry_delay": timedelta(minutes=5),}
with DAG(    dag_id="orders_pipeline",    default_args=default_args,    description="Load raw orders into the staging table every hour",    start_date=datetime(2026, 1, 1),    schedule="@hourly",    catchup=False,    max_active_runs=1,    tags=["orders", "etl"],) as dag:    load_staging = PythonOperator(        task_id="load_raw_orders_to_staging",        python_callable=load_raw_orders_to_staging,    )

load_raw_orders_to_staging is a plain function elsewhere — the kind covered in Working with ClickHouse from Python and Batch and Incremental Processing.

Testing it without the scheduler#

Two CLI commands run DAG code directly, without waiting for a schedule:

# does the file parse, and does the DAG have the tasks you expect?airflow dags test orders_pipeline 2026-03-01
# run a single task for a given logical date, printing its logsairflow tasks test orders_pipeline load_raw_orders_to_staging 2026-03-01

airflow tasks test does not touch the metadata database or trigger downstream tasks — it just executes the one task, which makes it the fastest way to debug a python_callable.

Common mistakes#

start_date=datetime.now()#

now() is re-evaluated on every parse, so the DAG's start keeps moving and the scheduler never settles on an interval to run. Use a fixed date.

catchup=True with an old start_date#

Enabling the DAG then creates a run for every missed interval at once. Keep catchup=False and backfill on purpose.

A mutable object shared in default_args#

default_args is shared by every task. Putting a mutable value there that a task modifies leaks state between tasks. Keep it to simple, immutable settings.

Renaming dag_id#

The dag_id keys all of the DAG's history. Renaming it creates a fresh DAG with no past runs and leaves the old one as a stale entry.

No retries#

Without a retry policy the first transient failure fails the run. Set retries and retry_delay in default_args.

Quick reference#

airflow dags listairflow dags test <dag_id> <date>airflow tasks test <dag_id> <task_id> <date>

See also#