Planifier et déclencher des DAG Airflow

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

Cette page explique comment la planification et le déclenchement des DAG fonctionnent dans Airflow, comment définir une planification pour un DAG, et comment déclencher un DAG manuellement ou le suspendre.

À propos des DAG Airflow dans Airflow géré

Les DAG Airflow dans Airflow géré sont exécutés dans un ou plusieurs environnements Airflow gérés de votre projet. Vous importez les fichiers sources de vos DAG Airflow dans un bucket Cloud Storage associé à un environnement. L'instance Airflow de l'environnement analyse ensuite ces fichiers et planifie les exécutions de DAG, comme défini par la planification de chaque DAG. Lors d'une exécution de DAG, Airflow planifie et exécute les tâches individuelles qui composent un DAG dans une séquence définie par le DAG.

Pour en savoir plus sur les concepts de base d'Airflow, tels que les DAG Airflow, les exécutions de DAG, les tâches ou les opérateurs, consultez la page Concepts de base de la documentation Airflow.

À propos de la planification des DAG dans Airflow

Airflow propose les concepts suivants pour son mécanisme de planification :

Date logique

Représente une date pour laquelle une exécution de DAG particulière est exécutée.

Il ne s'agit pas de la date réelle à laquelle Airflow exécute un DAG, mais d'une période que l'exécution d'un DAG particulier doit traiter. Par exemple, pour un DAG qui est programmé pour s'exécuter tous les jours à 12h00, la date logique serait également 12h00 un jour spécifique. Comme il s'exécute deux fois par jour, la période qu'il doit traiter correspond aux 12 dernières heures. En même temps, la logique définie dans le DAG lui-même peut ne pas utiliser du tout la date logique ni l'intervalle de temps. Par exemple, un DAG peut exécuter le même script une fois par jour sans utiliser la valeur de la date logique.

Dans les versions d'Airflow antérieures à la version 2.2, cette date est appelée date d'exécution.

Date d'exécution

Représente une date à laquelle une exécution de DAG particulière est exécutée.

Par exemple, pour un DAG qui est programmé pour s'exécuter tous les jours à 12h00, l'exécution réelle du DAG peut avoir lieu à 12h05, quelque temps après la date logique.

Intervalle de planification

Représente le moment et la fréquence d'exécution d'un DAG, en termes de dates logiques.

Par exemple, une planification quotidienne signifie qu'un DAG est exécuté une fois par jour, et que les dates logiques de ses exécutions de DAG ont des intervalles de 24 heures.

Date de début

Spécifie à quel moment Airflow doit commencer à planifier votre DAG.

Les tâches de votre DAG peuvent avoir des dates de début individuelles, ou vous pouvez spécifier une date de début unique pour toutes les tâches. En fonction de la date de début minimale des tâches de votre DAG et de l'intervalle de planification, Airflow planifie l'exécution du DAG.

Rattrapage, remplissage et nouvelles tentatives

Mécanismes d'exécution des DAG pour des dates antérieures.

Le rattrapage exécute les DAG qui ne se sont pas encore exécutés, par exemple si le DAG a été suspendu pendant une longue période, puis réactivé. Vous pouvez utiliser le remplissage pour exécuter des DAG pour une certaine plage de dates. Les nouvelles tentatives spécifient le nombre de tentatives qu'Airflow doit effectuer lors de l'exécution des tâches d'un DAG.

La planification fonctionne de la manière suivante :

  1. Une fois la date de début dépassée, Airflow attend la prochaine occurrence de l'intervalle de planification.

  2. Airflow planifie la première exécution du DAG à la fin de cet intervalle de planification.

    Par exemple, si un DAG est exécuté toutes les heures et que la date de début est à 12h00, la première exécution du DAG a lieu aujourd'hui à 13h.

La section Planifier un DAG Airflow de ce document décrit comment configurer la planification de vos DAG à l'aide de ces concepts. Pour en savoir plus sur les exécutions de DAG et la planification, consultez la page Exécutions de DAG dans la documentation Airflow.

À propos des méthodes pour déclencher un DAG

Airflow propose les méthodes suivantes pour déclencher un DAG :

  • Déclenchement planifié. Airflow déclenche le DAG automatiquement en fonction de la planification spécifiée dans le fichier DAG.

  • Déclenchement manuel. Vous pouvez déclencher un DAG manuellement à partir de Google Cloud la console, de l'interface utilisateur d'Airflow ou en exécutant une commande de CLI Airflow à partir de Google Cloud CLI.

  • Déclenchement en réponse à des événements. Le moyen standard de déclencher un DAG en réponse à des événements consiste à utiliser un capteur.

