DAGs aus GitHub testen, synchronisieren und bereitstellen

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

In diesem Leitfaden wird erläutert, wie Sie eine CI/CD-Pipeline erstellen, um DAGs aus Ihrem GitHub-Repository zu testen, zu synchronisieren und in Ihrer Cloud Composer-Umgebung bereitzustellen.

Wenn Sie nur Daten aus anderen Diensten synchronisieren möchten, lesen Sie den Hilfeartikel Daten aus anderen Diensten übertragen.

CI/CD-Pipeline – Übersicht

Abbildung 1. Architekturdiagramm mit den Schritten des Ablaufs (zum Vergrößern anklicken)

Die CI/CD-Pipeline zum Testen, Synchronisieren und Bereitstellen von DAGs umfasst die folgenden Schritte:

1. Sie nehmen eine Änderung an einem DAG vor und übertragen diese Änderung in einen Entwicklungszweig in Ihrem Repository.

2. Sie öffnen einen Pull-Request für den Hauptzweig Ihres Repositorys.

3. Cloud Build führt Unit-Tests aus, um zu prüfen, ob Ihre DAG gültig ist.

4. Ihre Pull-Anfrage wird genehmigt und mit dem Hauptzweig Ihres Repositorys zusammengeführt.

5. Cloud Build synchronisiert Ihre Cloud Composer-Entwicklungsumgebung mit diesen neuen Änderungen.

6. Sie prüfen, ob sich der DAG in Ihrer Entwicklungsumgebung wie erwartet verhält.

7. Wenn der DAG wie erwartet funktioniert, laden Sie ihn in Ihre Produktions-Cloud Composer-Umgebung hoch.

Lernziele

Ziel:

• Führen Sie mit Cloud Build eine automatisierte Vorabprüfung durch. Bei dieser Prüfung werden Unit-Tests für einen DAG ausgeführt.
• Synchronisieren Sie DAGs in Ihrer Cloud Composer-Entwicklungsumgebung mit DAGs in Ihrem GitHub-Repository.

Hinweis

• In diesem Leitfaden wird davon ausgegangen, dass Sie mit zwei identischen Cloud Composer-Umgebungen arbeiten: einer Entwicklungs- und einer Produktionsumgebung.

  In diesem Leitfaden konfigurieren Sie eine CI/CD-Pipeline nur für Ihre Entwicklungsumgebung. Achten Sie darauf, dass es sich nicht um eine Produktionsumgebung handelt.

• In dieser Anleitung wird davon ausgegangen, dass Sie Ihre DAGs und ihre Tests in einem GitHub-Repository gespeichert haben.

  Die Beispiel-CI/CD-Pipeline veranschaulicht den Inhalt eines Beispiel-Repositorys. DAGs und Tests werden im Verzeichnis dags/ gespeichert. An der obersten Ebene befinden sich die Anforderungsdateien, die Datei mit den Einschränkungen und die Cloud Build-Konfigurationsdateien. Das DAG-Synchronisierungstool und die zugehörigen Anforderungen befinden sich im Verzeichnis utils.

  Diese Struktur kann für Airflow 1-, Airflow 2-, Cloud Composer 1- und Cloud Composer 2-Umgebungen verwendet werden.

Job für die Vorabprüfung und Unit-Tests erstellen

Der erste Cloud Build-Job führt eine Vorabprüfung durch, bei der Unit-Tests für Ihre DAGs ausgeführt werden.

Unittests hinzufügen

Erstellen Sie Einheitstests für Ihre DAGs, falls Sie das noch nicht getan haben. Speichern Sie diese Tests zusammen mit den DAGs in Ihrem Repository, jeweils mit dem Suffix _test. Die Testdatei für den DAG in example_dag.py ist beispielsweise example_dag_test.py. Das sind die Tests, die in Ihrem Repository als Presubmit-Prüfung ausgeführt werden.

Cloud Build-YAML-Konfiguration für die Vorabprüfung erstellen

Erstellen Sie in Ihrem Repository eine YAML-Datei mit dem Namen test-dags.cloudbuild.yaml, in der Ihr Cloud Build-Job für Vorabprüfungen konfiguriert wird. Es umfasst drei Schritte:

1. Installieren Sie die Abhängigkeiten, die für Ihre DAGs erforderlich sind.
2. Installieren Sie die Abhängigkeiten, die für Ihre Unit-Tests erforderlich sind.
3. Führen Sie die DAG-Tests aus.

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/"]

Hinweis: Cloud Build ruft das Python-Image aus Docker Hub ab. Sie können das Tag in jedem Schritt im Feld name angeben. Wenn Sie python als entrypoint verwenden, muss entrypoint mit der im Tag angegebenen Version übereinstimmen. Andernfalls wird entrypoint standardmäßig auf latest gesetzt.

Cloud Build-Trigger für die Vorabprüfung erstellen

Folgen Sie der Anleitung unter Repositories aus GitHub erstellen, um einen GitHub App-basierten Trigger mit den folgenden Konfigurationen zu erstellen:

• Name: test-dags

• Ereignis: Pull-Anfrage

