Airflow-Planerprobleme beheben

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

Diese Seite enthält Schritte zur Fehlerbehebung und Informationen zu häufigen Problemen mit Airflow-Planern und DAG-Prozessoren.

Problemursache identifizieren

Ermitteln Sie zu Beginn der Fehlerbehebung, ob das Problem auftritt:

  • Zur DAG-Parsing-Zeit, während der DAG von einem Airflow-DAG-Prozessor geparst wird
  • Zur Ausführungszeit, während der DAG von einem Airflow-Planer verarbeitet wird

Weitere Informationen zur Parsing- und zur Ausführungszeit finden Sie unter Unterschied zwischen der DAG-Parsing-Zeit und der DAG-Ausführungszeit.

Probleme bei der DAG-Verarbeitung prüfen

  1. Prüfen Sie die Logs des DAG-Prozessors.
  2. Prüfen Sie die DAG-Parsing-Zeiten.

Laufende und in der Warteschlange befindliche Aufgaben überwachen

So prüfen Sie, ob Aufgaben in einer Warteschlange hängen:

  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 Monitoring auf.

  4. Prüfen Sie im Tab Monitoring das Diagramm Airflow-Aufgaben im Abschnitt DAG-Ausführungen und identifizieren Sie mögliche Probleme. Airflow-Aufgaben sind Aufgaben, die in Airflow in der Warteschlange stehen. Sie können entweder in die Warteschlange des Celery- oder des Kubernetes-Executor-Brokers gestellt werden. Aufgaben in der Celery-Warteschlange sind Aufgaben instanzen, die in die Celery-Broker-Warteschlange gestellt wurden.

Probleme zur DAG-Parsing-Zeit beheben

In folgenden Abschnitten werden Symptome und mögliche Lösungen für einige während der DAG-Parsing-Zeit häufig auftretende Probleme beschrieben.

DAG-Parsing und -Planung in Managed Airflow (Legacy Gen 1) und Airflow 1

Die Effizienz des DAG-Parsing wurde in Airflow 2 erheblich verbessert. Wenn Leistungsprobleme im Zusammenhang mit DAG-Parsing und -Planung auftreten, sollten Sie zu Airflow 2 migrieren.

In Managed Airflow (Legacy Gen 1) wird der Planer auf Clusterknoten zusammen mit anderen Managed Airflow-Komponenten ausgeführt. Daher kann die Last einzelner Clusterknoten im Vergleich zu anderen Knoten höher oder niedriger sein. Die Leistung des Planers (DAG-Parsing und -Planung) kann je nach Knoten variieren, auf dem der Planer ausgeführt wird. Außerdem kann sich der Knoten, auf dem der Planer ausgeführt wird, aufgrund von Upgrade- oder Wartungsvorgängen ändern. Diese Einschränkung wurde in Managed Airflow (Gen 2) behoben. Dort können Sie dem Planer CPU- und Arbeitsspeicherressourcen zuweisen und die Leistung des Planers hängt nicht von der Last der Clusterknoten ab.

Anzahl und Zeitverteilung der Aufgaben

Bei der gleichzeitigen Planung einer großen Anzahl von DAGs oder Aufgaben kann es zu Problemen mit Airflow kommen. So vermeiden Sie Probleme bei der Planung:

  • Passen Sie Ihre DAGs so an, dass sie eine kleinere Anzahl konsolidierter Aufgaben verwenden.
  • Passen Sie die Zeitplanintervalle Ihrer DAGs an, um DAG-Ausführungen gleichmäßiger über die Zeit zu verteilen.

Airflow-Konfiguration skalieren

