Tester, synchroniser et déployer vos DAG depuis GitHub

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Remarque :Cette page n'a pas encore été révisée pour Cloud Composer 3 et affiche pour Cloud Composer 2.

Ce guide explique comment créer un pipeline CI/CD pour tester, synchroniser et déployer des DAG dans votre environnement Cloud Composer à partir de votre un dépôt de clés.

Si vous souhaitez uniquement synchroniser les données d'autres services, consultez Transférer des données depuis d'autres services.

Présentation du pipeline CI/CD

Figure 1. Schéma de l'architecture illustrant les étapes du flux (cliquez pour agrandir)

Le pipeline CI/CD qui permet de tester, de synchroniser et de déployer des DAG comprend les étapes suivantes :

1. Vous modifiez un DAG et transférez cette modification vers une branche de développement de votre dépôt.

2. Vous ouvrez une demande d'extraction sur la branche principale de votre dépôt.

3. Cloud Build exécute des tests unitaires pour vérifier que votre DAG est valide.

4. Votre requête d'extraction est approuvée et fusionnée dans la branche principale de votre dépôt.

5. Cloud Build synchronise votre environnement Cloud Composer de développement avec ces nouvelles modifications.

6. Vous vérifierez que le DAG se comporte comme prévu dans votre environnement de développement environnement.

7. Si votre DAG fonctionne comme prévu, vous devez l'importer dans votre environnement de production Cloud Composer.

Objectifs

Objectif:

• Exécuter une vérification automatique avant l'envoi à l'aide de Cloud Build Cette vérification est exécutée des tests unitaires pour un DAG.
• Synchronisez les DAG de votre environnement Cloud Composer de développement avec les DAG de votre dépôt GitHub.

Avant de commencer

• Ce guide part du principe que vous travaillez avec Environnements Cloud Composer: un environnement de développement et environnement de production.

  Dans le cadre de ce guide, vous ne configurez qu'un pipeline CI/CD pour votre environnement de développement. Assurez-vous que l'environnement que vous utilisez n'est pas un environnement de production.

• Dans ce guide, nous partons du principe que vos DAG et leurs tests sont stockés dans un dépôt GitHub un dépôt de clés.

  L'exemple de pipeline CI/CD illustre le contenu d'un exemple de dépôt. Les DAG et les tests sont stockées dans le répertoire dags/, avec des fichiers d'exigences, les contraintes et les fichiers de configuration Cloud Build stockés au premier niveau. L'utilitaire de synchronisation des DAG et ses exigences se trouvent dans le utils.

Créer une tâche de vérification avant l'envoi et des tests unitaires

Le premier job Cloud Build exécute une vérification avant envoi, qui exécute les unités pour vos DAG.

Ajouter des tests unitaires

Si ce n'est pas déjà fait, tests unitaires pour vos DAG. Enregistrez ces tests à côté des DAG de votre dépôt, chacun avec le suffixe _test. Par exemple, le test du DAG dans example_dag.py est example_dag_test.py. Il s'agit des tests exécutés en tant que vérification préalable à l'envoi dans votre dépôt.

Créer une configuration YAML Cloud Build pour la vérification avant l'envoi

Dans votre dépôt, créez un fichier YAML nommé test-dags.cloudbuild.yaml qui configure votre tâche Cloud Build pour les vérifications préalables à l'envoi. Il comporte trois étapes :

1. Installez les dépendances requises par vos DAG.
2. Installez les dépendances nécessaires à vos tests unitaires.
3. Exécuter les tests du DAG

steps:
  # install dependencies
  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements.txt", "-c", "constraints.txt", "--user"]

  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements-test.txt", "--user"]

  # run in python 3.8 which is latest version in Cloud Composer
  - name: python:3.8-slim
    entrypoint: python3.8
    args: ["-m", "pytest", "-s", "dags/"]

