Probleme bei der Aufgabenplanung beheben

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

In dieser Anleitung erfahren Sie, wie Sie Probleme bei der Aufgabenplanung und beim Parsen diagnostizieren und beheben, die zu Fehlfunktionen des Schedulers, Parsing-Fehlern und Latenz sowie zu Aufgabenfehlern führen.

Einführung

Der Airflow-Planer wird hauptsächlich von zwei Faktoren beeinflusst: der Aufgabenplanung und dem DAG-Parsing. Probleme bei einem dieser Faktoren können sich negativ auf den Zustand und die Leistung der Umgebung auswirken.

Manchmal werden zu viele Aufgaben gleichzeitig geplant. In diesem Fall ist die Warteschlange voll und Aufgaben bleiben im Status „Geplant“ oder werden nach dem Eintragen in die Warteschlange neu geplant, was zu Aufgabenfehlern und Leistungslatenz führen kann.

Ein weiteres häufiges Problem sind Parsing-Latenz und Fehler, die durch die Komplexität eines DAG-Codes verursacht werden. Wenn DAG-Code beispielsweise Airflow-Variablen auf der obersten Ebene des Codes enthält, kann dies zu Parsing-Verzögerungen, Datenbanküberlastung, Planungsfehlern und DAG-Timeouts führen.

In diesem Tutorial diagnostizieren Sie die Beispiel-DAGs und erfahren, wie Sie Probleme bei der Planung und beim Parsen beheben, die DAG-Planung verbessern und Ihren DAG-Code und Ihre Umgebungskonfigurationen optimieren, um die Leistung zu verbessern.

Ziele

In diesem Abschnitt werden die Ziele für die Beispiele in dieser Anleitung aufgeführt.

Beispiel: Fehlfunktion des Schedulers und Latenz aufgrund hoher Aufgabenkonkurrenz

• Laden Sie die Beispiel-DAG hoch, die mehrmals gleichzeitig ausgeführt wird, und diagnostizieren Sie die Fehlfunktion des Schedulers und Latenzprobleme mit Cloud Monitoring.

• Optimieren Sie Ihren DAG-Code, indem Sie die Aufgaben konsolidieren, und bewerten Sie die Auswirkungen auf die Leistung.

• Verteilen Sie die Aufgaben gleichmäßiger über die Zeit und bewerten Sie die Auswirkungen auf die Leistung.

• Optimieren Sie Ihre Airflow- und Umgebungskonfigurationen und bewerten Sie die Auswirkungen.

Beispiel: Fehler beim Parsen von DAGs und Latenz aufgrund von komplexem Code

• Laden Sie die Beispiel-DAG mit Airflow-Variablen hoch und beheben Sie Parsing-Probleme mit Cloud Monitoring.

• Optimieren Sie den DAG-Code, indem Sie Airflow-Variablen auf der obersten Ebene des Codes vermeiden, und bewerten Sie die Auswirkungen auf die Parsing-Zeit.

• Optimieren Sie die Airflow-Konfigurationen und Umgebungskonfigurationen und bewerten Sie die Auswirkungen auf die Parsing-Zeit.

Kosten

In dieser Anleitung werden die folgenden kostenpflichtigen Komponenten von Google Cloudverwendet:

• Managed Airflow (siehe zusätzliche Kosten)
• Cloud Monitoring

Nach Abschluss dieser Anleitung können Sie weitere Kosten durch Löschen von erstellten Ressourcen vermeiden. Weitere Informationen finden Sie unter Bereinigen.

Hinweis

In diesem Abschnitt werden die Aktionen beschrieben, die erforderlich sind, bevor Sie mit dem Tutorial beginnen.

Projekt erstellen und konfigurieren

Für diese Anleitung benötigen Sie ein Google Cloud-Projekt. Konfigurieren Sie das Projekt so:

1. Wählen Sie in der Google Cloud Console ein Projekt aus oder erstellen Sie eines:

   Zur Projektauswahl

2. Die Abrechnung für Ihr Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.

3. Achten Sie darauf, dass der Nutzer Ihres Google Cloud Projekts die folgenden Rollen hat, um die erforderlichen Ressourcen zu erstellen:

   • Administrator für Umgebung und Storage-Objekte (roles/composer.environmentAndStorageObjectAdmin)
   • Compute-Administrator (roles/compute.admin)

Die APIs für Ihr Projekt aktivieren

Aktivieren Sie die Managed Airflow API, falls sie noch nicht aktiviert ist.

API aktivieren

Managed Airflow-Umgebung erstellen

Managed Airflow-Umgebung (Gen 2) erstellen

Beim Erstellen der Umgebung weisen Sie dem Composer-Dienst-Agent-Konto die Rolle Dienst-Agent-Erweiterung für die Cloud Composer v2 API (roles/composer.ServiceAgentV2Ext) zu. Managed Airflow verwendet dieses Konto, um Vorgänge in Ihrem Google Cloud Projekt auszuführen.

Beispiel: Planer funktioniert nicht richtig und Aufgabe schlägt aufgrund von Problemen bei der Aufgabenplanung fehl