Autres méthodes pour déclencher des DAG :

Avant de commencer

  • Assurez-vous que votre compte dispose d'un rôle qui peut gérer des objets dans les buckets d'environnement, et afficher et déclencher des DAG. Pour plus d'informations, consultez la section Contrôle des accès.

Planifier un DAG Airflow

Vous définissez une planification pour un DAG dans le fichier DAG. Modifiez la définition du DAG comme suit :

  1. Recherchez et modifiez le fichier DAG sur votre ordinateur. Si vous ne disposez pas du fichier DAG, vous pouvez télécharger sa copie à partir du bucket de l'environnement. Pour un nouveau DAG, vous pouvez définir tous les paramètres lorsque vous créez le fichier DAG.

  2. Dans le paramètre schedule, définissez la planification. Vous pouvez utiliser une expression Cron, telle que 0 0 * * *, ou un préréglage, tel que @daily. Pour en savoir plus, consultez la section Cron et intervalles de temps dans la documentation Airflow.

    Airflow détermine les dates logiques des exécutions de DAG en fonction de la planification que vous définissez.

  3. Dans le paramètre start_date, définissez la date de début.

    Airflow détermine la date logique de la première exécution du DAG à l'aide de ce paramètre.

  4. (Facultatif) Dans le paramètre catchup, définissez si Airflow doit exécuter toutes les exécutions précédentes de ce DAG depuis la date de début jusqu'à la date actuelle qui n'ont pas encore été exécutées.

    Les exécutions de DAG effectuées lors du rattrapage auront leur date logique dans le passé et leur date d'exécution reflétera l'heure à laquelle l'exécution du DAG a réellement eu lieu.

  5. (Facultatif) Dans le paramètre retries, définissez le nombre de fois qu'Airflow doit réessayer les tâches qui ont échoué (chaque DAG se compose d'une ou plusieurs tâches individuelles). Par défaut, les tâches dans Airflow géré sont retentées deux fois.

  6. Importez la nouvelle version du DAG dans le bucket de l'environnement.

  7. Attendez qu'Airflow analyse correctement le DAG. Par exemple, vous pouvez consulter la liste des DAG dans votre environnement dans la Google Cloud console ou dans l'interface utilisateur d'Airflow.

L'exemple de définition de DAG suivant s'exécute deux fois par jour à 00h00 et 12h00. Sa date de début est définie sur le 1er janvier 2024, mais Airflow ne l'exécute pas pour les dates antérieures après que vous l'avez importé ou suspendu, car le rattrapage est désactivé.

Le DAG contient une tâche nommée insert_query_job, qui insère une ligne dans une table avec l'opérateur BigQueryInsertJobOperator. Cet opérateur est l'un des Google Cloud opérateurs BigQuery, que vous pouvez utiliser pour gérer des ensembles de données et des tables, exécuter des requêtes et valider des données. Si une exécution particulière de cette tâche échoue, Airflow la retente quatre fois de plus avec l'intervalle de nouvelles tentatives par défaut. La date logique de ces nouvelles tentatives reste la même.

La requête SQL de cette ligne utilise des modèles Airflow pour écrire la date logique et le nom du DAG dans la ligne.

import datetime

from airflow.models.dag import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

with DAG(
  "bq_example_scheduling_dag",
  start_date=datetime.datetime(2024, 1, 1),
  schedule='0 */12 * * *',
  catchup=False
  ) as dag:

  insert_query_job = BigQueryInsertJobOperator(
    task_id="insert_query_job",
    retries=4,
    configuration={
        "query": {
            # schema: date (string), description (string)
            # example row: "20240101T120000", "DAG run: <DAG: bq_example_scheduling_dag>"
            "query": "INSERT example_dataset.example_table VALUES ('{{ ts_nodash }}', 'DAG run: {{ dag }}' )",
            "useLegacySql": False,
            "priority": "BATCH",
        }
    },
    location="us-central1"
  )

  insert_query_job

Pour tester ce DAG, vous pouvez le déclencher manuellement et ensuite afficher les journaux d'exécution des tâches.

Autres exemples de paramètres de planification

