Unknown-333/authoring-airflow-dags

Authoring Airflow DAGs

When to use

• Creating or refactoring Airflow DAGs and tasks.
• Configuring schedules, catchup/backfill, retries, and SLAs.
• Passing data between tasks (XCom) or using connections/variables.
• Do NOT use for diagnosing a broken running DAG (use debugging-airflow-pipelines).

Workflow

- [ ] Make each task idempotent and parameterized by the data interval
- [ ] Keep expensive/import-heavy code inside tasks, not at module top level
- [ ] Set schedule + catchup deliberately
- [ ] Configure retries, retry_delay, and SLAs
- [ ] Wire dependencies via TaskFlow return values or >> operators

1. Idempotent tasks — a task for the 2026-01-15 interval must produce the same result whether it runs once or is re-run. Use the data interval, not datetime.now().
2. No heavy top-level code — the scheduler parses every DAG file frequently; database calls, API calls, or big imports at module level slow scheduling and can break parsing. Put them inside tasks.
3. Schedule + catchup on purpose — catchup=True backfills every missed interval from start_date; default to False unless you want that.
4. Retries and SLAs — transient failures are normal; set retries and retry_delay; use SLAs/alerts for lateness.

Patterns

TaskFlow DAG, idempotent and cleanly wired:

from airflow.decorators import dag, task
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5)},
    tags=["orders"],
)
def orders_pipeline():

    @task
    def extract(data_interval_start=None):
        # Use the interval, not now(), so re-runs are deterministic.
        return fetch_orders(day=data_interval_start.date())

    @task
    def load(rows):
        # Delete-insert the partition -> idempotent on retry.
        overwrite_partition("fct_orders", rows)

    load(extract())

orders_pipeline()

Pass small data via XCom (return values); pass large data via storage — write to S3/GCS/warehouse and pass the path/key, never megabytes through XCom.

Use connections/variables for secrets and config (BaseHook.get_connection, Variable.get), never hard-coded credentials.

Common pitfalls

• Top-level API/DB calls or heavy imports — slow the scheduler and can fail DAG parsing across the whole deployment.
• datetime.now() inside tasks — breaks idempotency and backfills; use data_interval_start/_end.
• catchup=True unintentionally — floods the cluster with historical runs on first deploy.
• Large payloads through XCom — bloats the metadata DB; pass references.
• Dynamic start_date (e.g. days_ago) — makes schedules nondeterministic; use a fixed timestamp.
• One monster task — split extract/transform/load so retries are granular.

References

• Scheduling, catchup, and backfill reference