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

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 souhaitez 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 }}"
    )

Importer le fichier DAG dans le bucket de votre environnement

Chaque environnement Cloud Composer est associé à un bucket Cloud Storage. Airflow ne programme que les DAG situés dans le dossier /dags de ce bucket.

Pour planifier votre DAG, importez quickstart.py depuis votre ordinateur local dans le dossier /dags de votre environnement:

Pour importer quickstart.py avec la Google Cloud CLI, exécutez la commande suivante dans le 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é. La mise à disposition du DAG pour 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'interface utilisateur du DAG. L'UI du DAG est l'interface Cloud Composer permettant d'afficher les informations du DAG dans la console Google Cloud. Cloud Composer donne également accès à l'interface utilisateur Airflow, qui est une interface Web Airflow native.

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

2. Exécutez la commande suivante dans la Google Cloud CLI. Cette commande exécute la commande CLI Airflow dags list 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 répertorié 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
   

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 du 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 programmation du DAG spécifiée.

L'exemple de DAG contient une tâche, print_dag_run_conf, qui exécute la commande echo dans la console. Cette commande génère des méta-informations sur le DAG (identifiant numérique de l'exécution du DAG).

Exécutez la commande suivante dans la Google Cloud CLI. Cette commande répertorie les exécutions du 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

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 de tâches Airflow : l'UI du DAG Cloud Composer, l'UI Airflow ou Cloud Logging. Ce guide montre comment 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 à partir de Cloud Logging pour une exécution spécifique du DAG composer_quickstart. L'argument --format met en forme la sortie pour que seul le texte du message de journal soit affiché.

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 du résultat 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