DAGs aus GitHub testen, synchronisieren und bereitstellen

Managed Airflow (Gen 3) | Managed Airflow (Gen 2) | Managed Airflow (Legacy Gen 1)

In dieser Anleitung wird beschrieben, wie Sie eine CI/CD-Pipeline erstellen, um DAGs aus Ihrem GitHub-Repository in Ihrer Managed Airflow-Umgebung zu testen, zu synchronisieren und bereitzustellen.

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

Übersicht über die CI/CD-Pipeline

Abbildung 1. Architekturdiagramm mit den Schritten des Ablaufs (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 zu prüfen, ob Ihr DAG gültig ist.

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

5. Cloud Build synchronisiert Ihre Entwicklung Managed Airflow-Umgebung mit diesen neuen Änderungen.

6. Sie prüfen, ob der DAG in Ihrer Entwicklung Umgebung wie erwartet funktioniert.

7. Wenn der DAG wie erwartet funktioniert, laden Sie ihn in Ihre Managed Airflow-Produktionsumgebung hoch.

Ziele

• Führen Sie mit Cloud Build eine automatische Vorabprüfung durch. Bei dieser Prüfung werden Einheitentests für einen DAG ausgeführt.
• Synchronisieren Sie DAGs in Ihrer Managed Service for Apache Airflow-Entwicklungsumgebung mit DAGs in Ihrem GitHub-Repository.

Hinweis

• In dieser Anleitung wird davon ausgegangen, dass Sie mit zwei identischen Managed Airflow-Umgebungen arbeiten: einer Entwicklungsumgebung und einer Produktionsumgebung.

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

• In dieser Anleitung wird davon ausgegangen, dass Sie Ihre DAGs und die zugehörigen 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. Die Dateien mit den Anforderungen, die Einschränkungsdatei und die Cloud Build-Konfigurationsdateien werden auf der obersten Ebene gespeichert. Das DAG-Synchronisierungsprogramm und seine Anforderungen befinden sich im Verzeichnis utils.

Job für die Vorabprüfung und Einheitentests erstellen

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

Unittests hinzufügen

Erstellen Sie Einheitentests für Ihre DAGs, falls noch nicht geschehen. Speichern Sie diese Tests zusammen mit den DAGs in Ihrem Repository und fügen Sie jeweils das Suffix _test hinzu. 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, in der Sie Ihren Cloud Build-Job für Vorabprüfungen konfigurieren. Sie enthält drei Schritte:

1. Installieren Sie die für Ihre DAGs erforderlichen Abhängigkeiten.
2. Installieren Sie die für Ihre Einheitentests erforderlichen Abhängigkeiten.
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 von Docker Hub ab. Sie können das Tag im name Feld jedes Schritts angeben. Wenn Sie python als entrypoint verwenden, muss der 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 zum Erstellen von Repositorys aus GitHub , um einen auf einer GitHub-App basierenden Trigger mit den folgenden Konfigurationen zu erstellen:

• Name: test-dags

• Ereignis: Pull-Anfrage

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

• Quelle – Basiszweig: ^main$ (ändern Sie main bei Bedarf in den Namen des Basiszweigs Ihres Repositorys).

• Quelle – Kommentarsteuerung: nicht erforderlich

  **Achtung** :Wenn Sie die Kommentarsteuerung auf „nicht erforderlich“ setzen, kann jeder Nutzer eine Pull-Anfrage einreichen, die den Build auslöst. Bei größeren Repositories oder Repositories mit nicht vertrauenswürdigen Mitwirkenden wird die Kommentarsteuerung empfohlen.

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

DAG-Synchronisierungsjob erstellen und DAG-Dienstprogramm hinzufügen

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

Hinweis: Managed Airflow parst alle DAGs in regelmäßigen Abständen. Das geschieht auch dann, wenn sich die DAGs seit dem letzten Parsen nicht geändert haben. Aus diesem Grund lädt dieses Dienstprogramm alle DAGs in Managed Airflow hoch, nicht nur die, die sich geändert haben.

DAG-Dienstprogramm hinzufügen

Fügen Sie das DAG-Dienstprogramm Ihrem Repository hinzu. Dieses Dienstprogramm kopiert alle DAG-Dateien im Verzeichnis dags/ Ihres Repositorys in ein temporäres Verzeichnis und ignoriert alle Python-Dateien, die keine DAGs sind. Das Skript verwendet dann die Cloud Storage-Clientbibliothek, um alle Dateien aus diesem temporären Verzeichnis in das Verzeichnis dags/ im Bucket Ihrer Managed Airflow-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 gcloud storage cp --recursive 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 Sie Ihren Cloud Build-Job zum Synchronisieren von DAGs konfigurieren. Sie enthält zwei Schritte:

1. Installieren Sie die für das DAG-Dienstprogramm erforderlichen Abhängigkeiten.

2. Führen Sie das Dienstprogramm aus, um die DAGs in Ihrem Repository mit Ihrer Managed Airflow-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 von Docker Hub ab. Sie können das Tag im name Feld jedes Schritts angeben. Wenn Sie python als entrypoint verwenden, muss der entrypoint mit der im Tag angegebenen Version übereinstimmen. Andernfalls wird entrypoint standardmäßig auf latest gesetzt.

Cloud Build-Trigger erstellen

Folgen Sie der Anleitung zum Erstellen von Repositorys aus GitHub , um einen auf einer GitHub-App basierenden Trigger mit den folgenden Konfigurationen zu erstellen:

• Name: add-dags-to-composer

• Ereignis: Push in einen Zweig

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

• Quelle – Basiszweig: ^main$ (ändern Sie main bei Bedarf in den Namen des Basiszweigs Ihres Repositorys).

• 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)

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