In diesem Beispiel wird gezeigt, wie Sie Fehler im Scheduler und Latenz, die durch eine hohe Nebenläufigkeit von Tasks verursacht wird, beheben können.

Beispiel-DAG in Ihre Umgebung hochladen

Laden Sie den folgenden Beispiel-DAG in die Umgebung hoch, die Sie in den vorherigen Schritten erstellt haben. In dieser Anleitung heißt der DAG dag_10_tasks_200_seconds_1.

Dieser DAG hat 200 Aufgaben. Jede Aufgabe wartet eine Sekunde und gibt „Complete!“ aus. Der DAG wird nach dem Hochladen automatisch ausgelöst. Managed Airflow führt diesen DAG zehnmal aus und alle DAG-Ausführungen erfolgen parallel.

import time
from datetime import datetime, timedelta

from airflow import DAG
from airflow.decorators import task


tasks_amount = 200
seconds = 1
minutes = 5

with DAG(
    dag_id=f"dag_10_tasks_{tasks_amount}_sec_{seconds}",
    start_date=datetime(2023, 11, 22, 20, 0),
    end_date=datetime(2023, 11, 22, 20, 49),
    schedule=timedelta(minutes=minutes),
    catchup=True,
) as dag:

    @task
    def create_subtasks(seconds: int) -> None:
        time.sleep(seconds)

    for i in range(tasks_amount):
        create_subtasks(seconds)

Probleme mit dem Scheduler und Taskfehler diagnostizieren

Öffnen Sie nach Abschluss der DAG-Ausführungen die Airflow-Benutzeroberfläche und klicken Sie auf die dag_10_tasks_200_seconds_1-DAG. Sie sehen, dass insgesamt 10 DAG-Ausführungen erfolgreich waren und jede 200 erfolgreiche Aufgaben enthält.

Prüfen Sie die Airflow-Aufgabenlogs:

1. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

2. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung. Die Seite Umgebungsdetails wird geöffnet.

3. Rufen Sie den Tab Logs auf und wählen Sie dann Alle Logs > Airflow-Logs > Worker > Im Log-Explorer ansehen aus.

Im Histogramm der Logs werden Fehler und Warnungen rot und orange dargestellt:

Die Beispiel-DAG hat zu etwa 130 Warnungen und 60 Fehlern geführt. Klicken Sie auf eine beliebige Spalte mit gelben und roten Balken. In den Logs werden möglicherweise einige der folgenden Warnungen und Fehler angezeigt:

State of this instance has been externally set to success. Terminating
instance.

Received SIGTERM. Terminating subprocesses.

worker: Warm shutdown (MainProcess).

Diese Logs können darauf hinweisen, dass die Ressourcennutzung die Grenzwerte überschritten hat und der Worker neu gestartet wurde.

Wenn eine Airflow-Aufgabe zu lange in der Warteschlange verbleibt, markiert der Scheduler sie als fehlgeschlagen und als „up_for_retry“ (bereit für Wiederholungsversuch) und plant sie noch einmal für die Ausführung. Eine Möglichkeit, die Symptome dieser Situation zu beobachten, besteht darin, sich das Diagramm mit der Anzahl der in die Warteschlange eingereihten Aufgaben anzusehen. Wenn die Spitzen in diesem Diagramm nicht innerhalb von etwa 10 Minuten abnehmen, kommt es wahrscheinlich zu Fehlern bei Aufgaben (ohne Logs).

Überprüfen Sie die Monitoring-Informationen:

1. Rufen Sie den Tab Monitoring auf und wählen Sie Übersicht aus.

2. Sehen Sie sich das Diagramm Airflow-Tasks an.

   

   Im Diagramm für Airflow-Aufgaben ist ein Anstieg der in die Warteschlange eingereihten Aufgaben zu sehen, der länger als 10 Minuten dauert. Das kann darauf hindeuten, dass in Ihrer Umgebung nicht genügend Ressourcen vorhanden sind, um alle geplanten Aufgaben zu verarbeiten.

3. Sehen Sie sich das Diagramm Aktive Worker an:

   

   Das Diagramm Aktive Worker zeigt, dass durch den DAG-Lauf Autoscaling auf das maximal zulässige Limit von drei Workern ausgelöst wurde.

4. Diagramme zur Ressourcennutzung können auf mangelnde Kapazität in Airflow-Workern zum Ausführen von in die Warteschlange eingereihten Aufgaben hinweisen. Wählen Sie auf dem Tab Monitoring die Option Worker aus und sehen Sie sich die Diagramme Gesamte CPU-Nutzung der Worker und Gesamte Arbeitsspeichernutzung der Worker an.

   

   

   Die Diagramme zeigen, dass die Ausführung zu vieler Aufgaben gleichzeitig dazu geführt hat, dass das CPU-Limit erreicht wurde. Die Ressourcen wurden über 30 Minuten lang verwendet, was sogar länger ist als die Gesamtdauer von 200 Aufgaben in 10 DAG-Läufen, die nacheinander ausgeführt werden.

Dies sind die Anzeichen dafür, dass die Warteschlange voll ist und nicht genügend Ressourcen zur Verarbeitung aller geplanten Aufgaben vorhanden sind.

