Airflow-Planerprobleme beheben

Cloud Composer 1 Cloud Composer 2

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

Problemursache identifizieren

Ermitteln Sie zu Beginn der Fehlerbehebung, ob das Problem zur DAG-Parsing-Zeit oder bei der Verarbeitung von Aufgaben zur Ausführungszeit auftritt. Weitere Informationen zur Parsing- und zur Ausführungszeit finden Sie unter Unterschied zwischen der DAG-Parsing-Zeit und der DAG-Ausführungszeit.

DAG-Prozessorlogs prüfen

Bei komplexen DAGs parst der DAG-Prozessor, der vom Planer ausgeführt wird, möglicherweise nicht alle Ihre DAGs. Dies kann zu vielen Problemen mit den folgenden Symptomen führen.

Symptome:

  • Wenn beim Parsen der DAGs Probleme mit dem DAG-Prozessor auftreten, kann dies zu einer Kombination der unten aufgeführten Probleme führen. Wenn DAGs dynamisch generiert werden, sind diese Probleme möglicherweise stärker als statische DAGs.

  • DAGs sind in der Airflow-UI und der DAG-UI nicht sichtbar.

  • Die Ausführung von DAGs ist nicht geplant.

  • Die DAG-Prozessorlogs enthalten Fehler, z. B.:

    dag-processor-manager [2023-04-21 21:10:44,510] {manager.py:1144} ERROR -
Processor for /home/airflow/gcs/dags/dag-example.py with PID 68311 started
at 2023-04-21T21:09:53.772793+00:00 has timed out, killing it.

    oder

    dag-processor-manager [2023-04-26 06:18:34,860] {manager.py:948} ERROR -
Processor for /home/airflow/gcs/dags/dag-example.py exited with return
code 1.

  • Bei Airflow-Planern treten Probleme auf, die zu Neustarts des Planers führen.

  • Airflow-Tasks, die zur Ausführung geplant sind, werden abgebrochen. DAG-Ausführungen für DAGs, die nicht geparst werden konnten, werden möglicherweise als failed gekennzeichnet. Beispiel:

    airflow-scheduler Failed to get task '<TaskInstance: dag-example.task1--1
manual__2023-04-17T10:02:03.137439+00:00 [removed]>' for dag
'dag-example'. Marking it as removed.

Lösung

  • Anzahl der Parameter für das DAG-Parsing erhöhen:

    • Erhöhen Sie den Wert für dagbag-import-timeout auf mindestens 120 Sekunden (oder mehr, falls erforderlich).

    • Erhöhen Sie den Wert für dag-file-processor-timeout auf mindestens 180 Sekunden (oder mehr, falls erforderlich). Dieser Wert muss größer als dagbag-import-timeout sein.

  • Korrigieren oder entfernen Sie DAGs, die Probleme für den DAG-Prozessor verursachen.

DAG-Parsing-Zeiten prüfen

Führen Sie folgende Schritte aus, um festzustellen, ob das Problem beim DAG-Parsen auftritt.

Console

In der Google Cloud Console können Sie auf der Seite Monitoring und auf dem Tab Logs die DAG-Parsingzeiten prüfen.

Prüfen Sie die DAG-Analysezeiten auf der Monitoring-Seite von Cloud Composer:

  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 Monitoring wird geöffnet.

  3. Prüfen Sie im Tab Monitoring das Diagramm Gesamte Parsing-Zeit für alle DAG-Dateien im Abschnitt DAG-Ausführungen und identifizieren Sie mögliche Probleme.

    Im Abschnitt „DAG-Ausführungen“ auf dem Tab „Monitoring“ von Composer werden Statusmesswerte für die DAGs in Ihrer Umgebung angezeigt

Auf dem Tab Cloud Composer-Logs können Sie DAG-Analysezeiten prüfen:

  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 Monitoring wird geöffnet.

  3. Wechseln Sie zum Tab Logs und wählen Sie im Navigationsbaum Alle Logs den Abschnitt DAG-Prozessormanager aus.

  4. Prüfen Sie dag-processor-manager-Logs und ermitteln Sie mögliche Probleme.

    In den DAG-Prozessorlogs werden DAG-Parsing-Zeiten angezeigt