Les exemples suivants de paramètres de planification illustrent le fonctionnement de la planification avec différentes combinaisons de paramètres :

  • Si start_date est datetime(2024, 4, 4, 16, 25) et que schedule est 30 16 * * *, la première exécution du DAG a lieu à 16h30 le 5 avril 2024.

  • Si start_date est datetime(2024, 4, 4, 16, 35) et que schedule est 30 16 * * *, la première exécution du DAG a lieu à 16h30 le 6 avril 2024. Étant donné que la date de début est postérieure à l'intervalle de planification le 4 avril 2024, l'exécution du DAG n'a pas lieu le 5 avril 2024. Au lieu de cela, l'intervalle de planification se termine le 5 avril 2024 à 16h35. L'exécution du DAG suivante est donc programmée pour 16h30 le lendemain.

  • Si start_date est datetime(2024, 4, 4) et que schedule est @daily, la première exécution du DAG est planifiée pour le 5 avril 2024 à 00h00.

  • Si start_date est datetime(2024, 4, 4, 16, 30) et que schedule est 0 * * * *, la première exécution du DAG est planifiée pour 18h le 4 avril 2024. Une fois la date et l'heure spécifiées écoulées, Airflow planifie l'exécution d'un DAG à la minute zéro de chaque heure. Le moment le plus proche lorsque cela se produit est 17h00. Pour le moment, Airflow planifie l'exécution d'un DAG à la fin de l'intervalle de planification, c'est-à-dire à 18h00.

Déclencher un DAG manuellement

Lorsque vous déclenchez manuellement un DAG Airflow, Airflow exécute le DAG une seule fois, indépendamment de la planification spécifiée dans le fichier DAG.

Console

Pour déclencher un DAG à partir de la Google Cloud console :

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

    Accéder à la page Environnements

  2. Sélectionnez un environnement pour afficher ses détails.

  3. Sur la page Détails de l'environnement, accédez à l'onglet DAG.

  4. Cliquez sur le nom d'un DAG.

  5. Sur la page Détails du DAG, cliquez sur Déclencher le DAG. Une nouvelle exécution du DAG est créée.

Interface utilisateur d'Airflow

Pour déclencher un DAG à partir de l'interface utilisateur d'Airflow :

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

    Accéder à la page Environnements

  2. Dans la colonne Serveur Web Airflow, cliquez sur le lien Airflow correspondant à votre environnement.

  3. Connectez-vous avec le compte Google disposant des autorisations appropriées.

  4. Dans l'interface utilisateur d'Airflow, sur la page DAG, cliquez sur le bouton Déclencher le DAG pour votre DAG.

gcloud

Exécutez la commande de CLI Airflow dags trigger :

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags trigger -- DAG_ID

Remplacez les éléments suivants :

  • ENVIRONMENT_NAME : nom de votre environnement
  • LOCATION : région où se trouve l'environnement
  • DAG_ID : nom du DAG

Pour en savoir plus sur l'exécution de commandes de CLI Airflow dans les environnements Airflow gérés, consultez la page Exécuter des commandes de CLI Airflow.

Pour en savoir plus sur les commandes de CLI Airflow disponibles, consultez la documentation de référence sur la commande gcloud composer environments run.

Afficher les journaux et les détails d'exécution des DAG

Dans la Google Cloud console, vous pouvez effectuer les actions suivantes :

De plus, Airflow géré fournit un accès à l' interface utilisateur d'Airflow, qui est l'interface Web d'Airflow.

Suspendre un DAG

Console

Pour suspendre un DAG à partir de la Google Cloud console :

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

    Accéder à la page Environnements

  2. Sélectionnez un environnement pour afficher ses détails.

  3. Sur la page Détails de l'environnement, accédez à l'onglet DAG.

  4. Cliquez sur le nom d'un DAG.

  5. Sur la page Détails du DAG, cliquez sur Suspendre le DAG.

Interface utilisateur d'Airflow

Pour suspendre un DAG à partir de l'interface utilisateur d'Airflow :

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

Accéder à la page Environnements

  1. Dans la colonne Serveur Web Airflow, cliquez sur le lien Airflow correspondant à votre environnement.

  2. Connectez-vous avec le compte Google disposant des autorisations appropriées.

  3. Dans l'interface Web Airflow, sur la page DAG, cliquez sur le bouton à bascule à côté du nom du DAG.

gcloud

Exécutez la commande de CLI Airflow dags pause :

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags pause -- DAG_ID

Remplacez les éléments suivants :

  • ENVIRONMENT_NAME : nom de votre environnement
  • LOCATION : région où se trouve l'environnement
  • DAG_ID : nom du DAG

Pour en savoir plus sur l'exécution de commandes de CLI Airflow dans les environnements Airflow gérés, consultez la page Exécuter des commandes de CLI Airflow.

Pour en savoir plus sur les commandes de CLI Airflow disponibles, consultez la documentation de référence sur la commande gcloud composer environments run.

Étape suivante