{"id":"airflow-dag-patterns","name":"airflow-dag-patterns","summary":"運用者、センサー、テスト、展開のためのベストプラクティスを備えた本番のApache Airflow DAGを構築しましょう。","body":"# Apache Airflow DAG Patterns\n\nProduction-ready patterns for Apache Airflow including DAG design, operators, sensors, testing, and deployment strategies.\n\n## When to Use This Skill\n\n- Creating data pipeline orchestration with Airflow\n- Designing DAG structures and dependencies\n- Implementing custom operators and sensors\n- Testing Airflow DAGs locally\n- Setting up Airflow in production\n- Debugging failed DAG runs\n\n## Core Concepts\n\n### 1. DAG Design Principles\n\n| Principle       | Description                         |\n| --------------- | ----------------------------------- |\n| **Idempotent**  | Running twice produces same result  |\n| **Atomic**      | Tasks succeed or fail completely    |\n| **Incremental** | Process only new/changed data       |\n| **Observable**  | Logs, metrics, alerts at every step |\n\n### 2. Task Dependencies\n\n```python\n# Linear\ntask1 >> task2 >> task3\n\n# Fan-out\ntask1 >> [task2, task3, task4]\n\n# Fan-in\n[task1, task2, task3] >> task4\n\n# Complex\ntask1 >> task2 >> task4\ntask1 >> task3 >> task4\n```\n\n## Quick Start\n\n```python\n# dags/example_dag.py\nfrom datetime import datetime, timedelta\nfrom airflow import DAG\nfrom airflow.operators.python import PythonOperator\nfrom airflow.operators.empty import EmptyOperator\n\ndefault_args = {\n    'owner': 'data-team',\n    'depends_on_past': False,\n    'email_on_failure': True,\n    'email_on_retry': False,\n    'retries': 3,\n    'retry_delay': timedelta(minutes=5),\n    'retry_exponential_backoff': True,\n    'max_retry_delay': timedelta(hours=1),\n}\n\nwith DAG(\n    dag_id='example_etl',\n    default_args=default_args,\n    description='Example ETL pipeline',\n    schedule='0 6 * * *',  # Daily at 6 AM\n    start_date=datetime(2024, 1, 1),\n    catchup=False,\n    tags=['etl', 'example'],\n    max_active_runs=1,\n) as dag:\n\n    start = EmptyOperator(task_id='start')\n\n    def extract_data(**context):\n        execution_date = context['ds']\n        # Extract logic here\n        return {'records': 1000}\n\n    extract = PythonOperator(\n        task_id='extract',\n        python_callable=extract_data,\n    )\n\n    end = EmptyOperator(task_id='end')\n\n    start >> extract >> end\n```\n\n## Detailed patterns and worked examples\n\nDetailed pattern documentation lives in `references/details.md`. Read that file when the navigation tier above is insufficient.\n\n## Best Practices\n\n### Do's\n\n- **Use TaskFlow API** - Cleaner code, automatic XCom\n- **Set timeouts** - Prevent zombie tasks\n- **Use `mode='reschedule'`** - For sensors, free up workers\n- **Test DAGs** - Unit tests and integration tests\n- **Idempotent tasks** - Safe to retry\n\n### Don'ts\n\n- **Don't use `depends_on_past=True`** - Creates bottlenecks\n- **Don't hardcode dates** - Use `{{ ds }}` macros\n- **Don't use global state** - Tasks should be stateless\n- **Don't skip catchup blindly** - Understand implications\n- **Don't put heavy logic in DAG file** - Import from modules","author":"@wshobson","ownerProfile":null,"authorContacts":null,"sourceUrl":"https://github.com/wshobson/agents/tree/main/plugins/data-engineering/skills/airflow-dag-patterns","license":"MIT","category":"document","lang":"en","tokens":719,"stars":0,"calls30d":1,"claimed":false,"visibility":"public","origin":"crawler","version":"0.1.0","createdAt":"2026-08-22","updatedAt":"2026-08-22","files":[{"path":"references/details.md","size":11406,"sha256":"7ee04b0d4ae58b49a1977d8b41fcb2173e026e20b97330ec963eebbd5c45bab2"}],"requires":{"mcp":[],"tools":[]},"safety":{"flags":[],"scannedAt":"2026-08-22","hasScripts":false,"networkEndpoints":["api.example.com"]}}