## Instructions - Write one file per pipeline in `dags/`: `DAG(dag_id, schedule, start_date, catchup, tags)` with `pendulum` timezone-aware dates. - Keep tasks atomic + idempotent: same `logical_date` rerun produces same result; upsert loads, partitioned writes (`ds={{ ds }}`). - Choose the right operator: `PythonOperator` for logic, `BashOperator` for CLIs, `SqlOperator`/`S3` for data moves — not everything in Python. - Wait with sensors, not sleeps: `S3KeySensor`/`SqlSensor` with `poke_interval` + `timeout` + `mode='reschedule'` to free slots. - Pass small metadata via XCom only (<48 KB: table names, row counts); move real data through files/tables, never DataFrames in XCom. - Set retries deliberately: `retries=2`, `retry_delay=timedelta(min=5)`, `retry_exponential_backoff=True`; alert on `on_failure_callback`. - Gate quality: `PythonOperator` validation task (`row_count > 0`, `null_rate < 0.01`) raising `AirflowFailException` to stop bad data. - Backfill safely: `airflow dags backfill -s 2026-09-01 -e 2026-10-01 --rerun-failed-tasks`; throttle with `max_active_runs=1`. - Develop locally first: Astro CLI (`astro dev start`) or `airflow standalone`; test with `airflow tasks test `. - Version everything: DAGs + `requirements.txt` + connections in git; `catchup=False` by default, enable only for true backfills. - Complement python-automation skill: cron scripts there, orchestrated retries/SLAs/lineage/backfills here. - Never put secrets in DAG code: use Connections/Variables or env-backed secrets; log row counts, never PII samples. ## Examples - Daily sales ETL → sensor waits for CSV, Python loads to Postgres, validation task checks counts, Slack on failure. - Hourly API pull → `HttpOperator` + pagination Python task, partitioned Parquet writes, backfill a missed week safely. - ML retrain → weekly schedule, feature task → train task (XCom passes model path only), registry push, validation gate. ## Workflows 1. Sketch DAG on paper: schedule, inputs, tasks, retries, what XCom carries (small!) vs storage carries (big). 2. Scaffold `dags/.py` with defaults (`owner`, `retries`, `on_failure_callback`), one sensor + N operators + validator. 3. Test locally: `airflow dags test 2026-10-01` then per-task `tasks test`; run lint (`ruff`) on DAG files. 4. Deploy: commit, CI parses DAGs (`airflow dags show`), Astro deploy or image build; unpause and watch first 3 runs. 5. Operate: monitor grid view, backfill gaps with `--rerun-failed-tasks`, prune XCom (`xcom_cleanup`), review SLA misses weekly. ## Templates ```python # dags/sales_daily.py — sensor + load + validate + alerts, idempotent partitioned writes import pendulum from datetime import timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor from airflow.exceptions import AirflowFailException def load_sales(logical_date: str, **ctx): import pandas as pd # heavy imports inside task for worker isolation df = pd.read_csv(f"s3://landing/sales/{logical_date}.csv") df.to_parquet(f"s3://clean/sales/ds={logical_date}/data.parquet", index=False) ctx["ti"].xcom_push(key="rows", value=int(len(df))) # small metadata only def validate_sales(**ctx): rows = ctx["ti"].xcom_pull(task_ids="load", key="rows") if not rows or rows == 0: raise AirflowFailException(f"zero rows for {ctx['ds']}") with DAG(dag_id="sales_daily", schedule="@daily", start_date=pendulum.datetime(2026, 9, 1, tz="UTC"), catchup=False, max_active_runs=1, tags=["sales"]) as dag: wait = S3KeySensor(task_id="wait_csv", bucket_key="landing/sales/{{ ds }}.csv", bucket_name="my-bucket", poke_interval=300, timeout=7200, mode="reschedule") load = PythonOperator(task_id="load", python_callable=load_sales) check = PythonOperator(task_id="validate", python_callable=validate_sales) wait >> load >> check ``` ```bash # local-dev.sh — Astro or standalone test loop astro dev start # or: airflow standalone airflow dags test sales_daily 2026-10-01 airflow tasks test sales_daily load 2026-10-01 airflow dags backfill sales_daily -s 2026-09-20 -e 2026-09-25 --rerun-failed-tasks ``` ## Tools - Local: Astro CLI (`astro dev start/stop/logs`) or `pip install apache-airflow` + `airflow standalone`. - Test: `airflow dags test`, `airflow tasks test`, `airflow dags show ` for graph validation. - SARAH Run executes `airflow dags test `; logs stream beside the DAG file. ## Advanced - TaskFlow API: `@dag` + `@task` decorators auto-wire XCom; set `multiple_outputs=True` for dict returns under size limit. - Dynamic tasks: `.expand(partition=[...])` maps one task per partition; keep mapped count <500 to protect the scheduler. - Branching: `BranchPythonOperator` returns task_id; join with `trigger_rule='none_failed_min_one_success'`. - Connections: `BaseHook.get_connection('pg_shop')` in code; per-env JSON in Astro `.env`, IAM roles in production. - Performance: `max_active_tasks`, pool slots for API tasks, `depends_on_past=True` for strict sequences, deferrable operators. - Troubleshooting: zombie tasks → raise worker memory/CPU; import errors → top of DAG file imports light only; stuck sensor → reschedule mode + timeout. ## Documentation - Airflow docs: https://airflow.apache.org/docs/apache-airflow/stable/ - TaskFlow tutorial: https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html - Astro CLI: https://docs.astronomer.io/astro/cli/overview