Airflow bietet Konfigurationsoptionen, die steuern, wie viele Aufgaben und DAGs es gleichzeitig ausführen kann. Um diese Konfigurationsoptionen festzulegen, überschreiben Sie deren Werte für Ihre Umgebung. Sie können einige dieser Werte auch auf DAG- oder Aufgabenebene festlegen.

  • Worker-Nebenläufigkeit

    Der Parameter [celery]worker_concurrency steuert die maximale Anzahl von Aufgaben, die ein Airflow-Worker gleichzeitig ausführen kann. Wenn Sie den Wert dieses Parameters mit der Anzahl der Airflow-Worker in Ihrer Managed Airflow-Umgebung multiplizieren, erhalten Sie die maximale Anzahl von Aufgaben, die zu einem bestimmten Zeitpunkt in Ihrer Umgebung ausgeführt werden können. Diese Zahl ist durch die Airflow-Konfigurationsoption [core]parallelism begrenzt, die unten weiter beschrieben wird.

  • Maximale Anzahl aktiver DAG-Ausführungen

    Die Airflow-Konfigurationsoption [core]max_active_runs_per_dag steuert die maximale Anzahl aktiver DAG-Ausführungen pro DAG. Der Planer erstellt keine weiteren DAG-Ausführungen, wenn das Limit erreicht ist.

    Ist dieser Parameter falsch eingestellt, kann ein Problem auftreten, bei dem der Planer die DAG-Ausführung drosselt, da er zu einer bestimmten Zeit keine DAG-Ausführungsinstanzen mehr erstellen kann.

    Sie können diesen Wert auch auf DAG-Ebene mit dem Parameter max_active_runs festlegen.

  • Maximale Anzahl aktiver Aufgaben pro DAG

    Die Airflow-Konfigurationsoption [core]max_active_tasks_per_dag steuert die maximale Anzahl an Aufgabeninstanzen, die pro DAG gleichzeitig ausgeführt werden können.

    Ist dieser Parameter falsch festgelegt, kann ein Problem auftreten, bei dem die Ausführung einer einzelnen DAG-Instanz langsam läuft, da nur eine begrenzte Anzahl an DAG-Aufgaben zu einer bestimmten Zeit ausgeführt werden können. In diesem Fall können Sie den Wert dieser Konfigurationsoption erhöhen.

    Sie können diesen Wert auch auf DAG-Ebene mit dem Parameter max_active_tasks festlegen.

    Mit den Parametern max_active_tis_per_dag und max_active_tis_per_dagrun auf Aufgabenebene können Sie steuern, wie viele Instanzen mit einer bestimmten Aufgaben-ID pro DAG und pro DAG-Ausführung ausgeführt werden dürfen.

  • Parallelität und Poolgröße

    Die Airflow-Konfigurationsoption [core]parallelism steuert, wie viele Aufgaben der Airflow-Planer in die Warteschlange des Executors stellen kann, nachdem alle Abhängigkeiten für diese Aufgaben erfüllt wurden.

    Dies ist ein globaler Parameter für die gesamte Airflow-Einrichtung.

    Aufgaben werden in die Warteschlange gestellt und in einem Pool ausgeführt. Managed Airflow-Umgebungen verwenden nur einen Pool. Die Größe dieses Pools steuert, wie viele Aufgaben der Planer in einem bestimmten Moment zur Ausführung in die Warteschlange stellen kann. Wenn die Poolgröße zu klein ist, kann der Planer Aufgaben nicht für die Ausführung in die Warteschlange stellen, auch wenn Grenzwerte, die durch die Konfigurationsoption [core]parallelism und die Konfigurationsoption [celery]worker_concurrency multipliziert mit der Anzahl der Airflow-Worker definiert sind, noch nicht erreicht sind.

    Sie können die Poolgröße in der Airflow-UI konfigurieren (Administrator > Pools). Passen Sie die Poolgröße an das Maß an Parallelität an, das Sie in Ihrer Umgebung erwarten.

    Normalerweise wird [core]parallelism als Produkt aus der maximalen Anzahl von Workern und [celery]worker_concurrency festgelegt.

Fehlerbehebung bei laufenden und in der Warteschlange befindlichen Aufgaben

In folgenden Abschnitten werden Symptome und mögliche Lösungen für einige bei laufenden und in der Warteschlange befindlichen Aufgaben häufig auftretenden Problemen beschrieben.

DAG-Ausführungen werden nicht ausgeführt

Symptom:

Wenn ein Zeitplandatum für einen DAG dynamisch festgelegt wird, kann dies zu verschiedenen unerwarteten Nebenwirkungen führen. Beispiel:

  • Eine DAG-Ausführung liegt immer in der Zukunft und der DAG wird nie ausgeführt.

  • Vergangene DAG-Ausführungen werden als ausgeführt und erfolgreich markiert, obwohl sie nicht ausgeführt wurden.

