Écrire des DAG (workflows)

Ce guide explique comment écrire un DAG (graphe orienté acyclique) Apache Airflow exécuté dans un environnement Cloud Composer.

Structurer un DAG

Un DAG Airflow est défini dans un fichier Python et comprend les composants suivants : une définition du DAG, des opérateurs et des relations entre opérateurs. Les extraits de code suivants illustrent des exemples de chaque composant hors contexte :

  1. La définition du DAG.

    Airflow 2

    import datetime
    
    from airflow import models
    default_dag_args = {
        # The start_date describes when a DAG is valid / can be run. Set this to a
        # fixed point in time rather than dynamically, since it is evaluated every
        # time a DAG is parsed. See:
        # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
        'start_date': datetime.datetime(2018, 1, 1),
    }
    
    # Define a DAG (directed acyclic graph) of tasks.
    # Any task you create within the context manager is automatically added to the
    # DAG object.
    with models.DAG(
            'composer_sample_simple_greeting',
            schedule_interval=datetime.timedelta(days=1),
            default_args=default_dag_args) as dag:

    Airflow 1

    import datetime
    
    from airflow import models
    default_dag_args = {
        # The start_date describes when a DAG is valid / can be run. Set this to a
        # fixed point in time rather than dynamically, since it is evaluated every
        # time a DAG is parsed. See:
        # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
        'start_date': datetime.datetime(2018, 1, 1),
    }
    
    # Define a DAG (directed acyclic graph) of tasks.
    # Any task you create within the context manager is automatically added to the
    # DAG object.
    with models.DAG(
            'composer_sample_simple_greeting',
            schedule_interval=datetime.timedelta(days=1),
            default_args=default_dag_args) as dag:

  2. Des opérateurs pour décrire la tâche à effectuer. L'instanciation d'un opérateur s'appelle une tâche.

    Airflow 2

    from airflow.operators import bash_operator
    from airflow.operators import python_operator
        def greeting():
            import logging
            logging.info('Hello World!')
    
        # An instance of an operator is called a task. In this case, the
        # hello_python task calls the "greeting" Python function.
        hello_python = python_operator.PythonOperator(
            task_id='hello',
            python_callable=greeting)
    
        # Likewise, the goodbye_bash task calls a Bash script.
        goodbye_bash = bash_operator.BashOperator(
            task_id='bye',
            bash_command='echo Goodbye.')

    Airflow 1

    from airflow.operators import bash_operator
    from airflow.operators import python_operator
        def greeting():
            import logging
            logging.info('Hello World!')
    
        # An instance of an operator is called a task. In this case, the
        # hello_python task calls the "greeting" Python function.
        hello_python = python_operator.PythonOperator(
            task_id='hello',
            python_callable=greeting)
    
        # Likewise, the goodbye_bash task calls a Bash script.
        goodbye_bash = bash_operator.BashOperator(
            task_id='bye',
            bash_command='echo Goodbye.')

  3. Les relations entre les tâches pour décrire l'ordre dans lequel le travail doit être effectué.

    Airflow 2

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

    Airflow 1

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

Le workflow suivant est un exemple fonctionnel complet et se compose de deux tâches, hello_python et goodbye_bash :

Airflow 2

from __future__ import print_function

import datetime

from airflow import models
from airflow.operators import bash_operator
from airflow.operators import python_operator

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    'start_date': datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
        'composer_sample_simple_greeting',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:
    def greeting():
        import logging
        logging.info('Hello World!')

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id='hello',
        python_callable=greeting)

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id='bye',
        bash_command='echo Goodbye.')

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

Airflow 1

from __future__ import print_function

import datetime

from airflow import models
from airflow.operators import bash_operator
from airflow.operators import python_operator

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    'start_date': datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
        'composer_sample_simple_greeting',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:
    def greeting():
        import logging
        logging.info('Hello World!')

    # An instance of an operator is called a task. In this case, the
    # hello_python task calls the "greeting" Python function.
    hello_python = python_operator.PythonOperator(
        task_id='hello',
        python_callable=greeting)

    # Likewise, the goodbye_bash task calls a Bash script.
    goodbye_bash = bash_operator.BashOperator(
        task_id='bye',
        bash_command='echo Goodbye.')

    # Define the order in which the tasks complete by using the >> and <<
    # operators. In this example, hello_python executes before goodbye_bash.
    hello_python >> goodbye_bash

Consultez le tutoriel Airflow et les concepts Airflow pour en savoir plus sur la définition des DAG Airflow.

Opérateurs