Remarque: Cloud Build extrait l'image Python de Docker Hub et vous pouvez spécifiez le tag dans le champ name de chaque étape. Si vous utilisez python comme entrypoint, assurez-vous que votre entrypoint correspond à la version spécifiée dans votre balise, Sinon, entrypoint est défini par défaut sur latest.

Créer le déclencheur Cloud Build pour la vérification avant l'envoi

Suivez le guide Créer des dépôts à partir de GitHub pour créer un déclencheur basé sur une application GitHub avec les configurations suivantes :

• Nom : test-dags

• Événement : demande d'extraction

• Source – Dépôt: choisissez votre dépôt

• Source - Branche de base : ^main$ (remplacez main par le nom de la branche de base de votre dépôt, si nécessaire)

• Source – Contrôle des commentaires: non requis

  Attention :Si le contrôle des commentaires n'est pas obligatoire, tous les utilisateurs peut envoyer une demande d'extraction qui déclenche la compilation. Pour les dépôts plus volumineux ou ceux avec des contributeurs non approuvés, il est recommandé d'exiger le contrôle des commentaires.

• Configuration de la compilation : fichier de configuration Cloud Build : /test-dags.cloudbuild.yaml (chemin d'accès à votre fichier de compilation)

Créer un job de synchronisation des DAG et ajouter le script utilitaire des DAG

Configurez ensuite un job Cloud Build qui exécute un script utilitaire dédié aux DAG. Le script d'utilitaire de cette tâche synchronise vos DAG avec votre environnement Cloud Composer après leur fusion dans la branche principale de votre dépôt.

Remarque :Cloud Composer analyse tous les DAG selon un intervalle répété. Cela se produit même si les DAG n'ont pas changé depuis la dernière analyse. Pour cette raison, , cet utilitaire importe tous les DAG dans Cloud Composer, et pas seulement à celles qui ont changé.

Ajouter le script d'utilitaire DAG

Ajoutez le script d'utilitaire DAG à votre dépôt. Ce script d'utilitaire copie tous les fichiers DAG du répertoire dags/ de votre dépôt dans un répertoire temporaire, en ignorant tous les fichiers Python autres que les DAG. Le script utilise ensuite la bibliothèque cliente Cloud Storage pour importer tous les fichiers de ce répertoire temporaire dans le répertoire dags/ du bucket de votre environnement Cloud Composer.

from __future__ import annotations

import argparse
import glob
import os
from shutil import copytree, ignore_patterns
import tempfile

# Imports the Google Cloud client library
from google.cloud import storage


def _create_dags_list(dags_directory: str) -> tuple[str, list[str]]:
    temp_dir = tempfile.mkdtemp()

    # ignore non-DAG Python files
    files_to_ignore = ignore_patterns("__init__.py", "*_test.py")

    # Copy everything but the ignored files to a temp directory
    copytree(dags_directory, f"{temp_dir}/", ignore=files_to_ignore, dirs_exist_ok=True)

    # The only Python files left in our temp directory are DAG files
    # so we can exclude all non Python files
    dags = glob.glob(f"{temp_dir}/*.py")
    return (temp_dir, dags)


def upload_dags_to_composer(
    dags_directory: str, bucket_name: str, name_replacement: str = "dags/"
) -> None:
    """
    Given a directory, this function moves all DAG files from that directory
    to a temporary directory, then uploads all contents of the temporary directory
    to a given cloud storage bucket
    Args:
        dags_directory (str): a fully qualified path to a directory that contains a "dags/" subdirectory
        bucket_name (str): the GCS bucket of the Cloud Composer environment to upload DAGs to
        name_replacement (str, optional): the name of the "dags/" subdirectory that will be used when constructing the temporary directory path name Defaults to "dags/".
    """
    temp_dir, dags = _create_dags_list(dags_directory)

    if len(dags) > 0:
        # Note - the GCS client library does not currently support batch requests on uploads
        # if you have a large number of files, consider using
        # the Python subprocess module to run gsutil -m cp -r on your dags
        # See https://cloud.google.com/storage/docs/gsutil/commands/cp for more info
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)

        for dag in dags:
            # Remove path to temp dir
            dag = dag.replace(f"{temp_dir}/", name_replacement)

            try:
                # Upload to your bucket
                blob = bucket.blob(dag)
                blob.upload_from_filename(dag)
                print(f"File {dag} uploaded to {bucket_name}/{dag}.")
            except FileNotFoundError:
                current_directory = os.listdir()
                print(
                    f"{name_replacement} directory not found in {current_directory}, you may need to override the default value of name_replacement to point to a relative directory"
                )
                raise

    else:
        print("No DAGs to upload.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--dags_directory",
        help="Relative path to the source directory containing your DAGs",
    )
    parser.add_argument(
        "--dags_bucket",
        help="Name of the DAGs bucket of your Composer environment without the gs:// prefix",
    )

    args = parser.parse_args()

    upload_dags_to_composer(args.dags_directory, args.dags_bucket)

