airflow-dag-patterns

Apache Airflow DAG パターン

Apache Airflow 向けの本番環境対応パターン集です。DAG 設計、オペレーター、センサー、テスト、デプロイ戦略を網羅します。

このスキルを使う場面

• Airflow によるデータパイプラインのオーケストレーションを作成するとき
• DAG の構造と依存関係を設計するとき
• カスタムオペレーターとセンサーを実装するとき
• Airflow DAG をローカルでテストするとき
• 本番環境に Airflow をセットアップするとき
• 失敗した DAG 実行をデバッグするとき

中核となる概念

1. DAG 設計原則

原則	説明
冪等性	2 回実行しても同じ結果が得られる
原子性	タスクは完全に成功するか失敗するかのいずれか
増分性	新規または変更されたデータのみを処理する
可観測性	すべてのステップでログ、メトリクス、アラートを出力する

2. タスクの依存関係

# Linear
task1 >> task2 >> task3

# Fan-out
task1 >> [task2, task3, task4]

# Fan-in
[task1, task2, task3] >> task4

# Complex
task1 >> task2 >> task4
task1 >> task3 >> task4

クイックスタート

# dags/example_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(hours=1),
}

with DAG(
    dag_id='example_etl',
    default_args=default_args,
    description='Example ETL pipeline',
    schedule='0 6 * * *',  # Daily at 6 AM
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['etl', 'example'],
    max_active_runs=1,
) as dag:

    start = EmptyOperator(task_id='start')

    def extract_data(**context):
        execution_date = context['ds']
        # Extract logic here
        return {'records': 1000}

    extract = PythonOperator(
        task_id='extract',
        python_callable=extract_data,
    )

    end = EmptyOperator(task_id='end')

    start >> extract >> end

詳細なパターンと実例

詳細なパターンのドキュメントは references/details.md にあります。上記のナビゲーション階層では不十分な場合に、そのファイルを参照してください。

ベストプラクティス

推奨事項

• TaskFlow API を使用する - コードがよりクリーンになり、XCom が自動化されます
• タイムアウトを設定する - ゾンビタスクを防ぎます
• mode='reschedule' を使用する - センサーにおいて、ワーカーを解放できます
• DAG をテストする - ユニットテストと統合テストを実施します
• 冪等なタスクにする - 安全に再試行できます

非推奨事項

• depends_on_past=True を使用しない - ボトルネックを生み出します
• 日付をハードコードしない - {{ ds }} マクロを使用します
• グローバル状態を使用しない - タスクはステートレスであるべきです
• catchup を盲目的にスキップしない - 影響を理解しておきます
• DAG ファイルに重いロジックを書かない - モジュールからインポートします