Les exemples suivants présentent quelques opérateurs Airflow connus. Pour obtenir une documentation de référence faisant autorité sur les opérateurs Airflow, consultez la documentation de référence de l'API Apache Airflow ou parcourez le code source des opérateurs principaux, de contribution et fournisseurs.

BashOperator

Utilisez BashOperator pour exécuter des programmes de ligne de commande.

Airflow 2

from airflow.operators import bash
    # Create BigQuery output dataset.
    make_bq_dataset = bash.BashOperator(
        task_id='make_bq_dataset',
        # Executing 'bq' command requires Google Cloud SDK which comes
        # preinstalled in Cloud Composer.
        bash_command=f'bq ls {bq_dataset_name} || bq mk {bq_dataset_name}')

Airflow 1

from airflow.operators import bash_operator
    # Create BigQuery output dataset.
    make_bq_dataset = bash_operator.BashOperator(
        task_id='make_bq_dataset',
        # Executing 'bq' command requires Google Cloud SDK which comes
        # preinstalled in Cloud Composer.
        bash_command='bq ls {} || bq mk {}'.format(
            bq_dataset_name, bq_dataset_name))

Cloud Composer exécute les commandes fournies dans un script bash sur un nœud de calcul. Le nœud de calcul est un conteneur Docker basé sur Debian et comprend plusieurs packages.

PythonOperator

Utilisez PythonOperator pour exécuter le code Python de votre choix.

Cloud Composer exécute le code Python dans un conteneur qui inclut des packages pour la version d'image Cloud Composer utilisée dans votre environnement.

Pour installer des packages Python supplémentaires, reportez-vous à la section Installer des dépendances Python.

Opérateurs Google Cloud

Utilisez les opérateurs Airflow pour Google Cloud afin d'exécuter des tâches qui utilisent des produits Google Cloud. Cloud Composer configure automatiquement une connexion Airflow sur le projet de l'environnement.

EmailOperator

Utilisez EmailOperator pour envoyer un e-mail à partir d'un DAG. Pour envoyer un e-mail à partir d'un environnement Cloud Composer, vous devez configurer votre environnement pour qu'il utilise SendGrid.

Airflow 2

from airflow.operators import email
    # Send email confirmation (you will need to set up the email operator
    # See https://cloud.google.com/composer/docs/how-to/managing/creating#notification
    # for more info on configuring the email operator in Cloud Composer)
    email_summary = email.EmailOperator(
        task_id='email_summary',
        to=models.Variable.get('email'),
        subject='Sample BigQuery notify data ready',
        html_content="""
        Analyzed Stack Overflow posts data from {min_date} 12AM to {max_date}
        12AM. The most popular question was '{question_title}' with
        {view_count} views. Top 100 questions asked are now available at:
        {export_location}.
        """.format(
            min_date=min_query_date,
            max_date=max_query_date,
            question_title=(
                '{{ ti.xcom_pull(task_ids=\'bq_read_most_popular\', '
                'key=\'return_value\')[0][0] }}'
            ),
            view_count=(
                '{{ ti.xcom_pull(task_ids=\'bq_read_most_popular\', '
                'key=\'return_value\')[0][1] }}'
            ),
            export_location=output_file))

Airflow 1

from airflow.operators import email_operator
    # Send email confirmation
    email_summary = email_operator.EmailOperator(
        task_id='email_summary',
        to=models.Variable.get('email'),
        subject='Sample BigQuery notify data ready',
        html_content="""
        Analyzed Stack Overflow posts data from {min_date} 12AM to {max_date}
        12AM. The most popular question was '{question_title}' with
        {view_count} views. Top 100 questions asked are now available at:
        {export_location}.
        """.format(
            min_date=min_query_date,
            max_date=max_query_date,
            question_title=(
                '{{ ti.xcom_pull(task_ids=\'bq_read_most_popular\', '
                'key=\'return_value\')[0][0] }}'
            ),
            view_count=(
                '{{ ti.xcom_pull(task_ids=\'bq_read_most_popular\', '
                'key=\'return_value\')[0][1] }}'
            ),
            export_location=output_file))

Notifications

Définissez email_on_failure sur True pour envoyer une notification par e-mail lorsqu'un opérateur du DAG échoue. Pour envoyer des notifications par e-mail à partir d'un environnement Cloud Composer, vous devez configurer votre environnement pour qu'il utilise SendGrid.

Airflow 2

from airflow import models
default_dag_args = {
    'start_date': yesterday,
    # Email whenever an Operator in the DAG fails.
    'email': models.Variable.get('email'),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': project_id
}

