Airflow-DAGs testen

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Vor dem Bereitstellen von DAGs für die Produktion können Sie Airflow-CLI-Unterbefehle ausführen, um DAG-Code im selben Kontext zu analysieren, in dem der DAG ausgeführt wird.

Während der DAG-Erstellung testen

Sie können eine einzelne Aufgabeninstanz lokal ausführen und die Logausgabe durchsehen. Auf diese Weise lassen sich Syntax- und Aufgabenfehler feststellen. Bei lokalen Tests werden weder Abhängigkeiten noch der Kommunikationsstatus mit der Datenbank geprüft.

Wir empfehlen, die DAGs in einem Ordner namens data/test in Ihrer Testumgebung zu speichern.

Testverzeichnis erstellen

Erstellen Sie im Bucket Ihrer Umgebung ein Testverzeichnis und kopieren Sie Ihre DAGs in dieses Verzeichnis.

gcloud storage cp BUCKET_NAME/dags \
  BUCKET_NAME/data/test --recursive

Ersetzen Sie Folgendes:

  • BUCKET_NAME: Der Name des Buckets, der mit Ihrer Cloud Composer-Umgebung verknüpft ist.

Beispiel:

gcloud storage cp gs://us-central1-example-environment-a12bc345-bucket/dags \
  gs://us-central1-example-environment-a12bc345-bucket/data/test --recursive

Weitere Informationen zum Hochladen von DAGs finden Sie unter DAGs hinzufügen und aktualisieren.

Auf Syntaxfehler prüfen

Geben Sie den folgenden gcloud-Befehl ein, um nach Syntaxfehlern in DAGs zu suchen, die Sie in den Ordner /data/test hochgeladen haben:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  dags list -- --subdir /home/airflow/gcs/data/test

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  list_dags -- -sd /home/airflow/gcs/data/test

Ersetzen Sie Folgendes:

  • ENVIRONMENT_NAME: der Name der Umgebung.
  • ENVIRONMENT_LOCATION: die Region, in der sich die Umgebung befindet.

Auf Aufgabenfehler prüfen

Prüfung auf aufgabenspezifische Fehler in DAGs, die Sie in den /data/test hochgeladen haben führen Sie den folgenden gcloud-Befehl aus:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  DAG_ID TASK_ID \
  DAG_EXECUTION_DATE

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  test -- -sd /home/airflow/gcs/data/test DAG_ID \
  TASK_ID DAG_EXECUTION_DATE

Ersetzen Sie Folgendes:

  • ENVIRONMENT_NAME: der Name der Umgebung.
  • ENVIRONMENT_LOCATION: die Region, in der sich die Umgebung befindet.
  • DAG_ID: die ID des DAG.
  • TASK_ID: die ID der Aufgabe.
  • DAG_EXECUTION_DATE: das Ausführungsdatum des DAG. Dieses Datum wird für Vorlagenzwecke verwendet. Der DAG wird sofort ausgeführt, egal welches Datum Sie hier angeben.

Beispiel:

Airflow 2 

gcloud composer environments run \
  example-environment \
  --location us-central1 \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  hello_world print_date 2021-04-22

Airflow 1 

gcloud composer environments run example-environment \
  --location us-central1 \
  test -- -sd /home/airflow/gcs/data/test \
  hello_world print_date 2021-04-22

Bereitgestellten DAG aktualisieren und testen

