Exécuter un DAG Apache Airflow dans Cloud Composer 1 (CLI Google Cloud)

Exécuter un DAG Apache Airflow dans Cloud Composer 1 (CLI Google Cloud)

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Ce guide de démarrage rapide explique comment créer un environnement Cloud Composer et exécuter un DAG Apache Airflow dans Cloud Composer 1.

• Si vous êtes un nouvel utilisateur d'Airflow, consultez le tutoriel sur les concepts Airflow dans la documentation Apache Airflow pour en savoir plus sur les concepts, les objets et leur utilisation.

• Si vous préférez utiliser la console Google Cloud, consultez Exécuter un DAG Apache Airflow dans Cloud Composer.

• Si vous souhaitez créer un environnement à l'aide de Terraform, consultez la section Créer des environnements (Terraform).

Avant de commencer

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
2. Install the Google Cloud CLI.

3. To initialize the gcloud CLI, run the following command:

   gcloud init

4. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

5. Make sure that billing is enabled for your Google Cloud project.

6. Install the Google Cloud CLI.

7. To initialize the gcloud CLI, run the following command:

   gcloud init

8. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

9. Make sure that billing is enabled for your Google Cloud project.

10. Enable the Cloud Composer API:

    gcloud services enable composer.googleapis.com

11. Pour obtenir les autorisations nécessaires pour suivre ce guide de démarrage rapide, demandez à votre administrateur de vous accorder les rôles IAM suivants sur votre projet :

    • Pour afficher, créez et gérez l'environnement Cloud Composer:
      • Administrateur de l'environnement et des objets Storage (roles/composer.environmentAndStorageObjectAdmin)
      • Utilisateur du compte de service (roles/iam.serviceAccountUser)
    • Pour afficher les journaux: Lecteur de journaux (roles/logging.viewer)

    Pour en savoir plus sur l'attribution de rôles, consultez la page Gérer l'accès aux projets, aux dossiers et aux organisations.

    Vous pouvez également obtenir les autorisations requises via des rôles personnalisés ou d'autres rôles prédéfinis.

Créez un environnement

Créez un environnement nommé example-environment dans la région us-central1, avec la dernière version de Cloud Composer 1.

gcloud composer environments create example-environment \
    --location us-central1 \
    --image-version composer-1.20.12-airflow-1.10.15

Créer un fichier DAG

Un DAG Airflow est un ensemble de tâches organisées que vous voulez planifier et exécuter. Les DAG sont définis dans des fichiers Python standards.

Ce guide utilise un exemple de DAG Airflow défini dans le fichier quickstart.py. Le code Python de ce fichier effectue les opérations suivantes :

1. Il crée un DAG, composer_sample_dag, Ce DAG s'exécute tous les jours.
2. Il exécute une tâche, print_dag_run_conf, qui imprime la configuration de l'exécution du DAG à l'aide de l'opérateur bash.

Enregistrez une copie du fichier quickstart.py sur votre ordinateur local :

import datetime

from airflow import models
from airflow.operators import bash

# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = {
    "owner": "Composer Example",
    "depends_on_past": False,
    "email": [""],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "start_date": YESTERDAY,
}

with models.DAG(
    "composer_quickstart",
    catchup=False,
    default_args=default_args,
    schedule_interval=datetime.timedelta(days=1),
) as dag:
    # Print the dag_run id from the Airflow logs
    print_dag_run_conf = bash.BashOperator(
        task_id="print_dag_run_conf", bash_command="echo {{ dag_run.id }}"
    )

Importez le fichier DAG dans le bucket de votre environnement.

Chaque environnement Cloud Composer est associé à un bucket Cloud Storage. Planifications Airflow dans Cloud Composer uniquement DAG situés dans le dossier /dags de ce bucket.

Pour planifier votre DAG, importez quickstart.py depuis votre ordinateur local vers votre dossier /dags de l'environnement:

Pour importer quickstart.py avec Google Cloud CLI, exécutez la commande suivante dans Dossier où se trouve le fichier quickstart.py:

gcloud composer environments storage dags import \
--environment example-environment --location us-central1 \
--source quickstart.py

Afficher le DAG

Une fois le fichier DAG importé, Airflow effectue les opérations suivantes :

1. Analyse le fichier DAG que vous avez importé. L'affichage du DAG dans Airflow peut prendre quelques minutes.
2. Ajoute le DAG à la liste des DAG disponibles.
3. Exécute le DAG selon la planification que vous avez indiquée dans le fichier DAG.

Vérifiez que votre DAG est traité sans erreur et qu'il est disponible dans Airflow en l'affichant dans l'UI DAG. L'UI DAG est l'interface Cloud Composer permettant d'afficher les informations sur les DAG dans la console Google Cloud. Cloud Composer fournit également un accès à l'interface utilisateur d'Airflow, qui est une interface Web Airflow native.

1. Attendez environ cinq minutes pour laisser Airflow traiter le fichier DAG que vous avez importé précédemment et terminer la première exécution du DAG (expliqué plus loin).

2. Exécutez la commande suivante dans la Google Cloud CLI. Cette commande exécute dags list : commande CLI Airflow qui répertorie les DAG de votre environnement.

   gcloud composer environments run example-environment \
   --location us-central1 \
   dags list
   