Aufgaben konsolidieren

Im aktuellen Code werden viele DAGs und Aufgaben erstellt, ohne dass genügend Ressourcen vorhanden sind, um alle Aufgaben parallel zu verarbeiten. Dadurch wird die Warteschlange gefüllt. Wenn Aufgaben zu lange in der Warteschlange verbleiben, kann es sein, dass sie neu geplant werden oder fehlschlagen. In solchen Fällen sollten Sie sich für eine kleinere Anzahl von konsolidierteren Aufgaben entscheiden.

In der folgenden Beispiel-DAG wird die Anzahl der Aufgaben im ursprünglichen Beispiel von 200 auf 20 geändert und die Wartezeit von 1 auf 10 Sekunden erhöht, um konsolidiertere Aufgaben zu simulieren, die dieselbe Menge an Arbeit verrichten.

Laden Sie den folgenden Beispiel-DAG in die von Ihnen erstellte Umgebung hoch. In dieser Anleitung heißt der DAG dag_10_tasks_20_seconds_10.

import time
from datetime import datetime, timedelta

from airflow import DAG
from airflow.decorators import task


tasks_amount = 20
seconds = 10
minutes = 5

with DAG(
    dag_id=f"dag_10_tasks_{tasks_amount}_sec_{seconds}",
    start_date=datetime(2021, 12, 22, 20, 0),
    end_date=datetime(2021, 12, 22, 20, 49),
    schedule=timedelta(minutes=minutes),
    catchup=True,
) as dag:

    @task
    def create_subtasks(seconds: int) -> None:
        time.sleep(seconds)

    for i in range(tasks_amount):
        create_subtasks(seconds)

Bewerten Sie die Auswirkungen von konsolidierten Aufgaben auf die Planungsprozesse:

1. Warten Sie, bis die DAG-Ausführungen abgeschlossen sind.

2. Klicken Sie in der Airflow-Benutzeroberfläche auf der Seite DAGs auf den dag_10_tasks_20_seconds_10-DAG. Sie sehen 10 DAG-Ausführungen mit jeweils 20 erfolgreichen Aufgaben.

3. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

4. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung. Die Seite Umgebungsdetails wird geöffnet.

5. Rufen Sie den Tab Logs auf und wählen Sie dann Alle Logs > Airflow-Logs > Worker > Im Log-Explorer ansehen aus.

   Das zweite Beispiel mit konsolidierteren Aufgaben führte zu etwa 10 Warnungen und 7 Fehlern. Im Histogramm können Sie die Anzahl der Fehler und Warnungen im ersten Beispiel (frühere Werte) und im zweiten Beispiel (spätere Werte) vergleichen.

   

   Wenn Sie das erste Beispiel mit dem konsolidierten Beispiel vergleichen, sehen Sie, dass im zweiten Beispiel deutlich weniger Fehler und Warnungen vorhanden sind. Dieselben Fehler im Zusammenhang mit dem Warm Shutdown werden jedoch aufgrund von Ressourcenüberlastung weiterhin in den Logs angezeigt.

6. Wählen Sie auf dem Tab Monitoring die Option Worker aus und sehen Sie sich die Diagramme an.

   Wenn Sie das Diagramm Airflow-Aufgaben für das erste Beispiel (frühere Werte) mit dem Diagramm für das zweite Beispiel mit konsolidierteren Aufgaben vergleichen, sehen Sie, dass der Anstieg der in die Warteschlange eingereihten Aufgaben bei konsolidierteren Aufgaben kürzer war. Es dauerte jedoch fast 10 Minuten, was immer noch suboptimal ist.

   

   Im Diagramm „Aktive Worker“ sehen Sie, dass im ersten Beispiel (auf der linken Seite des Diagramms) Ressourcen über einen viel längeren Zeitraum verwendet wurden als im zweiten, obwohl in beiden Beispielen die gleiche Menge an Arbeit simuliert wird.

   

   Sehen Sie sich die Diagramme zum Ressourcenverbrauch der Worker an. Obwohl der Unterschied zwischen den im Beispiel mit mehr konsolidierten Aufgaben verwendeten Ressourcen und dem ursprünglichen Beispiel recht groß ist, steigt die CPU-Auslastung immer noch auf 70% des Limits.

   

   

Aufgaben gleichmäßiger über die Zeit verteilen

Zu viele gleichzeitige Aufgaben führen dazu, dass die Warteschlange voll ist. Das kann dazu führen, dass Aufgaben in der Warteschlange hängen bleiben oder neu geplant werden. In den vorherigen Schritten haben Sie die Anzahl der Aufgaben durch Konsolidierung reduziert. Die Ausgabelogs und das Monitoring haben jedoch gezeigt, dass die Anzahl der gleichzeitig ausgeführten Aufgaben immer noch suboptimal ist.

Sie können die Anzahl der gleichzeitigen Ausführungen von Aufgaben steuern, indem Sie einen Zeitplan implementieren oder Grenzwerte für die Anzahl der Aufgaben festlegen, die gleichzeitig ausgeführt werden können.

In dieser Anleitung verteilen Sie Aufgaben gleichmäßiger über die Zeit, indem Sie Parameter auf DAG-Ebene in den dag_10_tasks_20_seconds_10-DAG einfügen:

1. Fügen Sie dem DAG-Kontextmanager das Argument max_active_runs=1 hinzu. Mit diesem Argument wird ein Limit von nur einer Instanz eines DAG-Laufs zu einem bestimmten Zeitpunkt festgelegt.

2. Fügen Sie dem DAG-Kontextmanager das Argument max_active_tasks=5 hinzu. Mit diesem Argument wird die maximale Anzahl von Aufgabeninstanzen gesteuert, die in jedem DAG gleichzeitig ausgeführt werden können.

Laden Sie den folgenden Beispiel-DAG in die von Ihnen erstellte Umgebung hoch. In dieser Anleitung heißt der DAG dag_10_tasks_20_seconds_10_scheduled.py.

import time
from datetime import datetime, timedelta

from airflow import DAG
from airflow.decorators import task


tasks_amount = 20
seconds = 10
minutes = 5
active_runs = 1
active_tasks = 5


with DAG(
    dag_id=f"dag_10_tasks_{tasks_amount}_sec_{seconds}_runs_{active_runs}_tasks_{active_tasks}",
    start_date=datetime(2021, 12, 22, 20, 0),
    end_date=datetime(2021, 12, 22, 20, 49),
    schedule=timedelta(minutes=minutes),
    max_active_runs=active_runs,
    max_active_tasks=active_tasks,
    catchup=True,
) as dag:

    @task
    def create_subtasks(seconds: int) -> None:
        time.sleep(seconds)

    for i in range(tasks_amount):
        create_subtasks(seconds)

Auswirkungen der Verteilung von Aufgaben im Zeitverlauf auf die Planungsprozesse bewerten:

1. Warten Sie, bis die DAG-Ausführungen abgeschlossen sind.

2. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

3. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung. Die Seite Umgebungsdetails wird geöffnet.

4. Rufen Sie den Tab Logs auf und wählen Sie dann Alle Logs > Airflow-Logs > Worker > Im Log-Explorer ansehen aus.

5. Im Histogramm sehen Sie, dass für den dritten DAG mit einer begrenzten Anzahl aktiver Aufgaben und Ausführungen keine Warnungen oder Fehler generiert wurden und die Verteilung der Logs im Vergleich zu den vorherigen Werten gleichmäßiger aussieht.

   

Die Aufgaben im Beispiel dag_10_tasks_20_seconds_10_scheduled mit einer begrenzten Anzahl aktiver Aufgaben und Ausführungen haben keinen Ressourcenengpass verursacht, da die Aufgaben gleichmäßig in die Warteschlange gestellt wurden.

Nachdem Sie die beschriebenen Schritte ausgeführt haben, haben Sie die Ressourcennutzung optimiert, indem Sie kleine Aufgaben zusammengefasst und gleichmäßiger über die Zeit verteilt haben.

Umgebungskonfigurationen optimieren

Sie können Ihre Umgebungskonfigurationen anpassen, um sicherzustellen, dass in Airflow-Workern immer Kapazität zum Ausführen von in die Warteschlange gestellten Aufgaben vorhanden ist.

Anzahl der Worker und Worker-Concurrency

Sie können die maximale Anzahl von Workern anpassen, damit Managed Airflow Ihre Umgebung automatisch innerhalb der festgelegten Grenzwerte skaliert.

Der Parameter [celery]worker_concurrency definiert die maximale Anzahl von Aufgaben, die ein einzelner Worker aus der Aufgabenwarteschlange abrufen kann. Wenn Sie diesen Parameter ändern, wird die Anzahl der Aufgaben angepasst, die ein einzelner Worker gleichzeitig ausführen kann. Sie können diese Airflow-Konfigurationsoption überschreiben. Standardmäßig wird die Worker-Nebenläufigkeit basierend auf der Anzahl der leichten, gleichzeitigen Aufgabeninstanzen festgelegt, die ein Worker verarbeiten kann. Das bedeutet, dass der Wert von den Ressourcenlimits für Worker abhängt. Der Wert für die Worker-Concurrency hängt nicht von der Anzahl der Worker in Ihrer Umgebung ab.

Die Anzahl der Worker und die Worker-Nebenläufigkeit wirken zusammen. Die Leistung Ihrer Umgebung hängt stark von beiden Parametern ab. Die folgenden Überlegungen können Ihnen bei der Auswahl der richtigen Kombination helfen:

• Mehrere Schnellaufgaben werden parallel ausgeführt. Sie können die Worker-Nebenläufigkeit erhöhen, wenn Aufgaben in der Warteschlange warten und Ihre Worker gleichzeitig nur einen geringen Prozentsatz ihrer CPUs und ihres Arbeitsspeichers verwenden. Unter bestimmten Umständen füllt sich die Warteschlange jedoch möglicherweise nie, sodass das Autoscaling nie ausgelöst wird. Wenn kleine Aufgaben abgeschlossen sind, bevor die neuen Worker bereit sind, kann ein vorhandener Worker die verbleibenden Aufgaben übernehmen. In diesem Fall gibt es keine Aufgaben für neu erstellte Worker.

  In diesen Fällen wird empfohlen, die Mindestanzahl von Workern und die Worker-Nebenläufigkeit zu erhöhen, um eine zu schnelle Skalierung zu vermeiden.