Créer une configuration YAML Cloud Build pour synchroniser des DAG

Dans votre dépôt, créez un fichier YAML nommé add-dags-to-composer.cloudbuild.yaml qui configure votre tâche Cloud Build pour la synchronisation des DAG. Il comporte deux étapes :

1. Installez les dépendances requises par le script de l'utilitaire des DAG.

2. Exécutez le script de l'utilitaire pour synchroniser les DAG de votre dépôt avec votre Cloud Composer.

steps:
  # install dependencies
  - name: python
    entrypoint: pip
    args: ["install", "-r", "utils/requirements.txt", "--user"]

  # run
  - name: python
    entrypoint: python
    args: ["utils/add_dags_to_composer.py", "--dags_directory=${_DAGS_DIRECTORY}", "--dags_bucket=${_DAGS_BUCKET}"]

Remarque: Cloud Build extrait l'image Python de Docker Hub et vous pouvez spécifiez le tag dans le champ name de chaque étape. Si vous utilisez python comme entrypoint, assurez-vous que votre entrypoint correspond à la version spécifiée dans votre balise. Sinon, entrypoint est défini par défaut sur latest.

Créer le déclencheur Cloud Build

Suivez les instructions pour créer des dépôts à partir de GitHub. pour créer un déclencheur basé sur une application GitHub avec les configurations suivantes:

• Nom : add-dags-to-composer

• Événement : Déployer sur une branche

• Source – Dépôt: choisissez votre dépôt

• Source : branche de base : ^main$ (remplacez main par le nom de votre branche de base du dépôt, si nécessaire)