So testen Sie Aktualisierungen Ihrer DAGs in Ihrer Testumgebung:

  1. Kopieren Sie den bereitgestellten DAG, den Sie aktualisieren möchten, in data/test.
  2. Aktualisieren Sie den DAG.
  3. Testen Sie den DAG.
    1. Auf Syntaxfehler prüfen.
    2. Auf Aufgabenfehler prüfen.
  4. Achten Sie darauf, dass der DAG erfolgreich ausgeführt wird.
  5. Deaktivieren Sie den DAG in Ihrer Testumgebung.
    1. Wechseln Sie dazu in der Airflow-Benutzeroberfläche zur Seite "DAGs".
    2. Wenn der zu bearbeitende DAG ständig ausgeführt wird, deaktivieren Sie ihn.
    3. Zum Beschleunigen ausstehender Aufgaben klicken Sie auf die jeweilige Aufgabe und markieren diese als erfolgreich.
  6. Stellen Sie den DAG in Ihrer Produktionsumgebung bereit.
    1. Deaktivieren Sie den DAG in Ihrer Produktionsumgebung.
    2. Laden Sie den aktualisierten DAG in den Ordner dags/ Ihrer Produktionsumgebung.

FAQ zum Testen von DAGs

Wie isoliere ich DAG-Ausführungen in meinen Produktions- und Testumgebungen?

Beispiel: Airflow enthält ein globales Repository mit Quellcode im Ordner dags/, das alle DAG-Ausführungen gemeinsam verwenden. Sie möchten Quellcode in der Produktions- oder Testumgebung ändern können, ohne dass sich dies auf ausgeführte DAGs auswirkt.

Airflow bietet keine strikte DAG-Isolation. Es empfiehlt sich, die Produktions- und Testumgebungen von Cloud Composer getrennt zu halten, damit Test-DAGs nicht in Konflikt mit Produktions-DAGs geraten.

Wie kann ich DAG-Interferenzen vermeiden, wenn ich Integrationstests von verschiedenen GitHub-Branches ausführe?

Verwenden Sie eindeutige Aufgabennamen, um Interferenzen zu verhindern. Beispielsweise können Sie dem Namen eines Zweigs die Aufgaben-IDs voranstellen.

Welche Best Practice gilt für Integrationstests mit Airflow?

Wir empfehlen, für Integrationstests mit Airflow eine dedizierte Umgebung zu verwenden. Eine Möglichkeit, erfolgreiche DAG-Ausführung aufzuzeigen, besteht darin, Daten in eine Datei in einem Cloud Storage-Ordner zu schreiben und den Inhalt in eigenen Integrationstestläufen zu prüfen.

Wie arbeite ich effizient mit anderen DAG-Contributors zusammen?

Für jeden Contributor kann ein eigenes Unterverzeichnis im Ordner data/ für die Entwicklung angelegt werden.

Dem Ordner data/ hinzugefügte DAGs werden nicht automatisch vom Airflow-Planer oder vom Webserver erfasst.

Mit DAG-Contributors können manuelle DAG-Ausführungen erstellt werden. Dazu verwenden sie den Befehl gcloud composer environments run und den Unterbefehl test mit dem Flag --subdir, um das Entwicklungsverzeichnis des Contributors anzugeben.

Beispiel:

Airflow 2 

gcloud composer environments run test-environment-name \
  tasks test -- dag-id task-id execution-date \
  --subdir /home/airflow/gcs/data/alice_dev

Airflow 1 

gcloud composer environments run test-environment-name \
  test -- dag-id task-id execution-date \
  --subdir /home/airflow/gcs/data/alice_dev

Wie sorge ich dafür, dass meine Bereitstellungsumgebung mit meiner Produktionsumgebung synchron bleibt?

So verwalten Sie den Zugriff:

So geht die Bereitstellung von der Entwicklung in die Produktion:

  • Achten Sie auf eine konsistente Konfiguration, z. B. bei Umgebungsvariablen und PyPI-Paketen.

  • Achten Sie auf konsistente DAG-Argumente. Wir empfehlen die Verwendung von Airflow-Makros und -Variablen, um eine Hartcodierung zu vermeiden.

    Beispiel:

    Airflow 2 

    gcloud composer environments run test-environment-name \
  variables set -- DATA_ENDPOINT_KEY DATA_ENDPOINT_VALUE

    Airflow 1 

    gcloud composer environments run test-environment-name \
  variables -- --set DATA_ENDPOINT_KEY DATA_ENDPOINT_VALUE

Nächste Schritte