3. Vérifiez que le DAG composer_quickstart est listé dans le résultat de la commande.

   Exemple de résultat :

   Executing the command: [ airflow dags list ]...
   Command has been started. execution_id=d49074c7-bbeb-4ee7-9b26-23124a5bafcb
   Use ctrl-c to interrupt the command
   dag_id              | filepath              | owner            | paused
   ====================+=======================+==================+=======
   airflow_monitoring  | airflow_monitoring.py | airflow          | False
   composer_quickstart | dag-quickstart-af2.py | Composer Example | False
   

   Remarque :Cloud Composer utilise le DAG airflow_monitoring pour surveiller l'état de votre environnement.

Afficher les détails de l'exécution du DAG

Une seule exécution d'un DAG est appelée exécution DAG. Airflow exécute immédiatement une exécution DAG pour l'exemple de DAG, car la date de début du fichier DAG est définie sur hier. De cette manière, Airflow rattrape la planification du DAG spécifié.

L'exemple de DAG contient une tâche, print_dag_run_conf, qui exécute le modèle de DAG echo dans la console. Cette commande affiche des métadonnées sur le DAG (identifiant numérique de l'exécution du DAG).

Exécutez la commande suivante dans la Google Cloud CLI. Cette commande liste les exécutions de DAG pour le DAG composer_quickstart :

gcloud composer environments run example-environment \
--location us-central1 \
dags list-runs -- --dag-id composer_quickstart

Exemple de résultat :

dag_id              | run_id                                      | state   | execution_date                   | start_date                       | end_date
====================+=============================================+=========+==================================+==================================+=================================
composer_quickstart | scheduled__2024-02-17T15:38:38.969307+00:00 | success | 2024-02-17T15:38:38.969307+00:00 | 2024-02-18T15:38:39.526707+00:00 | 2024-02-18T15:38:42.020661+00:00

Remarque : Vous pouvez utiliser la commande tasks states-for-dag-run pour obtenir l'état de chaque tâche.

La CLI Airflow ne fournit pas de commande permettant d'afficher les journaux des tâches. Vous pouvez utiliser d'autres méthodes pour afficher les journaux des tâches Airflow : l'interface utilisateur du DAG Cloud Composer, l'interface utilisateur d'Airflow ou Cloud Logging. Ce guide montre un moyen d'interroger Cloud Logging pour obtenir les journaux d'une exécution de DAG spécifique.

Exécutez la commande suivante dans la Google Cloud CLI. Cette commande lit les journaux Cloud Logging pour une exécution spécifique du DAG composer_quickstart. L'argument --format met en forme la sortie de sorte que seul le texte du message de journal s'affiche.

gcloud logging read \
--format="value(textPayload)" \
--order=asc \
"resource.type=cloud_composer_environment \
resource.labels.location=us-central1 \
resource.labels.environment_name=example-environment \
labels.workflow=composer_quickstart \
(labels.\"execution-date\"=\"RUN_ID\")"

Remplacez :

• RUN_ID par la valeur run_id issue de la sortie de la commande tasks states-for-dag-run que vous avez exécutée précédemment. Par exemple, 2024-02-17T15:38:38.969307+00:00.

Exemple de résultat :

...

Starting attempt 1 of 2
Executing <Task(BashOperator): print_dag_run_conf> on 2024-02-17
15:38:38.969307+00:00
Started process 22544 to run task

...

Running command: ['/usr/bin/bash', '-c', 'echo 115746']
Output:
115746

...

Command exited with return code 0
Marking task as SUCCESS. dag_id=composer_quickstart,
task_id=print_dag_run_conf, execution_date=20240217T153838,
start_date=20240218T153841, end_date=20240218T153841
Task exited with return code 0
0 downstream tasks scheduled from follow-on schedule check

Effectuer un nettoyage

Pour éviter que les ressources utilisées sur cette page ne soient facturées sur votre compte Google Cloud, supprimez le projet Google Cloud contenant les ressources.

Supprimez les ressources utilisées dans ce tutoriel :

1. Supprimez l'environnement Cloud Composer.

   1. Dans la console Google Cloud, accédez à la page Environnements.

      Accéder à la page Environnements

   2. Sélectionnez example-environment, puis cliquez sur Supprimer.

   3. Attendez que l'environnement soit supprimé.

2. Supprimez le bucket de votre environnement. La suppression de l'environnement Cloud Composer ne supprime pas son bucket.

   1. Dans la console Google Cloud, accédez à Stockage > Navigateur.

      Accéder à Cloud Storage > Navigateur

   2. Sélectionnez le bucket de l'environnement, puis cliquez sur Supprimer. Par exemple, ce bucket peut être nommé us-central1-example-environ-c1616fe8-bucket.

3. Supprimez le disque persistant de la file d'attente Redis de votre environnement. La suppression de l'environnement Cloud Composer ne supprime pas son disque persistant.

   1. Dans la console Google Cloud, accédez à Compute Engine > Disques :

      Accéder à la page Disques

   2. Sélectionnez le disque persistant de la file d'attente Redis de l'environnement, puis cliquez sur Supprimer.

      Par exemple, ce disque peut être nommé gke-us-central1-exampl-pvc-b12055b6-c92c-43ff-9de9-10f2cc6fc0ee. Les disques pour Cloud Composer 1 ont toujours le type Standard persistent disk et la taille de 2 Go.

Étape suivante

• Ajouter et mettre à jour des DAG
• Écrire des DAG
• Remplacer les configurations Airflow
• Accéder à l'interface utilisateur d'Airflow