DAGs über GitHub testen, synchronisieren und bereitstellen

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

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

Wenn Sie nur Daten aus anderen Diensten synchronisieren möchten, finden Sie weitere Informationen unter Daten aus anderen Diensten übertragen.

CI/CD-Pipeline – Übersicht

Abbildung 1. Architekturdiagramm, das die Schritte des Ablaufs zeigt (zum Vergrößern klicken)

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 eine Pull-Anfrage für den Hauptzweig Ihres Repositorys.

3. Cloud Build führt Einheitentests aus, um die Gültigkeit des DAG zu prüfen.

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 die Cloud Composer-Produktionsumgebung hoch.

Lernziele

Ziel:

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

Hinweise

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

  Für die Zwecke dieser Anleitung konfigurieren Sie eine CI/CD-Pipeline nur für Ihre Entwicklungsumgebung. Achten Sie darauf, dass die verwendete Umgebung keine Produktionsumgebung ist.

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

  Die Beispiel-CI/CD-Pipeline zeigt den Inhalt eines Beispiel-Repositorys. DAGs und Tests werden im Verzeichnis dags/ gespeichert, wobei Anforderungsdateien, die Datei mit den Einschränkungen und Cloud Build-Konfigurationsdateien auf oberster Ebene gespeichert sind. Das Dienstprogramm für die DAG-Synchronisierung und seine Anforderungen befinden sich im Verzeichnis utils.

Job zur Vorabprüfung und Einheitentests erstellen

Der erste Cloud Build-Job führt eine Prüfung vor dem Senden aus, die Einheitentests für Ihre DAGs ausführt.

Einheitentests hinzufügen

Erstellen Sie gegebenenfalls Einheitentests für Ihre DAGs. 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. Dies sind die Tests, die als Vorabprüfung in Ihrem Repository 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, die den Cloud Build-Job für Prüfungen vor dem Senden konfiguriert. Er 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 Einheitentests 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 und Sie können das Tag im Feld name jedes Schritts angeben. Wenn Sie python als entrypoint verwenden, achten Sie darauf, dass Ihre entrypoint mit der im Tag angegebenen Version übereinstimmt. Andernfalls ist entrypoint standardmäßig auf latest gesetzt.

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

Folgen Sie dem Leitfaden Repositories aus GitHub erstellen, um einen auf GitHub basierenden Trigger mit den folgenden Konfigurationen zu erstellen:

• Name: test-dags

• Ereignis: Pull-Anfrage

• Quelle: Repository: Wählen Sie Ihr Repository aus.

• Source (Quelle): Basis-Branch: ^main$ (ändern Sie main in den Namen des Basiszweigs Ihres Repositorys, falls erforderlich.)

• Quelle – Kommentarsteuerung: nicht erforderlich

  Achtung :Wenn Sie die Kommentarsteuerung als nicht erforderlich festlegen, kann jeder Nutzer eine Pull-Anfrage senden, die den Build auslöst. Bei größeren Repositories oder Repositories mit nicht vertrauenswürdigen Mitwirkenden wird empfohlen, die Kommentarsteuerung zu verlangen.

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

DAG-Synchronisierungsjob erstellen und Dienstprogrammskript für DAGs hinzufügen

Konfigurieren Sie als Nächstes einen Cloud Build-Job, mit dem ein DAGs-Dienstprogrammskript ausgeführt wird. Das Dienstprogrammskript in diesem Job synchronisiert Ihre DAGs mit Ihrer Cloud Composer-Umgebung, nachdem sie mit dem Hauptzweig Ihres Repositorys zusammengeführt wurden.

Hinweis: Cloud Composer parst alle DAGs in einem wiederholten Intervall. Dies geschieht auch dann, wenn sich DAGs seit dem letzten Parsen nicht geändert haben. Aufgrund dieses Verhaltens lädt dieses Dienstprogramm alle DAGs in Cloud Composer hoch, nicht nur die, die geändert wurden.

DAGs-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 dabei alle Nicht-DAG-Python-Dateien. Das Skript 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)

YAML-Konfiguration für Cloud Build zum Synchronisieren von DAGs erstellen

Erstellen Sie in Ihrem Repository eine YAML-Datei mit dem Namen add-dags-to-composer.cloudbuild.yaml, die den Cloud Build-Job für die Synchronisierung von DAGs konfiguriert. Dieser enthält zwei Schritte:

1. Installieren Sie die Abhängigkeiten, die vom Skript des Dienstprogramms für DAGs benötigt werden.

2. Führen Sie das Dienstprogrammskript 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 und Sie können das Tag im Feld name jedes Schritts angeben. Wenn Sie python als entrypoint verwenden, achten Sie darauf, dass Ihre entrypoint mit der im Tag angegebenen Version übereinstimmt. Andernfalls ist entrypoint standardmäßig auf latest gesetzt.

Cloud Build-Trigger erstellen

Folgen Sie dem Leitfaden Repositories aus GitHub erstellen, um einen auf GitHub basierenden Trigger mit den folgenden Konfigurationen zu erstellen:

• Name: add-dags-to-composer

• Ereignis: Push zu einem Zweig

• Quelle: Repository: Wählen Sie Ihr Repository aus.

• Source (Quelle): Basis-Branch: ^main$ (ändern Sie main in den Namen des Basiszweigs Ihres Repositorys, falls erforderlich.)

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

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

