Cloud Composer

解決する課題

• 複数ステップから成るデータパイプラインを、依存関係・順序・再試行つきで確実に実行したい
• BigQuery、Dataflow、Dataproc、Cloud Storage など複数サービスをまたぐ処理を一元的にオーケストレーションしたい
• cron や自前スクリプトで散在したバッチ処理の運用を、可視化・履歴・アラートとともに集中管理したい
• Airflow を使いたいが、スケジューラやワーカーのインフラ運用・パッチ・スケールから解放されたい

主要概念と用語

• DAG（有向非巡回グラフ）: ワークフローの定義単位。複数のタスクとその依存関係を Python コードで表現する。循環しない（後戻りしない）順序を持つ
• タスク / オペレータ: DAG を構成する個々の処理がタスクで、その実体がオペレータ。BigQueryInsertJobOperator や DataflowTemplatedJobStartOperator など、GCP 連携用が多数用意されている
• スケジューラ / ワーカー / Web サーバー: DAG を解析して実行時刻を判定するスケジューラ、タスクを実際に動かすワーカー、UI を提供する Web サーバー。Composer はこれらをマネージドで構成する
• 環境（Environment）: Composer がプロビジョニングする Airflow 一式。GKE クラスタや Cloud Storage バケットなどの裏側リソースを含む管理単位
• DAG バケット: DAG ファイルを配置する Cloud Storage バケット。ここに Python ファイルを置くと環境へ同期される
• 接続（Connection）/ 変数（Variable）: 外部システムの認証情報やエンドポイントをまとめた接続、DAG 間で共有する設定値の変数。コードに直書きせず管理する
• XCom: タスク間で小さな値を受け渡す仕組み。大きなデータの受け渡しには使わず、参照（パスやジョブID）を渡すのが定石

仕様・制限・クォータ

• Composer にはメジャーバージョンの世代があり、新しい世代ではスケジューラやワーカーのスケーリングがより自律的になる。新規構築では原則として最新世代を選ぶ
• 環境は内部で GKE とマネージドな Airflow コンポーネントから構成され、ワーカー数や CPU・メモリを設定で調整できる。世代によっては負荷に応じた自動スケールに対応する
• DAG の解析間隔やタスクの最大同時実行数などは Airflow 構成（airflow.cfg 相当の上書き）で調整する。DAG 解析が重いと反映が遅延するため、ファイル先頭での重い処理は避ける
• Airflow のバージョンや Python パッケージは環境ごとに固定され、依存ライブラリは PyPI からの追加インストールで管理する。値が変動するクォータ・上限は公式ドキュメントで最新を確認する

内部の仕組み

Composer 環境を作成すると、裏側で GKE クラスタ上に Airflow のスケジューラ・ワーカー・Web サーバーなどがデプロイされ、メタデータを保持するデータベースと、DAG・ログ・プラグインを置く Cloud Storage バケットがセットで構成されます。利用者はこのバケットへ DAG の Python ファイルを置くだけでよく、クラスタの管理は Composer が担います。

スケジューラはバケット内の DAG を定期的に解析し、各 DAG のスケジュールに基づいて実行（DAG Run）を生成します。生成されたタスクはキューを介してワーカーへ割り当てられ、オペレータが実際の処理を実行します。GCP 連携オペレータは内部で各サービスの API を呼び出し、ジョブの完了をポーリングしながら成否を判定します。失敗時はタスク単位で再試行され、依存タスクは前段の成功を待ってから走ります。

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

• DAG は冪等に設計する。同じ実行日（論理日付）で再実行しても結果が二重にならないよう、出力先をパーティションや実行日で分離する
• 重い処理は Composer のワーカーで行わず、BigQuery / Dataflow / Dataproc へ委譲し、Composer は起動と完了監視に徹する
• タスク間の値の受け渡しは XCom で参照のみにとどめ、実データはオブジェクトストレージや BigQuery を経由させる
• 認証情報は DAG にハードコードせず、接続 / 変数または Secret Manager で管理する
• DAG ファイル冒頭での重い import・API 呼び出しを避け、解析を軽量に保つ。タスク本体の処理は実行時に行う
• 再試行回数・再試行間隔・タイムアウトをタスクに明示し、失敗時の挙動を制御する

運用・監視

• Airflow UI で DAG の実行状況、タスクのリトライ、失敗箇所、所要時間を可視化する。ツリービューやグラフビューで依存と進捗を確認する
• メトリクスとログは Cloud Monitoring / Cloud Logging に統合され、スケジューラの健全性、ワーカーのリソース使用率、DAG の実行失敗をアラート化できる
• スケジューラのハートビートが滞る、DAG 解析時間が長い、タスクがキューで滞留するといった兆候はワーカー不足や解析過負荷のサイン。ワーカー数・並列度・解析間隔を見直す
• DAG・ログは Cloud Storage バケットに蓄積されるため、古いログの保持とコストのバランスを管理する

コスト

セキュリティ

• 環境とワーカーにはサービスアカウントを割り当て、最小権限の IAM ロールで連携先サービスへのアクセスを制御する（AWS の IAM ロール相当）。鍵の埋め込みは避ける
• 機密情報は DAG に直書きせず、Airflow の接続 / 変数または Secret Manager で管理する
• ネットワークはプライベート IP 環境を選び、必要に応じて承認済みネットワークや VPC Service Controls で外部到達・データ流出を制限する
• Airflow UI へのアクセスは IAM と Identity-Aware Proxy で認可し、誰がどの環境を操作できるかを制御する

Well-Architected の観点

• 運用上の優秀性（operational）: ワークフローをコード（DAG）として管理することで、レビュー・バージョン管理・再現性が得られる。実行履歴・ログ・アラートが標準で揃い、障害時の原因追跡と再試行が容易になる。スケジューラやインフラの運用を Composer が肩代わりするため、チームはパイプラインのロジックに集中できる

試験で問われるポイント

関連サービス・比較

ハンズオン / CLI例

# 1) Composer 環境を作成（世代やリージョンは要件に合わせる）
gcloud composer environments create my-env \
  --location=asia-northeast1 \
  --image-version=composer-2-airflow-2

# 2) DAG の同期先バケットを確認
gcloud composer environments describe my-env \
  --location=asia-northeast1 \
  --format="value(config.dagGcsPrefix)"

# 3) ローカルの DAG ファイルを環境へアップロード（バケットに同期される）
gcloud composer environments storage dags import \
  --environment=my-env \
  --location=asia-northeast1 \
  --source=./my_pipeline.py

# 4) 環境内で Airflow CLI を実行して DAG 一覧を確認
gcloud composer environments run my-env \
  --location=asia-northeast1 \
  dags list