• Source – Filtre de fichiers inclus (glob): dags/**

• Configuration de la compilation – Fichier de configuration Cloud Build: /add-dags-to-composer.cloudbuild.yaml (chemin d'accès à votre fichier de compilation)

Dans "Configuration avancée", ajoutez Variables de substitution:

• _DAGS_DIRECTORY : répertoire dans lequel se trouvent les dags dans votre dépôt. Si vous utilisez l'exemple de dépôt de ce guide, il s'agit de dags/.

• _DAGS_BUCKET : bucket Cloud Storage contenant le Répertoire dags/ de votre environnement Cloud Composer de développement environnement. Omettez le préfixe gs://. Par exemple, us-central1-example-env-1234ab56-bucket.

Attention :Si vous ne supprimez pas le suffixe dags/ du fichier Cloud Storage nom de bucket dans la variable de substitution _DAGS_BUCKET, le job de synchronisation du DAG échoue avec une erreur NotFound. En effet, dans Cloud Storage, dags/ est un dossier, qui est considéré comme faisant partie du et non à celui du bucket.

Tester votre pipeline CI/CD

Dans cette section, suivez un flux de développement de DAG qui utilise vos déclencheurs Cloud Build nouvellement créés.

Exécuter une tâche de présoumission

Créez une demande d'extraction vers votre branche principale pour tester votre build. Localisez votre sur la page. Cliquez sur Détails et sélectionnez Affichez plus d'informations sur Google Cloud Build pour consulter vos journaux de compilation dans le console Google Cloud.

Important :Cloud Build exécute des tests unitaires avec les packages spécifiés dans requirements.txt et requirements-test.txt, qui incluent une version publique d'Airflow. Votre environnement Cloud Composer peut avoir un profil un ensemble de packages PyPI et packages préinstallés. Nous vous recommandons d'exécuter un environnement Airflow local lorsque vous développez vos DAG.

Figure 2 : Capture d'écran de l'état de la vérification avant l'envoi de Cloud Build sur GitHub (cliquez pour agrandir)

Si la vérification avant l'envoi a échoué, consultez Résoudre les échecs de compilation.

Vérifier que votre DAG fonctionne dans votre environnement Cloud Composer de développement

Une fois votre demande d'extraction approuvée, fusionnez-la avec votre branche principale. Utilisez la console Google Cloud pour afficher les résultats de votre compilation. Si vous avez de nombreux les déclencheurs Cloud Build, vous pouvez filtrer vos compilations en fonction de leur nom add-dags-to-composer

Une fois la tâche de synchronisation Cloud Build réussie, le DAG synchronisé s'affiche dans votre environnement Cloud Composer de développement. Vous pouvez alors vérifier que le DAG fonctionne comme prévu.

Ajouter le DAG à votre environnement de production

Une fois que le DAG fonctionne comme prévu, ajoutez-le manuellement à votre environnement de production. Pour ce faire, importez le fichier DAG dans le répertoire dags/ du bucket de votre environnement Cloud Composer de production.

Si votre tâche de synchronisation de DAG a échoué ou si votre DAG ne se comporte pas comme prévu dans votre environnement Cloud Composer de développement, consultez la section Résoudre les échecs de compilation.

Résoudre les échecs de compilation

Cette section explique comment résoudre les scénarios courants d'échec de compilation.

Que faire si la vérification avant l'envoi a échoué ?

À partir de votre demande d'extraction, cliquez sur Détails et sélectionnez Affichez plus d'informations sur Google Cloud Build pour consulter vos journaux de compilation dans le console Google Cloud. Utilisez ces journaux pour vous aider à déboguer le problème lié à votre DAG. Une fois les problèmes résolus, validez la correction et déployez-la ou une autre branche. La vérification préalable à l'envoi s'exécute à nouveau, et vous pouvez continuer à itérer en utilisant les journaux comme outil de débogage.

Que faire si mon job de synchronisation du DAG a échoué ?

Utilisez la console Google Cloud pour afficher les résultats de votre compilation. Si vous disposez de nombreux déclencheurs Cloud Build, vous pouvez filtrer vos compilations en fonction du nom du déclencheur add-dags-to-composer. Examinez les journaux de la tâche de compilation et corrigez les erreurs. Si vous avez besoin d'aide supplémentaire pour résoudre les erreurs, utilisez les canaux d'assistance.

Que faire si mon DAG ne fonctionne pas correctement dans mon environnement Cloud Composer ?

Si votre DAG ne fonctionne pas comme prévu dans votre développement dans l'environnement Cloud Composer, ne promouvez pas manuellement le DAG l'environnement Cloud Composer de production. Effectuez plutôt l'une des opérations suivantes :

• Annulez la demande de tirage avec les modifications qui ont endommagé votre DAG pour le restaurer à l'état immédiatement antérieur à vos modifications (cela annule également tous les autres fichiers de cette demande de tirage).
• Créez une demande d'extraction pour annuler manuellement les modifications apportées au DAG défectueux.
• Créez une demande de tirage pour corriger les erreurs de votre DAG.

Le suivi de l'une de ces étapes déclenche un nouveau contrôle avant envoi et, lors de la fusion, la tâche de synchronisation du DAG.

Étape suivante

• Exécuter des environnements Airflow locaux
• Écrire des DAG
• Déclencher des DAG
• Tester les DAG