gcloud – Airflow 1

Verwenden Sie den list_dags-Befehl mit dem -r-Flag, um die Parsing-Zeit für alle DAGs anzuzeigen.

gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    list_dags -- -r

Ersetzen Sie:

  • ENVIRONMENT_NAME durch den Namen der Umgebung.
  • LOCATION durch die Region, in der sich die Umgebung befindet.

Die Ausgabe sieht dann ungefähr so aus:

-------------------------------------------------------------------
DagBag loading stats for /home/airflow/gcs/dags
-------------------------------------------------------------------
Number of DAGs: 5
Total task number: 13
DagBag parsing time: 0.6765180000000001
-----------+----------+---------+----------+-----------------------
file       | duration | dag_num | task_num | dags
-----------+----------+---------+----------+-----------------------
/dag_1.py  | 0.6477   |       1 |        2 | ['dag_1']
/dag_2.py  | 0.018652 |       1 |        2 | ['dag_2']
/dag_3.py  | 0.004024 |       1 |        6 | ['dag_3']
/dag_4.py  | 0.003476 |       1 |        2 | ['dag_4']
/dag_5.py  | 0.002666 |       1 |        1 | ['dag_5']
-----------+----------+---------+----------+-----------------------

Suchen Sie nach dem Wert DagBag-Parsing-Zeit. Ein hoher Wert kann darauf hinweisen, dass einer Ihrer DAGs nicht optimal implementiert ist. In der Ausgabetabelle können Sie ermitteln, welche DAGs eine lange Parsing-Zeit haben.

gcloud – Airflow 2

Mit dem Befehl dags report können Sie sich die Parsing-Zeit für alle Ihre DAGs anzeigen lassen.

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

Ersetzen Sie:

  • ENVIRONMENT_NAME durch den Namen der Umgebung.
  • LOCATION durch die Region, in der sich die Umgebung befindet.

Die Ausgabe sieht dann ungefähr so aus:

Executing within the following Kubernetes cluster namespace: composer-2-0-31-airflow-2-3-3
file                  | duration       | dag_num | task_num | dags
======================+================+=========+==========+===================
/manydagsbig.py       | 0:00:00.038334 | 2       | 10       | serial-0,serial-0
/airflow_monitoring.py| 0:00:00.001620 | 1       | 1        | airflow_monitoring

Suchen Sie für jeden in der Tabelle aufgeführten DAG nach dem duration-Wert. Ein hoher Wert kann darauf hinweisen, dass einer Ihrer DAGs nicht optimal implementiert ist. In der Ausgabetabelle können Sie erkennen, welche DAGs eine lange Parsing-Zeit haben.

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 auf dem Tab Monitoring im Bereich DAG-Ausführungen das Diagramm Airflow-Aufgaben und identifizieren Sie mögliche Probleme. Airflow-Aufgaben sind Aufgaben, die sich in Airflow in einer Warteschlange befinden. Sie können entweder an die Celery- oder die Kubernetes Executor-Broker-Warteschlange gesendet werden. Aufgaben in der Celery-Warteschlange sind Aufgabeninstanzen, die in die Celery-Broker-Warteschlange gestellt werden.

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 Cloud Composer 1 und Airflow 1

Die Effizienz des DAG-Parsings wurde in Airflow 2 erheblich verbessert. Wenn Leistungsprobleme im Zusammenhang mit dem Parsen und Planen von DAGs auftreten, sollten Sie eine Migration zu Airflow 2 in Betracht ziehen.

In Cloud Composer 1 wird der Planer zusammen mit anderen Cloud Composer-Komponenten auf Clusterknoten ausgeführt. Aus diesem Grund kann die Auslastung 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. Darüber hinaus kann sich ein einzelner Knoten, auf dem der Planer ausgeführt wird, infolge von Upgrade- oder Wartungsvorgängen ändern. Diese Einschränkung wurde in Cloud Composer 2 behoben, in dem Sie dem Planer CPU- und Arbeitsspeicherressourcen zuweisen können. Die Leistung des Planers hängt nicht von der Auslastung der Clusterknoten ab.