In der erweiterten Konfiguration fügen Sie zwei Substitutionsvariablen hinzu:

• _DAGS_DIRECTORY ist das Verzeichnis, in dem sich „dags“ in Ihrem Repository befindet. Wenn Sie das Beispiel-Repository aus dieser Anleitung verwenden, ist es 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 Substitutionsvariable _DAGS_BUCKET entfernen, schlägt der DAG-Synchronisierungsjob mit dem Fehler NotFound fehl. Dies liegt daran, dass dags/ in Cloud Storage ein Ordner ist, der als Teil des Objektnamens und nicht als Teil des Bucket-Namens angesehen wird.

CI/CD-Pipeline testen

Folgen Sie in diesem Abschnitt dem DAG-Entwicklungsablauf, der Ihre neu erstellten Cloud Build-Trigger verwendet.

Vorabsendejob ausführen

Erstellen Sie eine Pull-Anfrage an Ihren Hauptzweig, um den Build zu testen. Suchen Sie auf der Seite nach der Vorabprüfung. Klicken Sie auf Details und wählen Sie Weitere Details zu Google Cloud Build ansehen aus, um Ihre Build-Logs in der Google Cloud Console anzusehen.

Wichtig: Cloud Build führt Einheitentests mit Paketen aus, die in requirements.txt und requirements-test.txt angegeben sind, die eine öffentliche Version von Airflow enthalten. Ihre Cloud Composer-Umgebung enthält möglicherweise andere PyPI-Pakete und vorinstallierte Pakete. Wir empfehlen, während der Entwicklung von DAGs eine lokale Airflow-Umgebung auszuführen.

Abbildung 2. Screenshot des Cloud Build-Vorabprüfungsstatus auf GitHub (zum Vergrößern klicken)

Wenn die Prüfung vor dem Senden fehlgeschlagen ist, lesen Sie den Abschnitt Build-Fehler beheben.

Prüfen, ob der DAG in der Entwicklungsumgebung von Cloud Composer funktioniert

Nachdem Ihre Pull-Anfrage genehmigt wurde, führen Sie sie mit Ihrem Hauptzweig zusammen. Verwenden Sie die Google Cloud Console, um Ihre Build-Ergebnisse anzusehen. 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 der Cloud Composer-Entwicklungsumgebung angezeigt. Dort können Sie prüfen, ob der DAG wie erwartet funktioniert.

DAG der Produktionsumgebung hinzufügen

Wenn der DAG wie erwartet funktioniert, fügen Sie ihn manuell Ihrer Produktionsumgebung hinzu. Dazu müssen Sie die DAG-Datei in das Verzeichnis dags/ im Bucket Ihrer Cloud Composer-Produktionsumgebung hochladen.

Wenn der DAG-Synchronisierungsjob fehlgeschlagen ist oder der DAG in Ihrer Cloud Composer-Entwicklungsumgebung nicht wie erwartet funktioniert, lesen Sie Build-Fehler beheben.

Build-Fehler beheben

In diesem Abschnitt wird erläutert, wie Sie häufige Build-Fehlerszenarien beheben.

Was passiert, wenn meine Vorabprüfung fehlgeschlagen ist?

Klicken Sie in der Pull-Anfrage auf Details und wählen Sie Weitere Details zu Google Cloud Build ansehen aus, um Ihre Build-Logs in der Google Cloud Console anzusehen. Verwenden Sie diese Logs, um das Problem mit Ihrem DAG zu beheben. Sobald Sie die Probleme behoben haben, übernehmen Sie die Korrektur und senden Sie die Änderungen an Ihren Zweig. Die Vorabprüfung wird noch einmal ausgeführt und Sie können mit den Logs als Debugging-Tool weiter iterieren.

Was passiert, wenn mein DAG-Synchronisierungsjob fehlgeschlagen ist?

Verwenden Sie die Google Cloud Console, um Ihre Build-Ergebnisse anzusehen. Wenn Sie viele Cloud Build-Trigger haben, können Sie Ihre Builds nach dem Triggernamen add-dags-to-composer filtern. Sehen Sie sich die Logs des Build-Jobs an und beheben Sie die Fehler. Wenn Sie zusätzliche Hilfe beim Beheben der Fehler benötigen, nutzen Sie die Supportkanäle.

Was passiert, wenn mein DAG in meiner Cloud Composer-Umgebung nicht ordnungsgemäß funktioniert?

Wenn der DAG in Ihrer Cloud Composer-Entwicklungsumgebung nicht wie erwartet funktioniert, stufen Sie den DAG nicht manuell zur Cloud Composer-Produktionsumgebung hoch. Führen Sie stattdessen einen der folgenden Schritte aus:

• Setzen Sie die Pull-Anfrage mit den Änderungen zurück, die den DAG unterbrochen haben, um ihn in den Status unmittelbar vor Ihren Änderungen wiederherzustellen. Dadurch werden auch alle anderen Dateien in dieser Pull-Anfrage zurückgesetzt.
• Erstellen Sie eine neue Pull-Anfrage, um Änderungen am fehlerhaften DAG manuell zurückzusetzen.
• Erstellen Sie eine neue Pull-Anfrage, um die Fehler im DAG zu beheben.

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

Nächste Schritte

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