How to Schedule an Airflow DAG (Extract, Transform, Load)

From cron to a DAG

A bare python3 etl.py cron job runs blindly: if it fails at 2 AM nobody knows until morning, there is no retry, and there is no record of which step broke. Airflow replaces that with a DAG (Directed Acyclic Graph) - a set of tasks with explicit dependencies, so Airflow knows to run transform only after extract succeeds, and to retry on failure before alerting.

The DAG file

Create dags/daily_etl.py. The modern idiom is a with DAG(...) context manager:

import csv, json
from datetime import datetime, timedelta
from airflow import DAG

try:
    from airflow.providers.standard.operators.python import PythonOperator
except ImportError:                              # older Airflow
    from airflow.operators.python import PythonOperator

def extract():
    rows = []
    with open("/tmp/raw_orders.csv") as f:
        for row in csv.DictReader(f):
            rows.append(row)
    with open("/tmp/raw_orders.json", "w") as f:
        json.dump(rows, f)
    print(f"extracted {len(rows)} rows")

def transform():
    with open("/tmp/raw_orders.json") as f:
        rows = json.load(f)
    for r in rows:
        r["total"] = float(r["price"]) * int(r["quantity"])
    with open("/tmp/transformed_orders.json", "w") as f:
        json.dump(rows, f)

def load():
    with open("/tmp/transformed_orders.json") as f:
        rows = json.load(f)
    total = sum(r["total"] for r in rows)
    with open("/tmp/load_summary.txt", "w") as f:
        f.write(f"rows={len(rows)} total_revenue={total:.2f}\n")

default_args = {
    "owner": "data-team",
    "retries": 2,
    "retry_delay": timedelta(seconds=10),
}

with DAG(
    dag_id="daily_etl",
    schedule="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    default_args=default_args,
) as dag:
    t_extract   = PythonOperator(task_id="extract",   python_callable=extract)
    t_transform = PythonOperator(task_id="transform", python_callable=transform)
    t_load      = PythonOperator(task_id="load",      python_callable=load)

    t_extract >> t_transform >> t_load

Key parameters to understand

• catchup=False - without this, Airflow back-fills every missed daily run since start_date. Almost always set it to False on new DAGs.
• schedule="@daily" - shorthand for 0 0 * * *. You can use any cron expression or Airflow's @hourly, @weekly, @monthly aliases.
• retries + retry_delay in default_args applies to every task in the DAG; you can override per task by passing retries= directly to PythonOperator.
• >> is the dependency operator. t_extract >> t_transform >> t_load chains all three in one line. << reverses direction.

Testing without a scheduler

airflow dags test runs the DAG in-process - no scheduler, no worker, no database writes - which makes it ideal for local/CI verification:

export AIRFLOW_HOME=/tmp/airflow
export AIRFLOW__CORE__DAGS_FOLDER=/workspace/dags
export AIRFLOW__CORE__LOAD_EXAMPLES=False

airflow dags test daily_etl 2024-01-01

All three tasks run sequentially and their output files appear under /tmp/. If a task fails, the traceback shows immediately - no need to dig through the UI.