Begrenzte Anzahl an Threads

Wenn Sie dem DAG-Prozessormanager (dem Teil des Planers, der DAG-Dateien verarbeitet) nur eine begrenzte Anzahl von Threads verwenden, kann sich das auf die DAG-Analysezeit auswirken.

Um das Problem zu beheben, überschreiben Sie die folgenden Airflow-Konfigurationsoptionen:

  • Überschreiben Sie für Airflow 1.10.12 und ältere Versionen den Parameter max_threads:

    Bereich Schlüssel Wert Notes
    scheduler max_threads NUMBER_OF_CORES_IN_MACHINE - 1 Ersetzen Sie NUMBER_OF_CORES_IN_MACHINE durch die Anzahl der Kerne
    auf den Maschinen mit Worker-Knoten.

  • Überschreiben Sie bei Airflow 1.10.14 und höheren Versionen den Parameter parsing_processes:

    Bereich Schlüssel Wert Notes
    scheduler parsing_processes NUMBER_OF_CORES_IN_MACHINE - 1 Ersetzen Sie NUMBER_OF_CORES_IN_MACHINE durch die Anzahl der Kerne
    auf den Maschinen mit Worker-Knoten.

Anzahl und Zeitverteilung der Aufgaben

Airflow ist dafür bekannt, dass bei der Planung einer großen Zahl kleiner Aufgaben Probleme auftreten. In solchen Situationen sollten Sie eine kleinere Anzahl konsolidierter Aufgaben verwenden.

Die gleichzeitige Planung einer großen Zahl an DAGs oder Aufgaben kann ebenfalls zu Problemen führen. Um dieses Problem zu vermeiden, verteilen Sie Ihre Aufgaben gleichmäßiger über die Zeit.

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.

Die Aufgabenwarteschlangen sind zu lang

In einigen Fällen kann eine Aufgabenwarteschlange für den Planer zu lang sein. Informationen zum Optimieren von Worker- und Celery-Parametern finden Sie unter Cloud Composer-Umgebung mit Ihrem Unternehmen skalieren.

TimeTable-Feature des Airflow-Planers verwenden

Ab Airflow 2.2 können Sie einen Fahrplan für einen DAG mithilfe eines neuen Features namens TimeTable definieren.

Sie können einen Fahrplan mit einer der folgenden Methoden definieren:

Eingeschränkte Clusterressourcen

Dieser Abschnitt gilt nur für Cloud Composer 1.

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 Cloud Composer-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 dafür sein, dass der Planer langsam ist.

Aufgabenplanung während Wartungsfenstern vermeiden

Sie können bestimmte Wartungsfenster für Ihre Umgebung definieren. In diesen Zeiträumen werden Wartungsereignisse für Cloud SQL und GKE ausgeführt.

So bestimmen Sie, dass der Airflow-Planer unnötige Dateien ignoriert:

Sie können die Leistung des Airflow-Planers verbessern, wenn Sie unnötige Dateien im DAGs-Ordner überspringen. Der Airflow-Planer ignoriert Dateien und Ordner, die in der Datei .airflowignore angegeben sind.

So bestimmen Sie, dass der Airflow-Planer unnötige Dateien ignoriert:

  1. Erstellen Sie eine .airflowignore-Datei.
  2. Listen Sie in dieser Datei Dateien und Ordner auf, die ignoriert werden sollen.
  3. Laden Sie diese Datei in den Ordner /dags im Bucket Ihrer Umgebung hoch.

Weitere Informationen zum Dateiformat .airflowignore finden Sie in der Airflow-Dokumentation.

Airflow-Planer verarbeitet pausierte DAGs

Airflow-Nutzer pausieren DAGs, um ihre Ausführung zu vermeiden. Das spart den Verarbeitungszyklen der Airflow-Worker.

Der Airflow-Planer parst pausierte DAGs weiter. Wenn Sie die Leistung des Airflow-Planers wirklich verbessern möchten, verwenden Sie .airflowignore oder löschen Sie pausierte DAGs aus dem DAGs-Ordner.