• Source (Quelle) – Repository: Wählen Sie Ihr Repository aus.

• Quelle – Basis-Branch: ^main$ (ändern Sie main bei Bedarf in den Namen des Basis-Branches Ihres Repositorys)

• Quelle – Kommentarsteuerung: nicht erforderlich

  Achtung :Wenn Sie die Kommentarsteuerung auf „Nicht erforderlich“ festlegen, kann jeder Nutzer einen Pull-Request einreichen, der den Build auslöst. Für größere Repositories oder Repositories mit nicht vertrauenswürdigen Mitwirkenden wird die Kommentarsteuerung empfohlen.

• Build-Konfiguration – Cloud Build-Konfigurationsdatei:/test-dags.cloudbuild.yaml (Pfad zur Build-Datei)

DAG-Synchronisierungsjob erstellen und DAG-Dienstprogrammscript hinzufügen

Konfigurieren Sie als Nächstes einen Cloud Build-Job, der ein DAG-Dienstprogrammskript ausführt. Das Dienstprogramm-Script in diesem Job synchronisiert Ihre DAGs mit Ihrer Cloud Composer-Umgebung, nachdem sie mit dem Hauptzweig in Ihrem Repository zusammengeführt wurden.

Hinweis: Cloud Composer parset alle DAGs in einem wiederholten Intervall. Das passiert auch dann, wenn sich die DAGs seit der letzten Analyse nicht geändert haben. Aus diesem Grund lädt dieses Dienstprogramm alle DAGs in Cloud Composer hoch, nicht nur die, die sich geändert haben.

DAG-Dienstprogrammskript hinzufügen

Fügen Sie Ihrem Repository das DAG-Dienstprogrammskript hinzu. Dieses Dienstprogrammskript kopiert alle DAG-Dateien im Verzeichnis dags/ Ihres Repositorys in ein temporäres Verzeichnis und ignoriert alle Python-Dateien, die keine DAG-Dateien sind. Das Script verwendet dann die Cloud Storage-Clientbibliothek, um alle Dateien aus diesem temporären Verzeichnis in das Verzeichnis dags/ im Bucket Ihrer Cloud Composer-Umgebung hochzuladen.

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)

Cloud Build-YAML-Konfiguration zum Synchronisieren von DAGs erstellen

Erstellen Sie in Ihrem Repository eine YAML-Datei mit dem Namen add-dags-to-composer.cloudbuild.yaml, in der Ihr Cloud Build-Job für die Synchronisierung von DAGs konfiguriert wird. Sie besteht aus zwei Schritten:

1. Installieren Sie die Abhängigkeiten, die für das DAG-Dienstprogrammskript erforderlich sind.

2. Führen Sie das Dienstprogramm-Script aus, um die DAGs in Ihrem Repository mit Ihrer Cloud Composer-Umgebung zu synchronisieren.

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}"]

Hinweis: Cloud Build ruft das Python-Image aus Docker Hub ab. Sie können das Tag in jedem Schritt im Feld name angeben. Wenn Sie python als entrypoint verwenden, muss entrypoint mit der im Tag angegebenen Version übereinstimmen. Andernfalls wird entrypoint standardmäßig auf latest gesetzt.

Cloud Build-Trigger erstellen

Folgen Sie der Anleitung unter Repositories aus GitHub erstellen, um einen GitHub App-basierten Trigger mit den folgenden Konfigurationen zu erstellen:

• Name: add-dags-to-composer

• Ereignis: Per Push-Befehl an Zweig übertragen

• Source (Quelle) – Repository: Wählen Sie Ihr Repository aus.

• Quelle – Basis-Branch: ^main$ (ändern Sie main bei Bedarf in den Namen des Basis-Branches Ihres Repositorys)

