Airflow Scheduling and Triggers

Scheduling tells Airflow when to run your DAG automatically. Triggers let you run a DAG on demand or based on an event. Understanding scheduling prevents the most common beginner mistakes around timing.

How the Scheduler Works

The Airflow Scheduler is a background process that runs continuously. It scans your DAG files, checks the current time against each DAG's schedule, and queues tasks when the schedule is due.

Scheduler Loop (repeats every few seconds):
┌──────────────────────────────────────────────┐
│  1. Read all DAG files from ~/airflow/dags/  │
│  2. Check: is any DAG scheduled to run now?  │
│  3. If yes → create a DAG Run in the DB      │
│  4. Queue the first tasks for that run       │
│  5. Workers pick up queued tasks and run them│
└──────────────────────────────────────────────┘

The schedule Parameter

Set the schedule parameter when you define your DAG. Airflow accepts three formats:

Format 1: Preset Shortcuts

Format 2: Cron Expression

A cron expression gives you precise control. It uses five fields separated by spaces:

┌──────────── minute (0–59)
│  ┌─────────── hour (0–23)
│  │  ┌────────── day of month (1–31)
│  │  │  ┌───────── month (1–12)
│  │  │  │  ┌────────── day of week (0=Sun, 6=Sat)
│  │  │  │  │
*  *  *  *  *

Format 3: timedelta (Airflow 2.4+)

Use a Python timedelta object to schedule by interval:

from datetime import timedelta

with DAG(
    dag_id="every_6_hours",
    schedule=timedelta(hours=6),
    ...
)

The Critical Concept: Execution Date vs Run Date

This confuses many beginners. Airflow schedules a DAG to run after its scheduled interval ends, not at the start.

Example: DAG with schedule="@daily" and start_date=2024-01-01

Interval              │ Execution Date │ When Airflow Actually Runs It
──────────────────────┼────────────────┼──────────────────────────────
Jan 1 00:00 → Jan 2   │ Jan 1          │ Jan 2 at 00:00
Jan 2 00:00 → Jan 3   │ Jan 2          │ Jan 3 at 00:00
Jan 3 00:00 → Jan 4   │ Jan 3          │ Jan 4 at 00:00

The execution date is the label for the interval, not the clock time when the run fires. Think of it like a newspaper: the "January 3rd edition" of the paper prints on the night of January 3rd but you read it on January 4th morning.

catchup: Handling Missed Runs

Scenario:
  start_date = 2024-01-01
  You deploy the DAG on  2024-01-10
  schedule = @daily
  catchup = True  ← default

Result: Airflow creates 9 back-filled runs for Jan 1–9 immediately.

Scenario with catchup=False:

  Only today's scheduled run starts. All past dates are ignored.

Set catchup=False in most cases unless you specifically need historical back-fills.

Triggering a DAG Manually

From the UI

Click the play ▶ button next to any DAG on the DAGs page. Airflow starts a DAG run immediately with the current timestamp as its logical date.

From the Command Line

airflow dags trigger my_dag_id

Trigger with a specific date:

airflow dags trigger my_dag_id --run-id "manual_run_001" --conf '{"key": "value"}'

From the Airflow REST API

curl -X POST "http://localhost:8080/api/v1/dags/my_dag_id/dagRuns" \
  -H "Content-Type: application/json" \
  -u "admin:admin" \
  -d '{"dag_run_id": "api_run_001"}'

Dataset-Driven Scheduling (Airflow 2.4+)

Airflow 2.4 introduced data-aware scheduling. A DAG can trigger automatically when another DAG updates a specific dataset (data output), not just on a time schedule.

Real World Analogy:
──────────────────────────────────────────────────────────────
A bakery (DAG A) bakes bread every morning.
The sandwich shop (DAG B) opens only when fresh bread arrives.
DAG B does not watch the clock — it watches for the bread.
──────────────────────────────────────────────────────────────

from airflow import Dataset
from airflow.decorators import dag, task
from datetime import datetime

sales_dataset = Dataset("s3://my-bucket/sales/daily.csv")

# DAG A: produces the dataset
@dag(start_date=datetime(2024, 1, 1), schedule="@daily")
def producer_dag():
    @task(outlets=[sales_dataset])
    def upload_sales():
        print("Uploading sales CSV to S3")
    upload_sales()

# DAG B: triggers when the dataset is updated
@dag(schedule=[sales_dataset])
def consumer_dag():
    @task
    def process_sales():
        print("Processing sales from S3")
    process_sales()

producer_dag()
consumer_dag()

Pausing and Unpausing a DAG

A paused DAG does not run on schedule. Use the toggle in the UI or the CLI:

# Pause
airflow dags pause my_dag_id

# Unpause
airflow dags unpause my_dag_id

Common Scheduling Mistakes