他の環境とプロジェクトで DAG をトリガーする

Managed Airflow（Gen 3） | Managed Airflow（Gen 2） | Managed Airflow（レガシー Gen 1）

このページでは、Managed Airflow 用の Airflow オペレーターを使用して、他の Managed Airflow 環境とプロジェクトで DAG をトリガーする DAG を実装する方法について説明します。

代わりに環境で DAG をトリガーする場合は、 DAG のスケジュール設定とトリガーをご覧ください。

IAM 権限を構成する

ターゲット環境が別のプロジェクトにある場合、環境のサービス アカウントには、そのプロジェクトの環境とやり取りできるロールが必要です。

プロジェクト リソース プリンシパル ロール
ターゲット環境が配置されているプロジェクト プロジェクト ソース環境の環境サービス アカウント Composer ワークロード ロール（composer.worker）
ターゲット環境が配置されているプロジェクト プロジェクト ソース環境の環境サービス アカウント カスタムロールと composer.environments.executeAirflowCommand権限

別の環境で DAG をトリガーする

このセクションで説明する DAG の例では、次のことを行います。

  1. 別の Managed Airflow 環境で DAG をトリガーします。
  2. DAG 実行が完了したかどうかを確認します。

別の環境での DAG 実行が完了すると、サンプル DAG は成功としてマークされます。

CloudComposerRunAirflowCLICommandOperator を使用して Airflow CLI コマンドを実行する

CloudComposerRunAirflowCLICommandOperator オペレーターを使用すると、別の Managed Airflow 環境で Airflow CLI コマンドを実行できます。サンプル DAG は、DAG をトリガーする dags trigger コマンドを実行します。

このオペレーターは遅延可能モードで実行できます。有効にするには、 deferrable パラメータを True に設定します。

run_airflow_cli_cmd = CloudComposerRunAirflowCLICommandOperator(
    task_id="run_airflow_cli_cmd",
    project_id="target-project",
    environment_id="target-composer-environment",
    region="us-central1",
    command="dags trigger -- target_dag",
    # You can run this operator in the deferrable mode:
    # deferrable=True
)

DAG 実行が完了したかどうかを確認する

CloudComposerDAGRunSensor センサーを使用すると、別の Managed Airflow 環境で DAG 実行が完了したかどうかを確認できます。

このセンサーは遅延可能モードで実行できます。有効にするには、deferrable パラメータを True に設定します。

dag_run_sensor = CloudComposerDAGRunSensor(
    task_id="dag_run_sensor",
    project_id="target-project",
    environment_id="target-composer-environment",
    region="us-central1",
    composer_dag_id="target_dag",
    allowed_states=["success"],
    # You can run this sensor in the deferrable mode:
    # deferrable=True
)

完全なサンプルコード

以下は、前述の 2 つのタスクを組み合わせた DAG の完全なコード例です。

from datetime import datetime, timedelta
from airflow.models.dag import DAG
from airflow.providers.google.cloud.operators.cloud_composer import (
    CloudComposerRunAirflowCLICommandOperator,
)
from airflow.providers.google.cloud.sensors.cloud_composer import CloudComposerDAGRunSensor

DAG_ID = "trigger_dag_in_another_composer_environment"

TARGET_PROJECT_ID = "example-target-project"
TARGET_REGION = "example-target-region"
TARGET_ENV_ID = "example-target-composer-environment"

TARGET_DAG = "example_target_dag_id"
COMMAND = f"dags trigger -- {TARGET_DAG}"

with DAG(
    DAG_ID,
    schedule="@once",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["example", "composer"],
) as dag:

    run_airflow_cli_cmd = CloudComposerRunAirflowCLICommandOperator(
        task_id="run_airflow_cli_cmd",
        project_id=TARGET_PROJECT_ID,
        environment_id=TARGET_ENV_ID,
        region=TARGET_REGION,
        command=COMMAND,
        # You can run this operator in the deferrable mode:
        # deferrable=True
    )

    dag_run_sensor = CloudComposerDAGRunSensor(
        task_id="dag_run_sensor",
        project_id=TARGET_PROJECT_ID,
        environment_id=TARGET_ENV_ID,
        region=TARGET_REGION,
        composer_dag_id=TARGET_DAG,
        allowed_states=["success"],
        execution_range=timedelta(minutes=5),
        # You can run this sensor in the deferrable mode:
        # deferrable=True
    )

    run_airflow_cli_cmd >> dag_run_sensor

次のステップ