• Quelle – Filter für enthaltene Dateien (glob): dags/**

• Build-Konfiguration – Cloud Build-Konfigurationsdatei:/add-dags-to-composer.cloudbuild.yaml (Pfad zur Build-Datei)

Fügen Sie in der erweiterten Konfiguration zwei Ersatzvariablen hinzu:

• _DAGS_DIRECTORY: das Verzeichnis, in dem sich die DAGs in Ihrem Repository befinden. Wenn Sie das Beispiel-Repository aus dieser Anleitung verwenden, lautet der Name dags/.

• _DAGS_BUCKET: Der Cloud Storage-Bucket, der das Verzeichnis dags/ in Ihrer Cloud Composer-Entwicklungsumgebung enthält. Lassen Sie das Präfix gs:// weg. Beispiel: us-central1-example-env-1234ab56-bucket.

Achtung :Wenn Sie das Suffix dags/ nicht aus dem Namen des Cloud Storage-Buckets in der _DAGS_BUCKET-Ersetzungsvariablen entfernen, schlägt der DAG-Synchronisierungsjob mit einem NotFound-Fehler fehl. Das liegt daran, dass dags/ in Cloud Storage ein Ordner ist, der als Teil des Objektnamens und nicht des Bucket-Namens betrachtet wird.

CI/CD-Pipeline testen

In diesem Abschnitt führen Sie einen DAG-Entwicklungsablauf aus, bei dem Ihre neu erstellten Cloud Build-Trigger verwendet werden.

Job vor dem Einreichen ausführen

Erstellen Sie einen Pull-Request für Ihren Hauptzweig, um Ihren Build zu testen. Suchen Sie auf der Seite nach der Prüfung vor dem Senden. Klicken Sie auf Details und wählen Sie Weitere Details in Google Cloud Build anzeigen aus, um Ihre Build-Logs in der Google Cloud Console aufzurufen.

Wichtig: Cloud Build führt Unit-Tests mit den in requirements.txt und requirements-test.txt angegebenen Paketen aus, die eine öffentliche Version von Airflow enthalten. Ihre Cloud Composer-Umgebung enthält möglicherweise eine andere Reihe von PyPI-Paketen und vorinstallierten Paketen. Wir empfehlen, während der Entwicklung Ihrer DAGs eine lokale Airflow-Umgebung auszuführen.

Abbildung 2. Screenshot des Cloud Build-Status vor dem Einreichen auf GitHub (zum Vergrößern anklicken)

Wenn die Prüfung vor dem Einreichen fehlgeschlagen ist, lesen Sie den Hilfeartikel Fehlerbehebung bei Build-Fehlern.

Prüfen, ob Ihr DAG in Ihrer Cloud Composer-Entwicklungsumgebung funktioniert

Nachdem Ihr Pull-Request genehmigt wurde, führen Sie ihn mit Ihrem Hauptzweig zusammen. In der Google Cloud Console können Sie sich Ihre Build-Ergebnisse ansehen. Wenn Sie viele Cloud Build-Trigger haben, können Sie Ihre Builds nach dem Triggernamen add-dags-to-composer filtern.

Nachdem der Cloud Build-Synchronisierungsjob erfolgreich war, wird der synchronisierte DAG in Ihrer Cloud Composer-Entwicklungsumgebung angezeigt. Dort können Sie prüfen, ob der DAG wie erwartet funktioniert.

DAG in die Produktionsumgebung aufnehmen

Wenn der DAG wie erwartet funktioniert, fügen Sie ihn Ihrer Produktionsumgebung manuell hinzu. Laden Sie dazu die DAG-Datei in das Verzeichnis dags/ des Buckets Ihrer Produktions-Cloud Composer-Umgebung hoch.

Wenn der DAG-Synchronisierungsjob fehlgeschlagen ist oder sich der DAG in Ihrer Cloud Composer-Entwicklungsumgebung nicht wie erwartet verhält, lesen Sie den Hilfeartikel Fehler bei der Build-Ausführung beheben.

Build-Fehler beheben

In diesem Abschnitt wird beschrieben, wie Sie häufige Szenarien für Buildfehler beheben.

Was passiert, wenn die Vorabprüfung fehlgeschlagen ist?

Klicken Sie in Ihrem Pull-Request auf Details und wählen Sie Weitere Details in Google Cloud Build anzeigen aus, um Ihre Build-Logs in der Google Cloud Console aufzurufen. Anhand dieser Protokolle können Sie das Problem mit Ihrem DAG beheben. Nachdem Sie die Probleme behoben haben, committen Sie die Korrektur und pushen Sie sie an Ihren Branch. Die Vorabprüfung wird noch einmal ausgeführt und Sie können mithilfe der Protokolle weiter iterieren.

Was passiert, wenn der DAG-Synchronisierungsauftrag fehlschlägt?

In der Google Cloud Console können Sie sich Ihre Build-Ergebnisse ansehen. Wenn Sie viele Cloud Build-Trigger haben, können Sie Ihre Builds nach dem Triggernamen add-dags-to-composer filtern. Prüfen Sie die Protokolle des Build-Jobs und beheben Sie die Fehler. Wenn Sie weitere Unterstützung beim Beheben der Fehler benötigen, nutzen Sie die Supportkanäle.

Was kann ich tun, wenn mein DAG in meiner Cloud Composer-Umgebung nicht richtig funktioniert?

Wenn Ihr DAG in Ihrer Cloud Composer-Entwicklungsumgebung nicht wie erwartet funktioniert, sollten Sie ihn nicht manuell in Ihre Cloud Composer-Produktionsumgebung hochstufen. Führen Sie stattdessen einen der folgenden Schritte aus:

• Stellen Sie den Pull-Request mit den Änderungen zurück, die Ihren DAG beschädigt haben, um ihn in den Zustand vor Ihren Änderungen wiederherzustellen. Dadurch werden auch alle anderen Dateien in diesem Pull-Request rückgängig gemacht.
• Erstellen Sie einen neuen Pull-Request, um die Änderungen an der fehlerhaften DAG manuell rückgängig zu machen.
• Erstellen Sie einen neuen Pull-Request, um die Fehler in Ihrem DAG zu beheben.

Wenn Sie einen dieser Schritte ausführen, wird eine neue Vorabprüfung ausgelöst und nach dem Zusammenführen der DAG-Synchronisierungsjob.

Nächste Schritte

• Lokale Airflow-Umgebungen ausführen
• DAGs schreiben
• DAGs planen und auslösen
• DAGs testen