• Mehrere lange Aufgaben werden parallel ausgeführt. Die hohe Worker-Nebenläufigkeit verhindert, dass das System die Anzahl der Worker skaliert. Wenn mehrere Aufgaben ressourcenintensiv sind und lange dauern, kann eine hohe Worker-Parallelität dazu führen, dass die Warteschlange nie gefüllt wird und alle Aufgaben von nur einem Worker übernommen werden, was zu Leistungsproblemen führt. In diesen Fällen empfiehlt es sich, die maximale Anzahl von Workern zu erhöhen und die Worker-Nebenläufigkeit zu verringern.

Die Bedeutung der Parallelität

Airflow-Planer steuern die Planung von DAG-Ausführungen und einzelnen Tasks aus DAGs. Mit der Airflow-Konfigurationsoption [core]parallelism wird gesteuert, wie viele Aufgaben der Airflow-Planer in die Warteschlange des Executors stellen kann, nachdem alle Abhängigkeiten für diese Aufgaben erfüllt sind.

Die Parallelität ist ein Schutzmechanismus von Airflow, der bestimmt, wie viele Aufgaben gleichzeitig pro Planer ausgeführt werden können, unabhängig von der Anzahl der Worker. Der Parallelitätswert multipliziert mit der Anzahl der Scheduler in Ihrem Cluster ist die maximale Anzahl von Task-Instanzen, die in Ihrer Umgebung in die Warteschlange gestellt werden können.

Normalerweise wird [core]parallelism als Produkt aus einer maximalen Anzahl von Workern und [celery]worker_concurrency festgelegt. Sie wird auch vom Pool beeinflusst. Sie können diese Airflow-Konfigurationsoption überschreiben. Weitere Informationen zum Anpassen von Airflow-Konfigurationen in Bezug auf die Skalierung finden Sie unter Airflow-Konfiguration skalieren.

Optimale Umgebungskonfigurationen finden

Die empfohlene Methode zur Behebung von Planungsproblemen besteht darin, kleine Aufgaben in größere Aufgaben zusammenzufassen und Aufgaben gleichmäßiger über die Zeit zu verteilen. Neben der Optimierung des DAG-Codes können Sie auch die Umgebungskonfigurationen optimieren, um genügend Kapazität für die gleichzeitige Ausführung mehrerer Aufgaben zu haben.

Angenommen, Sie konsolidieren Aufgaben in Ihrem DAG so weit wie möglich, aber die Begrenzung aktiver Aufgaben, um sie gleichmäßiger über die Zeit zu verteilen, ist für Ihren speziellen Anwendungsfall keine bevorzugte Lösung.

Sie können die Parameter für Parallelität, Anzahl der Worker und Worker-Nebenläufigkeit anpassen, um den dag_10_tasks_20_seconds_10-DAG auszuführen, ohne aktive Aufgaben einzuschränken. In diesem Beispiel wird der DAG zehnmal ausgeführt und jeder Lauf enthält 20 kleine Aufgaben. So führen Sie alle gleichzeitig aus:

• Sie benötigen eine größere Umgebungsgröße, da sie die Leistungsparameter der verwalteten Managed Airflow-Infrastruktur Ihrer Umgebung steuert.

• Airflow-Worker müssen 20 Aufgaben gleichzeitig ausführen können. Daher müssen Sie die Worker-Concurrency auf 20 festlegen.

• Die Worker benötigen ausreichend CPU und Arbeitsspeicher, um alle Aufgaben zu bewältigen. Die Worker-Nebenläufigkeit wird durch die Worker-CPU und den Worker-Arbeitsspeicher beeinflusst. Daher benötigen Sie mindestens worker_concurrency / 12 CPUs und least worker_concurrency / 8 Arbeitsspeicher.

• Sie müssen die Parallelität erhöhen, um der höheren Worker-Nebenläufigkeit gerecht zu werden. Damit Mitarbeiter 20 Aufgaben aus der Warteschlange übernehmen können, muss der Scheduler diese 20 Aufgaben zuerst planen.

Passen Sie die Konfigurationen Ihrer Umgebung so an:

1. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

2. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung. Die Seite Umgebungsdetails wird geöffnet.

3. Rufen Sie den Tab Umgebungskonfiguration auf.

4. Suchen Sie nach der Konfiguration Ressourcen > Arbeitslasten und klicken Sie auf Bearbeiten.

5. Geben Sie im Abschnitt Worker im Feld Arbeitsspeicher das neue Arbeitsspeicherlimit für Airflow-Worker an. Verwenden Sie in dieser Anleitung 4 GB.

6. Geben Sie im Feld CPU das neue CPU-Limit für Airflow-Worker an. Verwenden Sie in dieser Anleitung 2 vCPUs.

7. Speichern Sie die Änderungen und warten Sie einige Minuten, bis die Airflow-Worker neu gestartet wurden.

Überschreiben Sie als Nächstes die Airflow-Konfigurationsoptionen für Parallelität und Worker-Concurrency:

1. Rufen Sie den Tab Airflow-Konfigurationsüberschreibungen auf.

2. Klicken Sie auf Bearbeiten und dann auf Airflow-Konfigurationsüberschreibung hinzufügen.

3. Parallelitätskonfiguration überschreiben:

   Bereich	Schlüssel	Wert
   core	parallelism	20

4. Klicken Sie auf Airflow-Konfigurationsüberschreibung hinzufügen und überschreiben Sie die Worker-Concurrency-Konfiguration:

   Bereich	Schlüssel	Wert
   celery	worker_concurrency	20

5. Klicken Sie auf Speichern und warten Sie, bis die Konfiguration der Umgebung aktualisiert wurde.

Lösen Sie den DAG mit den angepassten Konfigurationen noch einmal aus:

1. Rufen Sie in der Airflow-UI die Seite DAGs auf.

2. Suchen Sie den dag_10_tasks_20_seconds_10-DAG und löschen Sie ihn.

   Nachdem der DAG gelöscht wurde, prüft Airflow den DAGs-Ordner im Bucket Ihrer Umgebung und führt den DAG automatisch noch einmal aus.

Sehen Sie sich nach Abschluss der DAG-Ausführungen noch einmal das Protokollhistogramm an. Im Diagramm sehen Sie, dass beim dag_10_tasks_20_seconds_10-Beispiel mit konsolidierteren Aufgaben keine Fehler und Warnungen generiert wurden, als es mit der angepassten Umgebungskonfiguration ausgeführt wurde. Vergleichen Sie die Ergebnisse mit den vorherigen Daten im Diagramm, bei denen für dasselbe Beispiel Fehler und Warnungen generiert wurden, als es mit der Standardumgebungskonfiguration ausgeführt wurde.

Umgebungskonfigurationen und Airflow-Konfigurationen spielen eine entscheidende Rolle bei der Planung von Aufgaben. Es ist jedoch nicht möglich, die Konfigurationen über bestimmte Grenzwerte hinaus zu erhöhen.

Wir empfehlen, den DAG-Code zu optimieren, Aufgaben zusammenzufassen und die Planung zu verwenden, um Leistung und Effizienz zu optimieren.

Beispiel: Fehler beim Parsen von DAGs und Latenz aufgrund von komplexem DAG-Code

In diesem Beispiel untersuchen Sie die Parsing-Latenz eines Beispiel-DAG, der eine große Anzahl von Airflow-Variablen imitiert.

Neue Airflow-Variable erstellen

Bevor Sie den Beispielcode hochladen, erstellen Sie eine neue Airflow-Variable.

1. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

2. Klicken Sie in der Spalte Airflow-Webserver auf den Link Airflow für Ihre Umgebung.

3. Klicken Sie auf Verwaltung > Variablen > Neuen Datensatz hinzufügen.

4. Legen Sie die folgenden Werte fest:

   • Schlüssel: example_var
   • val: test_airflow_variable

Beispiel-DAG in Ihre Umgebung hochladen

Laden Sie den folgenden Beispiel-DAG in die Umgebung hoch, die Sie in den vorherigen Schritten erstellt haben. In dieser Anleitung heißt der DAG dag_for_loop_airflow_variable.

Dieser DAG enthält eine For-Schleife,die 1.000 Mal ausgeführt wird und eine große Anzahl von Airflow-Variablen imitiert. Bei jeder Iteration wird die Variable example_var gelesen und eine Aufgabe generiert. Jede Aufgabe enthält einen Befehl, der den Wert der Variablen ausgibt.

from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.models import Variable

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 2, 17),
    'retries': 0
}

dag = DAG(
    'dag_for_loop_airflow_variable',
    default_args=default_args,
    catchup=False,
    schedule="@daily"
)

for i in range(1000):
    a = Variable.get('example_var', 'N/A')
    task = BashOperator(
        task_id=f'task_{i}',
        bash_command='echo variable foo=${foo_env}',
        dag=dag,
        env={'foo_env': a}
    )

Parsing-Probleme diagnostizieren

Die DAG-Parsingzeit ist die Zeit, die der Airflow-Planer benötigt, um eine DAG-Datei zu lesen und zu parsen. Bevor der Airflow-Planer eine Aufgabe aus einem DAG planen kann, muss er die DAG-Datei parsen, um die Struktur des DAG und die definierten Aufgaben zu ermitteln.

Wenn das Parsen eines DAG lange dauert, wird die Kapazität des Schedulers beansprucht und die Leistung von DAG-Ausführungen kann beeinträchtigt werden.

So überwachen Sie die DAG-Parsing-Zeit:

1. Führen Sie den dags report Airflow-Befehlszeilenbefehl in der gcloud CLI aus, um die Analysezeit für alle Ihre DAGs zu sehen:

   gcloud composer environments run ENVIRONMENT_NAME \
       --location LOCATION \
       dags report
   

   Ersetzen Sie Folgendes:

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

