Orchestration with DAGs

DAGs and tasks

A DAG (directed acyclic graph) is a set of tasks and the dependencies between them: extract must finish before transform, transform before publish, and publish fans out to three downstream notifications. Acyclic means no loops, so the orchestrator can always determine an order. Each task is one unit of work with a clear input and output that can be retried on its own. Good task boundaries follow the pipeline stages from the previous module: extract, stage, transform, validate, publish. A single task that does everything cannot be retried partially or observed meaningfully.

from datetime import datetime, timedelta

from airflow.decorators import dag, task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator


@dag(
    dag_id="orders_daily",
    schedule="0 3 * * *",                 # 03:00 UTC, after the source's nightly close
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
    default_args={"retries": 3, "retry_delay": timedelta(minutes=5), "retry_exponential_backoff": True,
                  "execution_timeout": timedelta(minutes=30), "owner": "data-platform"},
    tags=["orders", "daily"],
)
def orders_daily():
    @task
    def extract(data_interval_start=None, data_interval_end=None) -> str:
        from pipelines.orders import extract_to_raw
        return extract_to_raw(start=data_interval_start, end=data_interval_end)   # returns the raw path

    @task
    def validate_raw(path: str) -> str:
        from pipelines.orders import check_raw
        check_raw(path)                     # raises on empty file or schema drift
        return path

    transform = SQLExecuteQueryOperator(
        task_id="transform",
        conn_id="warehouse",
        sql="sql/orders_transform.sql",   # uses {{ data_interval_start }} for the partition
    )

    @task
    def publish_metrics(**context):
        from pipelines.orders import emit_run_metrics
        emit_run_metrics(run_id=context["run_id"])

    validate_raw(extract()) >> transform >> publish_metrics()


orders_daily()

Notice what the DAG does not do: it does not contain the transformation logic, it does not compute dates from the clock, and it does not hold data in memory between tasks. Tasks pass small references (a path, a partition key), never DataFrames. The heavy lifting runs in the warehouse, in a Spark job, in a container — anywhere but the orchestrator's own workers, which exist to coordinate.

Scheduling semantics: the logical date

The most misunderstood part of any orchestrator is when a run happens versus which data it covers. A daily DAG scheduled at 03:00 with a data interval of one day runs at 03:00 on March 2nd to process the interval March 1st 00:00 to March 2nd 00:00. The run's logical date (data_interval_start) is March 1st, even though the clock says March 2nd. Every task should derive its target partition from the interval it is given, never from datetime.now(). This is what makes reruns and backfills correct: rerunning the March 1st run in April still processes March 1st.

Failure handling without a human

Tasks fail for transient reasons (a network blip, a warehouse queue) and for real ones (bad data, a bug). Retries with backoff absorb the first kind; a task that is idempotent can be retried safely, which is why idempotency kept coming up. Timeouts stop a hung task from blocking the day. When retries are exhausted, the failure should reach a human with context: which DAG, which interval, which task, the log, and a runbook link — through an on-failure callback to the alerting system, not through someone checking the UI. SLAs (or freshness checks on the output tables) catch the failure mode retries do not: the run that succeeds late, or never starts.

from datetime import timedelta


def page_on_failure(context):
    from pipelines.alerts import send_alert
    ti = context["task_instance"]
    send_alert(
        severity="high",
        title=f"{ti.dag_id}.{ti.task_id} failed for {context['data_interval_start']:%Y-%m-%d}",
        body=f"attempt {ti.try_number} of {ti.max_tries + 1}; log: {ti.log_url}",
        runbook="https://runbooks.internal/data/orders_daily",
    )


default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "execution_timeout": timedelta(minutes=30),
    "on_failure_callback": page_on_failure,
    "sla": timedelta(hours=2),              # task should finish within 2 h of the interval's end
}

Decide per task whether downstream should run when it fails. The default (skip downstream) is right for a hard dependency; for optional enrichment a trigger rule such as all_done lets the pipeline continue and flag the gap. And separate "the data is bad" from "the task crashed": a validation task that fails on bad data should not be retried three times; it should stop the run and alert with the validation report.

Waiting for inputs: sensors and datasets

Pipelines wait: for a file to land, for an upstream DAG to finish, for a partition to appear in the warehouse. A sensor is a task that polls for a condition and succeeds when it holds, with a timeout so it does not wait forever. Use deferrable (or reschedule mode) sensors so a waiting task does not occupy a worker slot. Better than polling, where the platform supports it, is data-aware scheduling: a DAG declares that it produces a dataset, downstream DAGs declare they consume it, and the scheduler starts the consumer when the producer updates the dataset — no cron guessing about when upstream will be done.

from datetime import datetime, timedelta

from airflow.datasets import Dataset
from airflow.decorators import dag, task
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

ORDERS_SILVER = Dataset("s3://acme-lake/silver/orders/")


@dag(dag_id="partner_file_ingest", schedule="0 4 * * *", start_date=datetime(2026, 1, 1), catchup=False)
def partner_file_ingest():
    wait_for_file = S3KeySensor(
        task_id="wait_for_partner_file",
        bucket_name="acme-inbound",
        bucket_key="partner/orders_{{ data_interval_start | ds_nodash }}.csv",
        deferrable=True,                       # frees the worker while waiting
        timeout=6 * 60 * 60,                   # give up after 6 hours and alert
        poke_interval=300,
    )

    @task(outlets=[ORDERS_SILVER])
    def load_to_silver():
        from pipelines.partner import load
        load()

    wait_for_file >> load_to_silver()


@dag(dag_id="orders_reporting", schedule=[ORDERS_SILVER], start_date=datetime(2026, 1, 1), catchup=False)
def orders_reporting():
    @task
    def build_reports():
        from pipelines.reports import build
        build()

    build_reports()


partner_file_ingest()
orders_reporting()

Backfills and keeping DAGs boring

A backfill asks the orchestrator to run a DAG for a range of past intervals. Because each run gets its own logical date and every task is idempotent, the backfill is just many correct runs. Limit its concurrency so it does not starve the daily schedule or overwhelm the warehouse; clear only the tasks that need rerunning rather than the whole DAG; and watch the first few runs before letting the rest go. Backfills that need special handling (a different code path, skipped notifications) should get an explicit parameter, not a guess based on how old the interval is.

# rebuild a range with at most 2 intervals in flight, without triggering downstream datasets
airflow dags backfill orders_daily --start-date 2026-02-01 --end-date 2026-02-28 --max-active-runs 2

# rerun just the transform (and what depends on it) for one interval
airflow tasks clear orders_daily --task-regex '^transform$' --downstream \
  --start-date 2026-03-01 --end-date 2026-03-01 --yes

# run one task for one interval locally, no scheduler needed
airflow tasks test orders_daily extract 2026-03-01

• Thin DAG files: imports inside tasks, logic in a package with unit tests; the DAG file is parsed constantly and must be fast and side-effect free.
• No top-level work: no database calls or API requests at import time; they run every parse.
• Small payloads between tasks: paths and keys, not data; the metadata database is not a data store.
• One DAG per pipeline, one owner, tagged: hundreds of tasks in one DAG is hard to operate; hundreds of copy-paste DAGs is worse. Generate similar DAGs from configuration.
• Version the DAGs with the pipeline code and deploy through CI like any other code.

The orchestrator's job is to make the pipeline's behaviour predictable and visible. When the DAG is thin, the tasks idempotent and the schedule data-aware, the interesting logic lives where it is tested, and the orchestrator becomes what it should be: the boring, reliable clock and ledger of the platform.