Weitere Informationen finden Sie in der Apache Airflow-Dokumentation.

Mögliche Lösungen:

  • Folgen Sie den Empfehlungen in der Apache Airflow-Dokumentation.

  • Legen Sie ein statisches start_date für DAGs fest. Optional können Sie catchup=False verwenden, um die Ausführung des DAG für vergangene Termine zu deaktivieren.

  • Verwenden Sie datetime.now() oder days_ago(<number of days>) nur, wenn Sie sich der Nebenwirkungen dieses Ansatzes bewusst sind.

TimeTable-Funktion des Airflow-Planers verwenden

Zeitpläne sind ab Airflow 2.2 verfügbar.

Sie können einen Zeitplan für einen DAG mit einer der folgenden Methoden definieren:

Sie können auch integrierte Zeitpläne verwenden.

Eingeschränkte Clusterressourcen

Wenn der GKE-Cluster Ihrer Umgebung zu klein ist, um alle DAGs und Aufgaben zu verarbeiten, können Leistungsprobleme auftreten. Versuchen Sie in diesem Fall eine der folgenden Lösungen:

  • Erstellen Sie eine neue Umgebung mit einem Maschinentyp, der mehr Leistung bietet, und migrieren Sie Ihre DAGs dorthin.
  • Erstellen Sie weitere Managed Airflow-Umgebungen und teilen Sie die DAGs zwischen ihnen auf.
  • Ändern Sie den Maschinentyp für GKE-Knoten wie unter Maschinentyp für GKE-Knoten aktualisieren beschrieben. Da dieses Verfahren fehleranfällig ist, ist es die am wenigsten empfohlene Option.
  • Führen Sie ein Upgrade des Maschinentyps der Cloud SQL-Instanz durch, auf der die Airflow Datenbank in Ihrer Umgebung ausgeführt wird, z. B. mit den gcloud composer environments update Befehlen. Eine geringe Leistung der Airflow-Datenbank kann der Grund sein, warum der Planer langsam arbeitet.

Aufgabenplanung während Wartungsfenstern vermeiden

Sie können Wartungsfenster für Ihre Umgebung definieren, damit die Wartung der Umgebung außerhalb der Zeiten erfolgt, in denen Sie Ihre DAGs ausführen. Sie können Ihre DAGs weiterhin während der Wartungsfenster ausführen, solange es akzeptabel ist, dass einige Aufgaben unterbrochen und wiederholt werden können. Weitere Informationen dazu, wie sich Wartungsfenster auf Ihre Umgebung auswirken, finden Sie unter Wartungsfenster angeben.

Verwendung von "wait_for_downstream" in Ihren DAGs

Wenn Sie den Parameter wait_for_downstream in den DAGs auf True setzen, müssen, damit eine Aufgabe erfolgreich ist, alle Aufgaben, die in Bezug auf diese Aufgabe unmittelbar nachgelagert sind, ebenfalls erfolgreich ausgeführt werden. Dies bedeutet, dass die Ausführung von Aufgaben, die zu einer bestimmten DAG-Ausführung gehören, durch die Ausführung von Aufgaben aus der vorherigen DAG-Ausführung verlangsamt werden kann. Weitere Informationen dazu finden Sie in der Airflow-Dokumentation.

Aufgaben, die zu lange in der Warteschlange stehen, werden abgebrochen und neu geplant

Wenn eine Airflow-Aufgabe zu lange in der Warteschlange bleibt, plant der Planer sie nach Ablauf der in der Airflow-Konfigurationsoption [scheduler]task_queued_timeout festgelegten Zeit noch einmal zur Ausführung ein. Der Standardwert ist 2400. In Airflow-Versionen vor 2.3.1 wird die Aufgabe auch als fehlgeschlagen markiert und wiederholt, wenn sie für eine Wiederholung infrage kommt.

