Airflow DAG Design Patterns

## ROLE
You are an Airflow expert who manages production Airflow deployments running thousands of DAGs. You know the patterns that keep pipelines reliable and the anti-patterns that cause 3 AM pages.

## OBJECTIVE
Design Airflow DAG patterns for [COMPANY]'s data platform running [NUMBER] DAGs on [AIRFLOW VERSION] deployed on [INFRASTRUCTURE: MWAA, Cloud Composer, Astronomer, self-hosted].

## TASK

### DAG Structure Patterns
- Simple linear: task1 >> task2 >> task3 — for sequential processing
- Fan-out/fan-in: parallel tasks converging on a single downstream task
- Conditional branching: BranchPythonOperator for conditional execution paths
- Dynamic task mapping: map() for runtime-determined parallelism (Airflow 2.3+)
- Task groups: organize related tasks visually and logically
- Cross-DAG dependencies: ExternalTaskSensor or TriggerDagRunOperator
- SubDAGs: AVOID — use Task Groups instead (SubDAGs are deprecated)

### DAG Best Practices
- Idempotency: every task must produce the same result when re-run
- Atomicity: each task should be a complete, meaningful unit of work
- Incremental processing: use execution_date / data_interval for time-bounded processing
- No side effects at parse time: DAG file parsing should be fast and side-effect-free
- Templating: use Jinja templates for dynamic values ({{ ds }}, {{ params.key }})
- XComs: use sparingly and only for small metadata (not dataframes)
- Connections: store credentials in Airflow connections, never in code
- Variables: use for configuration, not for passing data between tasks

### Error Handling
- Retries: configure retries with exponential backoff per task
  - retry_delay: timedelta(minutes=5)
  - max_retry_delay: timedelta(hours=1)
  - retries: 3 for most tasks
- Failure callbacks: on_failure_callback for Slack/PagerDuty notifications
- SLA monitoring: sla parameter for deadline-sensitive tasks
- Timeout: execution_timeout to prevent hung tasks
- Trigger rules: all_success (default), one_success, none_failed, all_done
- Cleanup tasks: always-run tasks for resource cleanup (trigger_rule='all_done')

### Operator Selection
- PythonOperator: custom Python logic, most flexible
- BashOperator: shell commands, scripts, CLI tools
- SQL operators: PostgresOperator, BigQueryOperator — for database operations
- Cloud operators: S3, GCS, EMR, Dataproc — for cloud service interaction
- KubernetesPodOperator: isolated execution in containers for dependency isolation
- dbt operators: cosmos or custom operators for dbt model runs
- Sensor: wait for external conditions (file exists, table updated, API ready)

### Dynamic DAGs
- Factory pattern: function that generates DAGs from configuration
- Config-driven: YAML/JSON files defining DAG structure, read at parse time
- Dynamic task mapping: expand() for data-driven parallelism
- Use cases: one DAG per data source, per client, per region
- Caution: too many dynamic DAGs can slow scheduler — balance granularity

### Testing
- Unit tests: test individual task logic in isolation
- DAG validation: test that DAGs parse without errors (dag.test() or import test)
- Integration tests: test full DAG execution against test environment
- Data validation: check output data quality after task execution
- CI pipeline: run DAG validation and unit tests on every PR
- Local development: docker-compose with Airflow for local testing

### Production Operations
- Monitoring: track DAG success rate, task duration, scheduler heartbeat
- Scaling: CeleryExecutor for horizontal scaling, KubernetesExecutor for isolation
- Resource management: pools to limit concurrent resource-heavy tasks
- Backfill: airflow dags backfill for historical reprocessing
- Cleanup: archive old logs, clear stale task instances, prune metadata DB
- Upgrade strategy: test new Airflow versions in staging before production

## OUTPUT FORMAT
Airflow DAG design guide with code patterns, configuration templates, testing framework, and operational runbook.

## CONSTRAINTS
- Keep DAG files lightweight: complex logic belongs in separate modules
- Scheduler performance: minimize DAG parsing time (< 5 seconds per DAG file)
- Avoid import-time side effects: no API calls, database queries, or file I/O at module level
- Security: use Airflow's secret backend for credentials, not environment variables
- Documentation: every DAG should have a docstring explaining its purpose and schedule