Verwendung von "wait_for_downstream" in Ihren DAGs

Wenn Sie den Parameter wait_for_downstream in den DAGs auf True festlegen, 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 finden Sie in der Airflow-Dokumentation.

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

Wenn eine Airflow-Aufgabe zu lange in der Warteschlange bleibt, wird sie vom Planer noch einmal zur Ausführung geplant (in Airflow-Versionen vor 2.3.1 wird die Aufgabe ebenfalls als fehlgeschlagen markiert und wiederholt, wenn sie für einen Wiederholungsversuch infrage kommt).

Eine Möglichkeit, die Symptome dieser Situation zu beobachten, besteht darin, sich das Diagramm mit der Anzahl der Aufgaben in der Warteschlange anzusehen (Tab „Monitoring“ in der Cloud Composer-Benutzeroberfläche). Wenn die Spitzen in diesem Diagramm nicht innerhalb von etwa zwei Stunden fallen, werden die Aufgaben höchstwahrscheinlich neu geplant (ohne Logs), gefolgt von der Meldung „Angenommene Aufgaben waren noch ausstehend...“ in den Planerlogs. In solchen Fällen wird möglicherweise die Meldung „Log file is not found...“ in den Airflow-Task-Logs angezeigt, da die Task nicht ausgeführt wurde.

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

Lösung: Um dieses Problem zu beheben, müssen Sie dafür sorgen, dass in Airflow-Workern immer Kapazitäten zum Ausführen von Aufgaben in der Warteschlange vorhanden sind. Sie können beispielsweise die Anzahl der Worker oder „worker_concurrency“ erhöhen. Sie können auch die Parallelität oder Pools abstimmen, um zu verhindern, dass Aufgaben in der Warteschlange über die Ihnen zur Verfügung stehende Kapazität hinausgehen.

Es kann vorkommen, dass veraltete Aufgaben die Ausführung eines bestimmten DAG blockieren.

In regulären Fällen sollte der Airflow-Planer mit Situationen umgehen können, in denen sich veraltete Aufgaben in der Warteschlange befinden und diese aus irgendeinem Grund nicht korrekt ausgeführt werden können (z.B. wurde ein DAG, zu dem die veralteten Aufgaben gehören, gelöscht).

Wenn diese veralteten Aufgaben vom Planer nicht dauerhaft gelöscht werden, müssen Sie sie möglicherweise manuell löschen. Dies können Sie beispielsweise in der Airflow-UI tun. Dort können Sie zu (Menü > Browser > Aufgabeninstanzen) wechseln, in Warteschlange gestellte Aufgaben suchen, die zu einem veralteten DAG gehören, und sie löschen.

Führen Sie ein Upgrade Ihrer Umgebung auf Cloud Composer Version 2.1.12 oder höher durch, um dieses Problem zu beheben.

Cloud Composer-Ansatz für den [scheduler]min_file_process_interval-Parameter

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

Airflow 1

Wenn Cloud Composer 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 zum selben Ergebnis wie bei einer Einstellung von [scheduler]min_file_process_interval auf 600 Sekunden.

Airflow 2