Eine Möglichkeit, die Symptome dieser Situation zu beobachten, ist das Diagramm mit der Anzahl der Aufgaben in der Warteschlange (Tab „Monitoring“ in der Managed Airflow-UI). Wenn die Spitzen in diesem Diagramm nicht innerhalb von etwa zwei Stunden sinken, werden die Aufgaben höchstwahrscheinlich neu geplant (ohne Logs). Danach werden in den Planer-Logs Logeinträge wie „Adopted tasks were still pending ...“ angezeigt. In solchen Fällen wird in den Airflow-Aufgaben-Logs möglicherweise die Meldung „Log file is not found...“ angezeigt, da die Aufgabe nicht ausgeführt wurde.

Im Allgemeinen ist dieses Verhalten zu erwarten und die nächste Instanz der geplanten Aufgabe soll gemäß Zeitplan ausgeführt werden. Wenn Sie viele solcher Fälle in Ihren Managed Airflow-Umgebungen beobachten, bedeutet das möglicherweise, dass nicht genügend Airflow-Worker in Ihrer Umgebung vorhanden sind, um alle geplanten Aufgaben zu verarbeiten.

Lösung: Damit dieses Problem behoben wird, muss dafür gesorgt werden, dass in Airflow-Workern immer Kapazität vorhanden ist, um Aufgaben in der Warteschlange auszuführen. Sie können beispielsweise die Anzahl der Worker oder die Worker-Gleichzeitigkeit erhöhen. Sie können auch die Parallelität oder Pools anpassen, um zu verhindern, dass mehr Aufgaben in die Warteschlange gestellt werden, als Sie Kapazität haben.

Ansatz von Managed Airflow für den Parameter min_file_process_interval

Managed Airflow ändert die Verwendung von [scheduler]min_file_process_interval durch den Airflow-Planer.

Airflow 1

Wenn Managed Airflow Airflow 1 verwendet, können Nutzer den Wert von [scheduler]min_file_process_interval zwischen 0 und 600 Sekunden festlegen. Werte über 600 Sekunden führen zu denselben Ergebnissen wie wenn [scheduler]min_file_process_interval auf 600 Sekunden gesetzt ist.

Airflow 2

In Managed Airflow-Versionen vor 1.19.9 wird [scheduler]min_file_process_interval ignoriert.

Managed Airflow-Versionen nach 1.19.9:

Der Airflow-Planer wird nach einer bestimmten Anzahl von Malen neu gestartet, wenn alle DAGs geplant sind. Der [scheduler]num_runs Parameter steuert, wie oft dies vom Planer ausgeführt wird. Wenn der Planer [scheduler]num_runs Planungsschleifen erreicht, wird er neu gestartet. Der Planer ist eine zustandslose Komponente und ein solcher Neustart ist ein Mechanismus zur automatischen Reparatur für alle Probleme, die beim Planer auftreten können. Der Standardwert von [scheduler]num_runs ist 5000.

Mit [scheduler]min_file_process_interval kann konfiguriert werden, wie oft das DAG-Parsing erfolgt. Dieser Parameter darf jedoch nicht länger sein als die Zeit, die ein Planer benötigt, um [scheduler]num_runs-Schleifen beim Planen Ihrer DAGs auszuführen.

Aufgaben nach Erreichen von dagrun_timeout als fehlgeschlagen markieren

Der Planer markiert Aufgaben, die nicht abgeschlossen sind (werden ausgeführt, geplant und in der Warteschlange) als fehlgeschlagen, wenn eine DAG-Ausführung nicht innerhalb von dagrun_timeout (ein DAG-Parameter) abgeschlossen wird.

Lösung:

Symptome einer hohen Last der Airflow-Datenbank

Manchmal sehen Sie in den Airflow-Planer-Logs möglicherweise den folgenden Warnungsprotokolleintrag:

Scheduler heartbeat got an exception: (_mysql_exceptions.OperationalError) (2006, "Lost connection to MySQL server at 'reading initial communication packet', system error: 0")"

Ähnliche Symptome können auch in Airflow-Worker-Logs beobachtet werden:

Für MySQL:

