# Composer Dag Builder

> Write and review Airflow DAGs for Cloud Composer using GCP-native operators. Use when the user mentions Airflow, Composer, a DAG, a scheduled pipeline, backfills, task dependencies, sensors, SLAs, or asks how to orchestrate BigQuery, Dataform, Cloud Run, or Dataflow work on a schedule.

- Skill: `rk-chavali/composer-dag-builder` (Agent Skill)
- Install (CLI): `npx skillmds@latest add rk-chavali/composer-dag-builder`
- Raw SKILL.md: https://api.skillmd.com/api/skills/rk-chavali/composer-dag-builder/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: rk-chavali (https://skillmd.com/u/rk-chavali)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/rk-chavali/composer-dag-builder

---


# Composer DAG builder

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

## Default DAG skeleton

```python
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

| Work | Operator | Notes |
| --- | --- | --- |
| SQL in BigQuery | `BigQueryInsertJobOperator` | the only BQ operator worth using; the older ones are deprecated |
| Dataform run | `DataformCreateWorkflowInvocationOperator` | pair with the sensor for completion |
| Container job | `CloudRunExecuteJobOperator` | preferred over KubernetesPodOperator on Composer |
| File to table | `GCSToBigQueryOperator` | or skip Airflow and use an external table |
| Wait on a partition | `BigQueryTablePartitionExistenceSensor` | reschedule mode |
| Wait on another DAG | `ExternalTaskSensor` | prefer Datasets over this |
| Arbitrary Python | `PythonOperator` | last resort, and never for heavy compute |

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

```python
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.