2. Suchen Sie in der Ausgabe des Befehls nach dem Dauerwert für den dag_for_loop_airflow_variables-DAG. Ein großer Wert kann darauf hindeuten, dass dieser DAG nicht optimal implementiert ist. Wenn Sie mehrere DAGs haben, können Sie in der Ausgabetabelle sehen, welche DAGs eine lange Parsing-Zeit haben.

   Beispiel:

   file                | duration       | dag_num | task_num | dags
   ====================+================+=========+==========+=====================
   /dag_for_loop_airfl | 0:00:14.773594 | 1       | 1000     | dag_for_loop_airflow
   ow_variable.py      |                |         |          | _variable
   /airflow_monitoring | 0:00:00.003035 | 1       | 1        | airflow_monitoring
   .py
   
   

3. DAG-Parse-Zeiten in der Google Cloud -Console ansehen:

   1. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

4. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung. Die Seite Umgebungsdetails wird geöffnet.

5. Rufen Sie den Tab Logs auf und wählen Sie dann Alle Logs > DAG-Prozessormanager aus.

6. Sehen Sie sich die dag-processor-manager-Logs an und identifizieren Sie mögliche Probleme.

   

Wenn die Gesamtzeit für die DAG-Analyse etwa 10 Sekunden überschreitet, sind Ihre Scheduler möglicherweise mit der DAG-Analyse überlastet und können DAGs nicht effektiv ausführen.

DAG-Code optimieren

Es wird empfohlen, unnötigen Python-Code der obersten Ebene in Ihren DAGs zu vermeiden. DAGs mit vielen Importen, Variablen und Funktionen außerhalb des DAG führen zu längeren Parsing-Zeiten für den Airflow-Planer. Dadurch werden Leistung und Skalierbarkeit von Managed Airflow und Airflow verringert. Wenn zu viele Airflow-Variablen gelesen werden, führt das zu einer langen Parsing-Zeit und einer hohen Datenbanklast. Wenn sich dieser Code in einer DAG-Datei befindet, werden diese Funktionen bei jedem Scheduler-Heartbeat ausgeführt, was langsam sein kann.

Mit den Vorlagenfeldern von Airflow können Sie Werte aus Airflow-Variablen und Jinja-Vorlagen in Ihre DAGs einfügen. So wird verhindert, dass während der Scheduler-Heartbeats unnötige Funktionen ausgeführt werden.

Um das DAG-Beispiel besser zu implementieren, sollten Sie Airflow-Variablen nicht im Python-Code der obersten Ebene von DAGs verwenden. Übergeben Sie stattdessen Airflow-Variablen über eine Jinja-Vorlage an vorhandene Operatoren. Dadurch wird das Lesen des Werts bis zur Ausführung des Tasks verzögert.

Laden Sie die neue Version des Beispiel-DAG in Ihre Umgebung hoch. In dieser Anleitung heißt der DAG dag_for_loop_airflow_variable_optimized.

from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 2, 17),
    'retries': 0
}

dag = DAG(
    'dag_for_loop_airflow_variable_optimized',
    default_args=default_args,
    catchup=False,
    schedule='@daily'
)

for i in range(1000):
    task = BashOperator(
        task_id=f'bash_use_variable_good_{i}',
        bash_command='echo variable foo=${foo_env}',
        dag=dag,
        env={'foo_env': '{{ var.value.get("example_var") }}'},
    )

Prüfen Sie die neue DAG-Parsing-Zeit:

1. Warten Sie, bis der DAG-Lauf abgeschlossen ist.

2. Führen Sie den Befehl dags report noch einmal aus, um die Parsing-Zeit für alle Ihre DAGs zu sehen:

   file                | duration       | dag_num | task_num | dags
   ====================+================+=========+==========+=====================
   /dag_for_loop_airfl | 0:00:37.000369 | 1       | 1000     | dag_for_loop_airflow
   ow_variable.py      |                |         |          | _variable
   /dag_for_loop_airfl | 0:00:01.109457 | 1       | 1000     | dag_for_loop_airflow
   ow_variable_optimiz |                |         |          | _variable_optimized
   ed.py               |                |         |          |
   /airflow_monitoring | 0:00:00.040510 | 1       | 1        | airflow_monitoring
   .py                 |                |         |          |
   

3. Prüfen Sie die dag-processor-manager-Logs noch einmal und analysieren Sie die Parsing-Dauer.

   

Durch das Ersetzen der Umgebungsvariablen durch Airflow-Vorlagen haben Sie den DAG-Code vereinfacht und die Parsing-Latenz um das Zehnfache reduziert.

Airflow-Umgebungskonfigurationen optimieren

Der Airflow-Planer versucht ständig, neue Tasks auszulösen, und parst alle DAGs in Ihrem Umgebungs-Bucket. Wenn das Parsen Ihrer DAGs lange dauert und der Scheduler viele Ressourcen verbraucht, können Sie die Airflow-Scheduler-Konfigurationen optimieren, damit der Scheduler Ressourcen effizienter nutzt.