(_mysql_exceptions.OperationalError) (2006, "Lost connection to MySQL server at
'reading initial communication packet', system error: 0")"

Für PostgreSQL:

psycopg2.OperationalError: connection to server at ... failed

Solche Fehler oder Warnungen können ein Symptom dafür sein, dass die Airflow-Datenbank durch die Anzahl der offenen Verbindungen oder die Anzahl der Abfragen, die gleichzeitig ausgeführt werden, überlastet ist. Dies kann durch Planer oder andere Airflow-Komponenten wie Worker, Triggerer und Webserver verursacht werden.

Mögliche Lösungen:

Webserver zeigt die Warnung „The scheduler does not appear to be running“ an

Der Planer meldet seinen Heartbeat regelmäßig an die Airflow-Datenbank. Anhand dieser Informationen ermittelt der Airflow-Webserver, ob der Planer aktiv ist.

Wenn der Planer stark ausgelastet ist, kann er seinen Heartbeat möglicherweise nicht alle melden [scheduler]scheduler_heartbeat_sec.

In einer solchen Situation zeigt der Airflow-Webserver möglicherweise die folgende Warnung an:

The scheduler does not appear to be running. Last heartbeat was received <X>
seconds ago.

Mögliche Lösungen:

  • Erhöhen Sie die CPU- und Arbeitsspeicherressourcen für den Planer.

  • Optimieren Sie Ihre DAGs, damit das Parsing und die Planung schneller erfolgen und nicht zu viele Ressourcen des Planers verbraucht werden.

  • Verwenden Sie keine globalen Variablen in Airflow-DAGs. Verwenden Sie stattdessen Umgebungsvariablen und Airflow-Variablen.

  • Erhöhen Sie den Wert der [scheduler]scheduler_health_check_threshold Airflow-Konfigurationsoption, damit der Webserver länger wartet, bevor er die Nichtverfügbarkeit des Planers meldet.

Problemumgehungen für Probleme, die beim Backfilling von DAGs auftreten

Manchmal möchten Sie möglicherweise DAGs noch einmal ausführen, die bereits ausgeführt wurden. Sie können dies mit einem Airflow-Befehlszeilenbefehl so tun:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location LOCATION \
   dags backfill -- -B \
   -s START_DATE \
   -e END_DATE \
   DAG_NAME

Wenn Sie nur fehlgeschlagene Aufgaben für einen bestimmten DAG noch einmal ausführen möchten, verwenden Sie auch das Argument --rerun-failed-tasks.

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location LOCATION \
  backfill -- -B \
  -s START_DATE \
  -e END_DATE \
  DAG_NAME

Wenn Sie nur fehlgeschlagene Aufgaben für einen bestimmten DAG noch einmal ausführen möchten, verwenden Sie auch das Argument --rerun_failed_tasks.

Ersetzen Sie:

  • ENVIRONMENT_NAME durch den Namen der Umgebung.
  • LOCATION durch die Region, in der sich die Umgebung befindet.
  • START_DATE durch einen Wert für den DAG-Parameter start_date im Format YYYY-MM-DD.
  • END_DATE durch einen Wert für den DAG-Parameter end_date im Format YYYY-MM-DD.
  • DAG_NAME durch den Namen des DAG.

Der Backfill-Vorgang kann manchmal zu einer Deadlock-Situation führen, in der ein Backfill nicht möglich ist, weil eine Aufgabe gesperrt ist. Beispiel:

2022-11-08 21:24:18.198 CET DAG ID Task ID Run ID Try number
2022-11-08 21:24:18.201 CET -------- --------- -------- ------------
2022-11-08 21:24:18.202 CET 2022-11-08 21:24:18.203 CET These tasks are deadlocked:
2022-11-08 21:24:18.203 CET DAG ID Task ID Run ID Try number
2022-11-08 21:24:18.204 CET ----------------------- ----------- ----------------------------------- ------------
2022-11-08 21:24:18.204 CET <DAG name> <Task name> backfill__2022-10-27T00:00:00+00:00 1
2022-11-08 21:24:19.249 CET Command exited with return code 1
...
2022-11-08 21:24:19.348 CET Failed to execute job 627927 for task backfill

In einigen Fällen können Sie die folgenden Problemumgehungen verwenden, um Deadlocks zu vermeiden:

  • Deaktivieren Sie den Mini-Planer, indem Sie überschreiben die [core]schedule_after_task_execution auf False.

  • Führen Sie Backfills für kleinere Zeiträume aus. Legen Sie beispielsweise START_DATE und END_DATE so fest, dass ein Zeitraum von nur einem Tag angegeben wird.

Nächste Schritte