Cloud Composer (Apache Airflow)

解決する課題

依存関係のある多数のジョブを「A が終わったら B、B と C が揃ったら D」のように順序立てて回すデータパイプラインを、自前の cron や常駐スクリプトで組むと、再実行・リトライ・進行状況の把握が次第に破綻します。Cloud Composer は Apache Airflow をマネージドで提供し、こうしたワークフローを Python で定義した DAG として管理させます。

• 複数ステップの バッチ処理やデータパイプラインを、依存関係込みで確実に順序実行したい
• Airflow を使いたいが、スケジューラやワーカーの クラスタ運用・パッチ適用・スケールを自前で抱えたくない
• 失敗したタスクの リトライ・再実行・過去日分のバックフィルを仕組みとして任せたい
• BigQuery・Dataflow・Dataproc・Cloud Storage など 多数のサービスをまたぐ連携を、共通の枠組みで記述したい
• パイプラインの 実行状況・依存グラフ・ログを UI で可視化し、障害箇所をすぐ特定したい

主要概念と用語

• Apache Airflow: ワークフローを Python コードで定義し、スケジュール実行・監視する OSS。Cloud Composer はこれをマネージドに提供する
• 環境（Environment）: Airflow 一式（スケジューラ・ワーカー・ウェブサーバ・メタデータ DB）をまとめた実行基盤の単位。プロジェクト内に複数作成できる
• DAG（有向非巡回グラフ）: タスクとその依存関係を表したワークフロー定義。循環しない（戻らない）一方向のグラフとして組む
• タスク（Task）とオペレータ（Operator）: DAG を構成する処理の単位がタスク。オペレータは「何をするか」のテンプレート（例: BigQuery 実行、HTTP 呼び出し）で、これをインスタンス化してタスクにする
• スケジューラ（Scheduler）: DAG を解析し、実行すべきタスクを依存関係に従ってワーカーへ割り当てるコンポーネント
• ワーカー（Worker）: 実際にタスクを実行するプロセス。負荷に応じてスケールする
• DAG バッグ（DAGs フォルダ）: DAG 定義の Python ファイルを置く Cloud Storage バケット上の場所。ここへ配置すると環境に取り込まれる
• メタデータ DB: 実行履歴・タスク状態・スケジュール情報を保持するデータベース。Airflow の状態管理の中核
• XCom: タスク間で小さな値を受け渡すための仕組み

仕様・制限・クォータ

• ワークフローは Python で記述した DAG として定義する。分岐・並列・リトライ・センサー（条件待ち）・バックフィルなどを表現できる
• 環境は 常時稼働する基盤であり、スケジューラ・ワーカー・ウェブサーバ・メタデータ DB が裏で動き続ける。単発の関数を都度起動するサーバーレス型とは性質が異なる
• 環境は GKE 上のマネージド構成として作られ、ワーカー数や環境サイズに応じてリソースが割り当てられる。負荷に応じたワーカーのオートスケールに対応する版もある
• 1 環境あたりの DAG 数・タスク同時実行数・ワーカー数などには構成や環境サイズに依存する上限があり、必要に応じて環境サイズやワーカー設定を調整する
• Airflow にはメジャーバージョン系統が複数あり、利用できるバージョンや機能は Composer のメジャー世代に依存する。バージョンアップは互換性を確認のうえ計画的に行う
• DAG ファイルは DAGs フォルダ（Cloud Storage） に配置し、スケジューラが定期的に解析して取り込む。反映には解析間隔ぶんの時間差がある
• 具体的なバージョン番号・上限値・解析間隔などの数値は変動するため、最新は公式ドキュメントで確認すること

内部の仕組み

環境を作成すると、Cloud Composer は Airflow の各コンポーネント（スケジューラ・ワーカー・ウェブサーバ・メタデータ DB）をマネージドな基盤上に構築します。利用者は DAG の Python ファイルを DAGs フォルダへ置くだけで、スケジューラがそれを解析し、依存関係とスケジュールに従ってタスクをワーカーへ割り当てます。

• スケジューラは DAG を解析して 実行すべきタスクを判定し、依存が満たされたものから順にワーカーへ渡す
• ワーカーがタスクを実行し、成否・ログ・状態をメタデータ DB に記録する。これにより再実行や進行状況の追跡が可能になる
• 失敗したタスクは リトライポリシーに従って再試行され、上限まで失敗すると DAG 全体やダウンストリームの扱いが設定に応じて決まる
• センサーは外部条件（ファイル到着・他ジョブ完了など）が満たされるまで待機し、整ったら後続を起動する
• オペレータや各種フック経由で BigQuery・Dataflow・Dataproc・Cloud Storage などを呼び出し、結果を次のタスクへつなぐ

