Airflow DAGの作成
使用するタイミング
- AirflowのDAGやタスクを作成またはリファクタリングするとき。
- スケジュール、キャッチアップ/バックフィル、リトライ、SLAを設定するとき。
- タスク間でデータを受け渡す（XCom）とき、または接続情報/変数を使用するとき。
- 実行中のDAGが壊れた場合の診断には使用しない（
debugging-airflow-pipelinesを使用）。
ワークフロー
- [ ] 各タスクをべき等にし、データインターバルでパラメータ化する
- [ ] 負荷の高い/インポートが多いコードはタスク内部に置き、モジュールのトップレベルに置かない
- [ ] スケジュール + catchupを意図的に設定する
- [ ] リトライ、retry_delay、SLAを設定する
- [ ] TaskFlowの戻り値または >> 演算子で依存関係を配線する
- べき等タスク —
2026-01-15インターバルのタスクは、1回実行しても再実行しても同じ結果を生成しなければなりません。datetime.now()ではなく、データインターバルを使用します。 - 重いトップレベルコードを置かない — スケジューラはすべてのDAGファイルを頻繁にパースします。モジュールレベルでのデータベース呼び出し、API呼び出し、大きなインポートはスケジューリングを遅くし、パースを壊す可能性があります。それらはタスク内部に置いてください。
- スケジュール + catchupを意図的に設定する —
catchup=Trueはstart_dateからのすべての見逃したインターバルをバックフィルします。それが望みでなければ、デフォルトはFalseにします。 - リトライとSLA — 一時的な障害は普通です。
retriesとretry_delayを設定し、遅延に対してSLA/アラートを使用します。
パターン
TaskFlow DAG、べき等でクリーンに配線:
from airflow.decorators import dag, task
import pendulum
@dag(
schedule="@daily",
start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
catchup=False,
default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5)},
tags=["orders"],
)
def orders_pipeline():
@task
def extract(data_interval_start=None):
# インターバルを使用し、now() は使わないこと。再実行が決定論的になる。
return fetch_orders(day=data_interval_start.date())
@task
def load(rows):
# パーティションを削除して挿入 -> リトライ時にべき等。
overwrite_partition("fct_orders", rows)
load(extract())
orders_pipeline()
小さなデータはXCom（戻り値）経由で渡し、大きなデータはストレージ経由で渡す — S3/GCS/ウェアハウスに書き込み、パス/キーを渡します。メガバイト単位のデータをXComで渡さないでください。
シークレットや設定には接続情報/変数を使用する（BaseHook.get_connection、Variable.get）。ハードコードされた認証情報は決して使わないでください。
よくある落とし穴
- トップレベルのAPI/DB呼び出しや重いインポート — スケジューラを遅くし、デプロイ全体でDAGのパースに失敗する可能性があります。
- タスク内での
datetime.now()の使用 — べき等性とバックフィルを壊します。data_interval_start/_endを使用してください。 - 意図しない
catchup=True— 最初のデプロイ時に履歴実行でクラスタをあふれさせます。 - XCom経由の大きなペイロード — メタデータデータベースを肥大化させます。参照を渡してください。
- 動的な
start_date（例：days_ago） — スケジュールを非決定論的にします。固定のタイムスタンプを使用してください。 - 一つの巨大タスク — 抽出/変換/ロードを分割して、リトライを細かくします。