In diesem Tutorial dauert das Parsen der DAG-Dateien sehr lange und die Parsing-Zyklen beginnen sich zu überschneiden, wodurch die Kapazität des Planers erschöpft wird. In unserem Beispiel dauert das Parsen des ersten DAG mehr als 5 Sekunden. Daher konfigurieren Sie den Scheduler so, dass er seltener ausgeführt wird, um Ressourcen effizienter zu nutzen. Sie überschreiben die Airflow-Konfigurationsoption scheduler_heartbeat_sec. Mit dieser Konfiguration wird festgelegt, wie oft der Scheduler ausgeführt werden soll (in Sekunden). Standardmäßig ist der Wert auf 5 Sekunden festgelegt. Sie können diese Airflow-Konfigurationsoption überschreiben.

Überschreiben Sie die Airflow-Konfigurationsoption scheduler_heartbeat_sec:

1. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

2. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung. Die Seite Umgebungsdetails wird geöffnet.

3. Rufen Sie den Tab Airflow-Konfigurationsüberschreibungen auf.

4. Klicken Sie auf Bearbeiten und dann auf Airflow-Konfigurationsüberschreibung hinzufügen.

5. Überschreiben Sie die Airflow-Konfigurationsoption:

   Bereich	Schlüssel	Wert
   scheduler	scheduler_heartbeat_sec	10

6. Klicken Sie auf Speichern und warten Sie, bis die Konfiguration der Umgebung aktualisiert wurde.

Planermesswerte prüfen:

1. Rufen Sie den Tab Monitoring auf und wählen Sie Schedulers aus.

2. Klicken Sie im Diagramm Scheduler-Heartbeat auf die Schaltfläche Weitere Optionen (drei Punkte) und dann auf Im Messwert-Explorer ansehen.

Im Diagramm sehen Sie, dass der Scheduler nach der Änderung der Standardkonfiguration von 5 auf 10 Sekunden zweimal seltener ausgeführt wird. Wenn Sie die Häufigkeit von Heartbeats reduzieren, wird der Scheduler nicht gestartet, während der vorherige Parsing-Zyklus noch läuft und die Ressourcenkapazität des Schedulers nicht erschöpft ist.

Scheduler mehr Ressourcen zuweisen

In Managed Airflow (Gen 2) können Sie dem Scheduler mehr CPU- und Arbeitsspeicherressourcen zuweisen. So können Sie die Leistung Ihres Planers steigern und die Parsing-Zeit für Ihren DAG verkürzen.

Zusätzliche CPU und Arbeitsspeicher für den Planer zuweisen:

1. Rufen Sie in der Google Cloud Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

2. Klicken Sie in der Liste der Umgebungen auf den Namen Ihrer Umgebung. Die Seite Umgebungsdetails wird geöffnet.

3. Rufen Sie den Tab Umgebungskonfiguration auf.

4. Suchen Sie nach der Konfiguration Ressourcen > Arbeitslasten und klicken Sie auf Bearbeiten.

5. Geben Sie im Abschnitt Scheduler im Feld Arbeitsspeicher das neue Arbeitsspeicherlimit an. Verwenden Sie in dieser Anleitung 4 GB.

6. Geben Sie im Feld CPU das neue CPU-Limit an. Verwenden Sie in dieser Anleitung 2 vCPUs.

7. Speichern Sie die Änderungen und warten Sie einige Minuten, bis die Airflow-Scheduler neu gestartet werden.

8. Rufen Sie den Tab Logs auf und wählen Sie dann Alle Logs > DAG-Prozessormanager aus.

9. Überprüfen Sie die dag-processor-manager-Logs und vergleichen Sie die Analysedauer für die Beispiel-DAGs:

   

Durch die Zuweisung von mehr Ressourcen an den Planer haben Sie die Kapazität des Planers erhöht und die Parsing-Latenz im Vergleich zu den Standardumgebungskonfigurationen deutlich reduziert. Mit mehr Ressourcen kann der Planer die DAGs schneller parsen. Die Kosten für Managed Airflow-Ressourcen steigen jedoch auch. Außerdem ist es nicht möglich, die Ressourcen über ein bestimmtes Limit hinaus zu erhöhen.

Wir empfehlen, Ressourcen erst zuzuweisen, nachdem die möglichen DAG-Code- und Airflow-Konfigurationsoptimierungen implementiert wurden.

Bereinigen

Damit Ihrem Google Cloud -Konto die in dieser Anleitung verwendeten Ressourcen nicht in Rechnung gestellt werden, können Sie entweder das Projekt löschen, das die Ressourcen enthält, oder das Projekt beibehalten und die einzelnen Ressourcen löschen.

Projekt löschen

1. Wechseln Sie in der Google Cloud -Console zur Seite Ressourcen verwalten.

   Zur Seite „Ressourcen verwalten“

2. Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten, und klicken Sie dann auf Löschen.
3. Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf Shut down (Beenden), um das Projekt zu löschen.

Einzelne Ressourcen löschen

Wenn Sie mehrere Anleitungen und Kurzanleitungen durcharbeiten möchten, können Sie die Überschreitung von Projektkontingenten verhindern, indem Sie Projekte wiederverwenden.

Managed Airflow-Umgebung löschen Bei diesem Vorgang löschen Sie auch den Bucket der Umgebung.

Nächste Schritte

• Umgebungsleistung und ‑kosten optimieren
• Umgebungen skalieren
• Fehlerbehebung bei DAGs