設計パターン / ベストプラクティス

• データパイプラインのオーケストレーション: 抽出・変換・ロード（ETL/ELT）の各ステップを DAG のタスクとして並べ、BigQuery や Dataflow などの実処理を呼び出す中心役にする
• 冪等なタスク設計: リトライやバックフィルで同じタスクが再実行されうるため、各タスクは何度実行しても結果が壊れない冪等な処理にする
• 実行日（論理日付）を基準にする: 処理対象を実行時刻ではなく DAG の論理日付で切り出し、過去分のバックフィルや再実行を安全にする
• DAG は軽く保つ: DAG ファイルのトップレベルに重い処理や外部呼び出しを書くと解析のたびに走り負荷になる。定義は軽量にし、実処理はタスク内に置く
• 依存とトリガーの使い分け: イベント駆動の疎結合な配信は Pub/Sub や Eventarc、明確な依存グラフを持つバッチの統括は Composer、と役割を分ける
• 環境の分離: 本番と検証で環境を分け、DAG の変更はまず検証環境で確認してから本番へ反映する

運用・監視

• DAG ごとの 実行状況・依存グラフ・タスク状態・ログを Airflow の Web UI で確認でき、どのタスクで失敗したかを追跡できる
• Cloud Logging に Airflow とタスクのログが出力され、Cloud Monitoring でスケジューラの健全性・タスク失敗数・キュー滞留などのメトリクスを監視してアラートを設定する
• 失敗時は該当タスクのログと リトライ設定を確認し、必要に応じて UI や CLI からタスク単位・DAG 単位で再実行する
• DAG が反映されない → DAGs フォルダへの配置漏れや 解析エラー（import error）、解析間隔ぶんの遅延を確認する
• タスクが滞留する → ワーカー数や同時実行設定がボトルネックになっていないかを点検し、環境サイズやワーカーをスケールする
• バージョンアップやライブラリ追加は 互換性を検証環境で確認してから本番へ適用する

コスト

Cloud Composer は環境を 常時稼働させる基盤型のサービスで、環境を動かしている間はコンピュート・ストレージ・データベースなどの料金が継続的に発生します。単発トリガーのように「実行した分だけ」ではない点が、軽量なスケジューラ系サービスとの大きな違いです。

セキュリティ

• 環境やタスクの実行には サービスアカウントを割り当て、呼び出すサービス（BigQuery・Cloud Storage 等）に必要な 最小権限の IAM ロールだけを付与する
• 認証情報やトークンを DAG コードに直書きしない。Airflow の接続（Connection）や変数、Secret Manager 連携を使って機密値を外部管理する
• Airflow Web UI へのアクセスは IAM で制御し、必要な担当者だけに権限を絞る
• ネットワーク要件に応じて、限定公開（プライベート）構成で環境を作り、外部からの到達経路を絞る設計を検討する
• 保存データは暗号化され、要件に応じて CMEK（顧客管理鍵） による保護にも対応する

関連サービス・比較

同じく連携を司る Workflows と比べると、Composer は 常時稼働の Airflow 基盤で依存グラフ型のバッチ／データパイプラインを統括するのに向き、Workflows は サーバーレスで軽量な手順オーケストレーションに向きます。用途の重さと運用形態で選び分けます。

ハンズオン / CLI例

# 1) Composer 環境を作成（サービスアカウントを指定。作成には時間がかかる）
gcloud composer environments create my-env \
  --location=asia-northeast1 \
  --image-version=composer-2-airflow-2 \
  --service-account=composer-sa@PROJECT_ID.iam.gserviceaccount.com

# 2) DAG 定義ファイルを環境の DAGs フォルダ(Cloud Storage)へ配置
gcloud composer environments storage dags import \
  --environment=my-env \
  --location=asia-northeast1 \
  --source=./my_dag.py

# 3) 取り込まれた DAG の一覧を確認（Airflow CLI を環境内で実行）
gcloud composer environments run my-env \
  --location=asia-northeast1 \
  dags list

# 4) DAG を手動トリガーして動作を確認
gcloud composer environments run my-env \
  --location=asia-northeast1 \
  dags trigger -- my_dag

# 5) Airflow Web UI の URL を確認
gcloud composer environments describe my-env \
  --location=asia-northeast1 \
  --format="value(config.airflowUri)"