Composer Dag Builder

Composer DAG builder

Read references/conventions.md first. DAG id pattern is <domain>_<cadence>_<purpose>.

Default DAG skeleton

from datetime import datetime, timedelta

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import (
    BigQueryInsertJobOperator,
)
from airflow.providers.google.cloud.operators.cloud_run import (
    CloudRunExecuteJobOperator,
)
from airflow.providers.google.cloud.sensors.bigquery import (
    BigQueryTablePartitionExistenceSensor,
)

PROJECT_ID = "acme-shop-prod"
REGION = "us-central1"

default_args = {
    "owner": "data-platform",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(hours=1),
    "depends_on_past": False,
    "email_on_failure": False,
}

with DAG(
    dag_id="orders_daily_load",
    description="Loads storefront orders into mart_retail.fct_order.",
    start_date=datetime(2026, 1, 1),
    schedule="0 6 * * *",
    catchup=False,
    max_active_runs=1,
    default_args=default_args,
    tags=["retail", "daily", "owner:data-platform"],
    sla_miss_callback=None,
) as dag:

    wait_for_raw = BigQueryTablePartitionExistenceSensor(
        task_id="wait_for_raw_orders",
        project_id=PROJECT_ID,
        dataset_id="raw_storefront",
        table_id="orders",
        partition_id="{{ ds_nodash }}",
        mode="reschedule",
        poke_interval=300,
        timeout=60 * 60 * 3,
    )

    build_fct_order = BigQueryInsertJobOperator(
        task_id="build_fct_order",
        project_id=PROJECT_ID,
        location="US",
        configuration={
            "query": {
                "query": "{% include 'sql/fct_order.sql' %}",
                "useLegacySql": False,
                "priority": "BATCH",
            }
        },
        params={"run_date": "{{ ds }}"},
        sla=timedelta(hours=2),
    )

    wait_for_raw >> build_fct_order

Non-negotiables

• catchup=False unless the user explicitly wants a backfill on deploy. A DAG with a 2023 start date and catchup on will launch hundreds of runs the moment it is unpaused.
• max_active_runs=1 for anything that writes to the same table.
• Sensors in reschedule mode, never poke. Poke mode holds a worker slot for hours.
• Every sensor has a timeout. A sensor with no timeout is an outage that never pages anyone.
• Idempotent tasks. Use {{ ds }} to scope writes to one partition, and write with WRITE_TRUNCATE into that partition rather than appending.
• No credentials in the DAG. Use the environment service account or Secret Manager.
• No heavy work at parse time. Anything at module level runs every 30 seconds on every scheduler heartbeat. No API calls, no BigQuery queries, no pandas.

Choosing the operator

Prefer Airflow Datasets over ExternalTaskSensor for cross-DAG dependencies. Datasets remove the execution-date alignment problem that makes sensors flaky.

from airflow.datasets import Dataset

FCT_ORDER = Dataset("bigquery://acme-shop-prod/mart_retail/fct_order")

# producer
build_fct_order = BigQueryInsertJobOperator(..., outlets=[FCT_ORDER])

# consumer DAG
with DAG(dag_id="finance_daily_revenue", schedule=[FCT_ORDER], ...):
    ...

Deferrable operators

On a Composer environment with the triggerer enabled, use deferrable=True on long-running operators. A deferred task holds no worker slot. This is the single biggest lever on Composer worker cost.

What runs and what does not

Read references/execution-model.md. Reading DAG state, task history, and the BigQuery jobs a DAG produced are all reads and worth doing directly. Triggering a run, clearing a task, pausing a DAG, and deploying to the environment bucket are writes: emit the command.

Backfills are the case that matters most. Never trigger one. State the window, the number of runs, and the estimated bytes, then hand over the command.

Review checklist

Run through this on any DAG you are handed:

1. Does it parse fast? Any top-level API call is a finding.
2. catchup, max_active_runs, start_date sane?
3. Retries with backoff, and an execution_timeout on every task?
4. Is every task idempotent for a given {{ ds }}?
5. Sensors: reschedule mode, timeout set?
6. Are failures actionable? A task named task_1 in a DAG with no owner tag is an incident waiting to be nobody's problem.
7. Is anything in here that Dataform or a scheduled query should own instead? Airflow should orchestrate, not transform.