In Airflow 2 kann [scheduler]min_file_process_interval nur mit den Versionen 1.19.9 und 2.0.26 oder höher verwendet werden

  • Cloud Composer-Versionen vor 1.19.9 und 2.0.26

    In diesen Versionen wird [scheduler]min_file_process_interval ignoriert.

  • Cloud Composer-Versionen 1.19.9 oder 2.0.26 oder neuere Versionen

    Der Airflow-Planer wird nach einer bestimmten Anzahl von geplanten DAGs neu gestartet. Der Parameter [scheduler]num_runs steuert, wie oft er vom Planer ausgeführt wird. Wenn der Planer die [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 aller Probleme, die im Planer auftreten können. Wenn keine Angabe erfolgt, wird der Standardwert [scheduler]num_runs von 5.000 angewendet.

    Mit [scheduler]min_file_process_interval kann konfiguriert werden, wie häufig das DAG-Parsing ausgeführt wird. Dieser Parameter darf jedoch nicht länger als die Zeit sein, die ein Planer zum Ausführen von [scheduler]num_runs-Schleifen bei der Planung Ihrer DAGs benötigt.

Airflow-Konfiguration skalieren

Airflow bietet Konfigurationsoptionen, die steuern, wie viele Aufgaben und DAGs es gleichzeitig ausführen kann. Wenn Sie diese Konfigurationsoptionen festlegen möchten, überschreiben Sie deren Werte für Ihre Umgebung.

  • Worker-Gleichzeitigkeit

    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 Cloud Composer-Umgebung multiplizieren, erhalten Sie die maximale Anzahl von Aufgaben, die zu einem bestimmten Zeitpunkt in Ihrer Umgebung ausgeführt werden können. Diese Anzahl ist durch die Airflow-Konfigurationsoption [core]parallelism begrenzt, die näher beschrieben wird.

    In Cloud Composer 2-Umgebungen wird der Standardwert von [celery]worker_concurrency automatisch berechnet.

    • Bei Airflow-Versionen 2.3.3 und höher ist [celery]worker_concurrency auf einen Mindestwert von 32, 12 × Worker_CPU und 8 × Worker_Arbeitsspeicher festgelegt.

    • Bei Airflow-Versionen 2.2.5 oder niedriger ist [celery]worker_concurrency auf 12 × Anzahl der Worker-CPUs gesetzt.

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

  • Maximale Anzahl aktiver Tasks 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. Dies ist ein Parameter auf DAG-Ebene.

    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.

    Lösung: [core]max_active_tasks_per_dag erhöhen.

  • 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. Cloud Composer-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 (Menü > Administrator > Pools). Passen Sie die Poolgröße an das Maß an Parallelität an, das Sie in Ihrer Umgebung erwarten.

    Normalerweise ist [core]parallelism ein Produkt aus der maximalen Anzahl von Workern und [celery]worker_concurrency.

DAGs werden aufgrund von Zeitüberschreitungen von DAG-Prozessoren nicht vom Planer geplant

Weitere Informationen zu diesem Problem finden Sie unter Fehlerbehebung bei DAGs.

Aufgaben werden nach Erreichen von dagrun_timeout als fehlgeschlagen markiert

Wenn eine DAG-Ausführung nicht innerhalb von dagrun_timeout (einem DAG-Parameter) abgeschlossen wird, markiert der Planer Aufgaben, die noch nicht abgeschlossen sind (ausgeführt, geplant und in die Warteschlange gestellt), als fehlgeschlagen.

Lösung:

Symptome einer Airflow-Datenbank unter Lastdruck

Manchmal wird in den Logs des Airflow-Planers der folgende Warnlogeintrag angezeigt:

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 gleichzeitig ausgeführten Abfragen von Planern oder anderen Airflow-Komponenten wie Workern, Triggern und Webservern überlastet ist.

Mögliche Lösungen:

Webserver zeigt die Warnung „Der Planer wird anscheinend nicht ausgeführt“ aus

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

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

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 sind und nicht zu viele Planerressourcen beansprucht werden.

  • Vermeiden Sie die Verwendung globaler Variablen in Airflow-DAGs: Cloud Composer-Umgebungsvariablen und Airflow-Variablen.

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

Problemumgehungen für Probleme beim Backfill von DAGs

Manchmal möchten Sie möglicherweise bereits ausgeführte DAGs noch einmal ausführen. Mit dem Airflow-Befehlszeilentool können Sie dazu so vorgehen:

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.

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.

Ersetzen Sie:

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

Ein 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 Deadlocks mithilfe der folgenden Problemumgehungen umgehen:

  • Deaktivieren Sie den Mini-Planer, indem Sie [core]schedule-after-task-execution mit False überschreiben.

  • Führen Sie Backfills für kürzere Zeiträume aus. Legen Sie beispielsweise START_DATE und END_DATE fest, um einen Zeitraum von nur 1 Tag anzugeben.

Nächste Schritte