with models.DAG(
        'composer_sample_bq_notify',
        schedule_interval=datetime.timedelta(weeks=4),
        default_args=default_dag_args) as dag:

Airflow 1

from airflow import models
default_dag_args = {
    'start_date': yesterday,
    # Email whenever an Operator in the DAG fails.
    'email': models.Variable.get('email'),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

with models.DAG(
        'composer_sample_bq_notify',
        schedule_interval=datetime.timedelta(weeks=4),
        default_args=default_dag_args) as dag:

Consignes

  1. Placez toutes les bibliothèques Python personnalisées dans l'archive ZIP des DAG, dans un répertoire imbriqué. Ne placez pas de bibliothèques au niveau supérieur du répertoire des DAG.

    Lorsque Airflow analyse le dossier dags/, il ne recherche que les DAG des modules Python situés au niveau supérieur du dossier des DAG et au niveau supérieur d'une archive ZIP également située au niveau supérieur du dossier dags/. Si Airflow rencontre un module Python dans une archive ZIP ne contenant pas les deux sous-chaînes airflow et DAG, il cesse de traiter l'archive ZIP. Airflow ne renvoie que les DAG trouvés jusqu'à ce point.

  2. Pour la tolérance aux pannes, ne définissez pas plusieurs objets DAG dans le même module Python.

  3. Ne définissez pas les subDAG en tant qu'objets de niveau supérieur.

    En règle générale, Airflow sélectionne les objets DAG dans l'espace de noms global d'un module situé dans le répertoire dags/ en tant que DAG de niveau supérieur. Tous les subDAG définis en tant qu'objets de niveau supérieur s'exécutent selon leur propre planification, en plus des planifications d'autres DAG intégrant les subDAG.

  4. Placez les fichiers requis au moment de l'analyse du DAG dans le répertoire dags/, et non dans le répertoire data/. Le répertoire data/ n'est pas installé sur le serveur Web.

Questions fréquentes sur l'écriture des DAG

Comment réduire la répétition du code lorsque j'exécute des tâches identiques ou semblables dans plusieurs DAG ?

Nous vous recommandons de définir des bibliothèques et des wrappers afin de réduire la répétition du code.

Comment réutiliser le code à travers plusieurs fichiers DAG ?

Stockez les fonctions utilitaires dans une bibliothèque Python locale, puis importez-les. Vous pouvez référencer les fonctions de n'importe quel DAG situé dans le dossier dags/ du bucket de votre environnement.

Comment minimiser le risque d'apparition de définitions différentes ?

Imaginons que deux équipes souhaitent agréger des données brutes en métriques de revenus. Les équipes rédigent deux tâches légèrement différentes pour effectuer la même opération. Définissez les bibliothèques pour qu'elles fonctionnent avec les données sur les revenus. Ainsi, les développeurs des DAG doivent clarifier la définition des revenus en cours d'agrégation.

Comment définir des dépendances entre les DAG ?

Tout dépend de la manière dont vous souhaitez définir la dépendance.

Si vous avez deux DAG (DAG A et DAG B) et que vous souhaitez que le DAG B se déclenche après le DAG A, vous pouvez placer un opérateur TriggerDagRunOperator à la fin du DAG A.

Si le DAG B ne dépend que d'un artefact généré par le DAG A, tel qu'un message Pub/Sub, il est possible qu'un capteur fonctionne mieux.

Si le DAG B est étroitement intégré au DAG A, vous pourrez peut-être fusionner les deux DAG en un seul.

Comment transférer des ID d'exécution uniques à un DAG et à ses tâches ?

Imaginons que vous souhaitiez transférer les noms de cluster Dataproc et les chemins des fichiers.

Vous pouvez générer un ID aléatoire unique en renvoyant str(uuid.uuid4()) dans un opérateur PythonOperator. L'ID est alors placé dans XComs afin que vous puissiez y faire référence dans d'autres opérateurs via des champs modélisés.

Avant de générer un uuid, essayez de déterminer si un ID propre à DagRun serait plus utile. Vous pouvez également référencer ces ID dans les substitutions Jinja à l'aide de macros.

Comment séparer les tâches dans un DAG ?

Chaque tâche doit constituer une unité de travail idempotente. Par conséquent, évitez d'encapsuler un workflow en plusieurs étapes au sein d'une tâche unique, comme un programme complexe exécuté dans un opérateur PythonOperator.

Le mécanisme Airflow natif permettant de réutiliser le code de définition du workflow est l'opérateur SubDagOperator. Toutefois, il existe des mises en garde lorsque vous utilisez cet opérateur dans Cloud Composer.

Faut-il définir plusieurs tâches dans un même DAG pour agréger des données provenant de plusieurs sources ?

Par exemple, vous avez plusieurs tables avec des données brutes et vous souhaitez créer des agrégats quotidiens pour chaque table. Les tâches ne dépendent pas les unes des autres. Faut-il créer une tâche et un DAG pour chaque table ou créer un DAG général ?

Si vous acceptez que chaque tâche partage les mêmes propriétés au niveau du DAG, telles que schedule_interval, il est judicieux de définir plusieurs tâches dans un même DAG. Sinon, pour réduire la répétition du code, plusieurs DAG peuvent être générés à partir d'un seul module Python en les plaçant dans les globals() du module.

Comment limiter le nombre de tâches simultanées en cours d'exécution dans un DAG ?

Par exemple, vous souhaitez éviter de dépasser les limites/quotas d'utilisation de l'API ou éviter d'exécuter trop de processus simultanés.

Vous pouvez définir des pools Airflow dans l'interface utilisateur Web d'Airflow et associer des tâches à des pools existants dans vos DAG.

Questions fréquentes sur l'utilisation des opérateurs

Faut-il utiliser l'opérateur DockerOperator ?

Nous vous déconseillons d'utiliser l'opérateur DockerOperator, sauf s'il est utilisé pour lancer des conteneurs sur une installation Docker distante (et non dans un cluster d'environnement). Dans un environnement Cloud Composer, l'opérateur n'a pas accès aux daemons Docker.

