在其他環境和專案中觸發 DAG

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer„�

本頁面說明如何實作 DAG，透過 Cloud Composer 適用的 Airflow 運算子，在其他 Cloud Composer 環境和專案中觸發 DAG。

如要在環境中觸發 DAG，請參閱「排定及觸發 DAG」。

設定 IAM 權限

如果目標環境位於其他專案，環境的服務帳戶必須具備可與該專案環境互動的角色。

注意： 如要授予單一權限，請在目標專案中建立自訂角色，將權限新增至該角色，然後將這個角色授予主體。

專案	資源	主體	角色
目標環境所在的專案	專案	來源環境的環境服務帳戶	Composer Worker 角色 (composer.worker)
目標環境所在的專案	專案	來源環境的環境服務帳戶	具有 composer.environments.executeAirflowCommand 權限的自訂角色

在其他環境中觸發 DAG

本節所述的 DAG 範例會執行下列操作：

1. 在其他 Cloud Composer 環境中觸發 DAG。
2. 檢查 DAG 執行作業是否完成。

另一個環境中的 DAG 執行作業完成後，範例 DAG 會標示為成功。

注意： Google 提供更多可與 Cloud Composer 環境搭配使用的 Airflow 運算子、感應器、觸發條件和掛鉤。舉例來說，您可以使用這些運算子建立、更新及刪除環境。

使用 CloudComposerRunAirflowCLICommandOperator 執行 Airflow CLI 指令

您可以使用 CloudComposerRunAirflowCLICommandOperator 運算子，在另一個 Cloud Composer 環境中執行 Airflow CLI 指令。範例 DAG 會執行 dags trigger 指令，觸發 DAG。

這個運算子可以在可延遲模式中執行，您可以將 deferrable 參數設為 True 來啟用。

重要事項： 這項運算子會耗用 environments.executeAirflowCommand 配額。如要觸發大量 DAG，建議使用 Airflow REST API。透過 CloudComposerRunAirflowCLICommandOperator 觸發 DAG 可能會導致環境達到配額上限，因此 Airflow CLI 指令將無法再執行。

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 感應器，檢查另一個 Cloud Composer 環境中的 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
)

完整程式碼範例

以下是 DAG 的完整程式碼範例，其中結合了先前說明的兩項工作。

注意： 如果目標環境位於其他專案中，請務必設定 IAM 權限，再執行這個 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

後續步驟

• 排定及觸發 DAG
• 使用 IAM 控管存取權