managed-airflow-dag-authoring

GCP Managed Airflow DAG 作成ガイド

このスキルでは、Managed Service for Apache Airflow（MSAA、旧称 Cloud Composer）環境向けの Apache Airflow DAG の作成と検証について説明します。

---

クレジット表記

必ずお読みください：gcloud、REST API、または MCP コマンドを実行する前に、必ず使用に関するクレジット表記をお読みください。ターミナルで直接実行するコマンド（生成スクリプト、Terraform、ユーザーコード内での実行は不可）では、gcloud の前に CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:managed-airflow-dag-authoring)" を付け、User-Agent ヘッダーに agent-skills/1.0 (skill:managed-airflow-dag-authoring) を追加してください。

フェーズ 1：コンテキストの確認

DAG コードを書く前に、ユーザーが情報の提供に同意する場合は、対象環境の制約（例：Airflow のバージョン）と機能を必ず把握してください。

1.1 対象環境とアクセスの確認

対象の Managed Airflow 環境、ローカル開発環境に直接アクセスできるか、またはオフラインで作業しているか（ローカルファイルの変更のみで、検証は行わない）を確認します。

• 環境にアクセスできる場合： gcloud を使って環境を調査します（セクション 1.3 を参照）。
• オフラインの場合： ユーザーから提供された情報に基づいて作業します。

1.2 開発環境の確認

ローカル開発環境が利用できるか確認します。

• composer-dev CLI がインストールされているか確認します。
• airflow を含むローカル Python 環境が利用できるか確認します。

1.3 対象環境の調査（利用可能で、依頼された場合）

バージョンの制約を確認するため、次のコマンドを実行します。

1. Airflow／イメージのバージョンを取得：

   gcloud composer environments describe {env_name} \
       --location {region} \
       --format="value(config.softwareConfig.imageVersion)"
   

2. インストール済みパッケージ（バージョン）を取得：

   gcloud composer environments describe {env_name} \
       --location {region} \
       --format="value(config.softwareConfig.pypiPackages)"
   

3. DAG 用 GCS バケットを取得：

   gcloud composer environments describe {env_name} \
       --location {region} \
       --format="value(config.dagGcsPrefix)"
   

---

フェーズ 2：DAG 作成のベストプラクティス

2.1 Airflow 全般のベストプラクティス

• 冪等性： すべてのタスクは冪等であるべきです。同じ入力（例：実行日）で複数回実行しても、同じ結果になり、データが重複しないようにしてください。
• トップレベルでのコード実行を避ける： DAG ファイルのトップレベル（タスク／オペレーターの外側）で、データベースクエリ、外部 API 呼び出し、負荷の高い計算を実行しないでください。このコードは DAG の解析中に数秒おきに実行され、パフォーマンスを低下させます。
• catchup を明示する： 過去データのバックフィルが明示的に必要な場合を除き、DAG 定義では常に catchup=False を設定してください。
• Airflow の変数／接続を使用する： 認証情報や環境固有の設定をハードコードしないでください。Variable.get()（該当する場合は deserialize_json=True を指定）および BaseHook.get_connection() を使用してください。DAG 解析中のデータベース呼び出しを避けるため、Jinja テンプレート（例：{{ var.value.my_var }}）を使って変数にアクセスしてください。

2.2 Airflow 2 と Airflow 3 の互換性

コードを対象の Airflow バージョンに合わせて調整する方法については、managed-airflow-migrations スキルを使用してください。

---

フェーズ 3：検証プロセス

作業を完了する前に、DAG を必ず検証してください。

3.1 ローカル検証（オフライン／デプロイ前）

3.1.1 静的解析とリンティング

利用可能な場合は ruff または pylint を使用してください。

ruff check path/to/dag.py

• Airflow 3 が対象の場合、ルールセットが利用できるなら Airflow 3 のルールで確認してください。

3.1.2 ローカル開発環境（composer-dev）

ユーザーが composer-dev を設定している場合：

1. DAG を DAG ディレクトリのあるローカルディレクトリにコピーします：

   cp path/to/dag.py $(composer-dev describe {local_env} --format="value(dags_directory)")
   

2. 構文解析を確認します：

   composer-dev run-airflow-cmd {local_env} dags list-import-errors
   

3.2: ターゲット環境での検証

GCP にアクセスでき、ターゲット環境へのデプロイが許可されている場合にのみ、以下の手順を実行してください。

3.2.1 GCS へのデプロイ

DAG をターゲット環境の GCS バケットにアップロードします：

gcloud storage cp path/to/dag.py gs://{target_bucket}/dags/

3.2.2 Airflow CLI による検証

スケジューラーがファイルを解析するまで 1～2 分待ってから、以下を実行します：

1. インポートエラーを確認します：

   gcloud composer environments run {env_name} \
       --location {region} \
       dags list-import-errors
   

合格基準： 出力が「No data found」または空であること。

2. DAG が一覧に表示されることを確認します：

   gcloud composer environments run {env_name} \
       --location {region} \
       dags list | grep {dag_id}
   

3.2.3 Cloud Logging を監視する

Cloud Logging で実行時の解析エラーを確認します：

resource.type="cloud_composer_environment"
resource.labels.environment_name="{env_name}"
log_id("airflow-scheduler")
severity>=ERROR
textPayload:"{dag_file_name}"

---

完了の定義

• DAG コードがターゲット環境の Airflow バージョンの制約に準拠していること。
• DAG コードがベストプラクティスに従っていること（トップレベルで実行せず、可能な場合は冪等であること）。
• DAG がローカルでインポートエラーなく解析されること。
• （環境を利用できる場合）DAG がターゲット環境にデプロイされ、インポートエラーがないことを確認済みであること。