Utilisez plutôt KubernetesPodOperator ou GKEPodOperator. Ces opérateurs peuvent lancer des pods Kubernetes dans des clusters Kubernetes ou GKE, respectivement. Notez qu'il est déconseillé de lancer des pods dans le cluster d'un environnement, car cela peut entraîner une concurrence des ressources.

Faut-il utiliser l'opérateur SubDagOperator ?

Nous vous déconseillons d'utiliser l'opérateur SubDagOperator.

Même si l'opérateur SubDagOperator peut fournir une encapsulation, les tâches SubDag nécessitent un emplacement de tâche. Si un nœud de calcul Airflow exécutant la tâche SubDag plante, toutes les tâches au sein du SubDag échouent, ce qui entraîne des workflows non fiables.

Faut-il exécuter du code Python uniquement dans des opérateurs PythonOperators pour séparer complètement les opérateurs Python ?

En fonction de votre objectif, plusieurs options s'offrent à vous.

Si votre seule préoccupation est de conserver des dépendances Python distinctes, vous pouvez utiliser l'opérateur PythonVirtualenvOperator.

Envisagez d'utiliser le KubernetesPodOperator. Cet opérateur vous permet de définir des pods Kubernetes et d'exécuter les pods dans d'autres clusters.

Comment utiliser l'opérateur KubernetesPodOperator en dehors de Google Cloud ?

Vous pouvez installer un fichier de configuration qui spécifie la méthode d'authentification avec le cluster GKE et le placer dans le dossier /data dans le bucket de votre environnement.

Ce dossier est installé dans l'environnement Cloud Composer.

Comment ajouter des packages binaires personnalisés ou non PyPI ?

Vous pouvez installer des packages hébergés dans des dépôts de packages privés.

Vous pouvez également utiliser KubernetesPodOperator pour exécuter un pod Kubernetes avec votre propre image créée à partir de packages personnalisés.

Comment transmettre uniformément des arguments à un DAG et à ses tâches ?

La compatibilité avec Airflow intégrée à la modélisation Jinja permet de transmettre des arguments pouvant être utilisés dans des champs modélisés.

Quand se produit la substitution de modèle ?

La substitution de modèle se produit sur les nœuds de calcul Airflow juste avant l'appel de la fonction pre_execute d'un opérateur. En pratique, cela signifie que les modèles ne sont remplacés que juste avant l'exécution d'une tâche.

Comment connaître les arguments d'opérateur compatibles avec la substitution de modèle ?

Les arguments d'opérateur compatibles avec la substitution de modèle Jinja2 sont explicitement marqués comme tels.

Recherchez le champ template_fields dans la définition de l'opérateur, qui contient la liste des noms d'arguments qui subiront une substitution de modèle.

Par exemple, consultez la page sur l'opérateur BashOperator, qui est compatible avec la modélisation pour les arguments bash_command et env.

Étape suivante