Déclencher des DAG dans d'autres environnements et projets

Managed Airflow (3e génération) | Managed Airflow (2e génération) | Managed Airflow (1re génération héritée)

Cette page explique comment implémenter un DAG qui déclenche des DAG dans d'autres environnements et projets Managed Airflow à l'aide d'opérateurs Airflow pour Managed Airflow.

Si vous souhaitez déclencher des DAG dans votre environnement, consultez Programmer et déclencher des DAG.

Configurer les autorisations IAM

Si l'environnement cible se trouve dans un autre projet, le compte de service de votre environnement doit disposer de rôles qui lui permettent d'interagir avec les environnements de ce projet.

**Remarque** : Pour accorder une seule autorisation, créez un rôle personnalisé dans le projet cible, ajoutez-y l'autorisation, puis accordez ce rôle à un compte principal.

Projet	Ressource	Principal	Rôle
Projet où se trouve l'environnement cible	Projet	Compte de service de l'environnement source	Rôle Nœud de calcul Composer (composer.worker)
Projet où se trouve l'environnement cible	Projet	Compte de service de l'environnement source	Rôle personnalisé doté de l' composer.environments.executeAirflowCommand autorisation

Déclencher un DAG dans un autre environnement

L'exemple de DAG décrit dans cette section effectue les opérations suivantes :

1. Déclencher un DAG dans un autre environnement Managed Airflow.
2. Vérifier si une exécution de DAG est terminée.

Une fois l'exécution du DAG terminée dans un autre environnement, l'exemple de DAG est marqué comme réussi.

Remarque : Google fournit d'autres opérateurs, capteurs, déclencheurs, et hooks Airflow qui fonctionnent avec les environnements Managed Airflow. Par exemple, vous pouvez créer, mettre à jour et supprimer des environnements à l'aide de ces opérateurs.

Exécuter des commandes de CLI Airflow avec CloudComposerRunAirflowCLICommandOperator

Vous pouvez utiliser l' CloudComposerRunAirflowCLICommandOperator pour exécuter des commandes de CLI Airflow dans un autre environnement Managed Airflow. L'exemple de DAG exécute la commande dags trigger, qui déclenche un DAG.

Cet opérateur peut s'exécuter en mode différé. Vous pouvez l'activer en définissant le paramètre deferrable sur True.

Important : Cet opérateur consomme le environments.executeAirflowCommand quota. Si vous souhaitez déclencher un grand nombre de DAG, nous vous recommandons de le faire avec l'API REST Airflow. Le déclenchement de DAG via CloudComposerRunAirflowCLICommandOperator peut entraîner le dépassement de la limite de quota de votre environnement. Par conséquent, les commandes de CLI Airflow ne seront plus exécutées.

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
)

Vérifier si une exécution de DAG est terminée

Vous pouvez utiliser le CloudComposerDAGRunSensor capteur pour vérifier si une exécution de DAG est terminée dans un autre environnement Managed Airflow.

Ce capteur peut s'exécuter en mode différé. Vous pouvez l'activer en définissant le paramètre deferrable sur 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
)

Exemple de code complet

Vous trouverez ci-dessous l'exemple de code complet d'un DAG qui combine les deux tâches décrites précédemment.

Remarque : Si l'environnement cible se trouve dans un autre projet, veillez à configurer les autorisations IAM avant d'exécuter ce 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

Étape suivante

• Programmer et déclencher des DAG
• Contrôle des accès avec IAM