• _DAGS_DIRECTORY – das Verzeichnis, in dem sich die DAGs in Ihrem Repository befinden. Wenn Sie das Beispiel-Repository aus dieser Anleitung verwenden, ist es dags/.

• _DAGS_BUCKET – der Cloud Storage-Bucket, der das dags/ Verzeichnis in Ihrer Managed Airflow 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 Substitutionsvariablen _DAGS_BUCKET entfernen, schlägt der DAG-Synchronisierungsjob mit einem NotFound Fehler fehl. Das liegt daran, dass in Cloud Storage, dags/ ein Ordner ist, der als Teil des Objektnamens und nicht als Teil des Bucket-Namens betrachtet wird.

CI/CD-Pipeline testen

In diesem Abschnitt folgen Sie einem DAG-Entwicklungsablauf, bei dem die neu erstellten Cloud Build-Trigger verwendet werden.

Vorabprüfung ausführen

Erstellen Sie eine Pull-Anfrage für Ihren Hauptzweig, um Ihren Build zu testen. Suchen Sie auf der Seite nach Ihrer Vorabprüfung. 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 Einheitentests mit Paketen aus, die in requirements.txt und requirements-test.txt angegeben sind, einschließlich einer öffentlichen Version von Airflow. Ihre Managed Airflow-Umgebung hat möglicherweise eine andere Reihe von PyPI-Paketen und vorinstallierten Paketen. Wir empfehlen, beim Entwickeln Ihrer DAGs.

eine lokale Airflow-Umgebung auszuführen.

Abbildung 2. Screenshot des Status der Cloud Build Vorabprüfung auf GitHub (zum Vergrößern klicken)

Wenn die Vorabprüfung fehlgeschlagen ist, lesen Sie Build-Fehler beheben.

Prüfen, ob der DAG in Ihrer Entwicklungsumgebung 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 ausgeführt wurde, wird der synchronisierte DAG in Ihrer Managed Airflow-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. Laden Sie dazu die DAG-Datei in das Verzeichnis dags/ im Bucket Ihrer Managed Airflow Produktionsumgebung hoch.

Wenn der DAG-Synchronisierungsjob fehlgeschlagen ist oder sich der DAG in Ihrer Managed Airflow-Entwicklungsumgebung nicht wie erwartet verhält, lesen Sie Build-Fehler beheben.

Build-Fehler beheben

In diesem Abschnitt wird beschrieben, wie Sie häufige Build-Fehler beheben.

Was passiert, wenn die Vorabprüfung fehlgeschlagen ist?

Klicken Sie in Ihrer Pull-Anfrage 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 Logs können Sie das Problem mit Ihrem DAG beheben. Nachdem Sie die Probleme behoben haben, übernehmen Sie die Korrektur und übertragen Sie sie in Ihren Zweig. Die Vorabprüfung wird noch einmal ausgeführt und Sie können die Logs als Debugging-Tool verwenden.

Was passiert, wenn der 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. Prüfen Sie die Logs des Build-Jobs und beheben Sie die Fehler. Wenn Sie weitere Hilfe bei der Behebung der Fehler benötigen, nutzen Sie die Supportkanäle.

Was passiert, wenn mein DAG in meiner Managed Airflow-Umgebung nicht richtig funktioniert?

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

• Machen Sie die Pull-Anfrage mit den Änderungen, die den DAG beschädigt haben, rückgängig, um ihn in den Zustand unmittelbar vor den Änderungen zurückzuversetzen. Dadurch werden auch alle anderen Dateien in dieser Pull-Anfrage zurückgesetzt.
• Erstellen Sie eine neue Pull-Anfrage, um die Änderungen am beschädigten DAG manuell zurückzusetzen.
• Erstellen Sie eine neue Pull-Anfrage, um die Fehler in Ihrem DAG zu beheben.

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

Nächste Schritte

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