Fehlerbehebung bei DAGs

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Auf dieser Seite finden Sie Schritte zur Fehlerbehebung und Informationen für häufige Workflowprobleme.

Viele DAG-Ausführungsprobleme werden durch eine nicht optimale Umgebungsleistung verursacht. Folgen Sie der Anleitung unter Umgebungsleistung und -kosten optimieren, um Ihre Cloud Composer 2-Umgebung zu optimieren.

Einige Probleme bei der DAG-Ausführung können darauf zurückzuführen sein, dass der Airflow-Planer nicht richtig oder optimal funktioniert. Folgen Sie der Anleitung zur Fehlerbehebung für Planer, um diese Probleme zu beheben.

Fehlerbehebung beim Workflow

So beginnen Sie mit der Fehlerbehebung:

  1. Prüfen Sie die Airflow-Logs.

    Sie können die Logging-Ebene von Airflow erhöhen, indem Sie die folgende Airflow-Konfigurationsoption überschreiben.

    Airflow 2

    Bereich Schlüssel Wert
    logging logging_level Der Standardwert ist INFO. Legen Sie DEBUG fest, um die Ausführlichkeit von Logeinträgen zu erhöhen.

    Airflow 1

    Bereich Schlüssel Wert
    core logging_level Der Standardwert ist INFO. Legen Sie DEBUG fest, um die Ausführlichkeit von Logeinträgen zu erhöhen.

  2. Prüfen Sie das Monitoring-Dashboard.

  3. Machen Sie sich mit Cloud Monitoring vertraut.

  4. Suchen Sie in der Google Cloud Console auf den Seiten für die Komponenten Ihrer Umgebung nach Fehlern.

  5. Prüfen Sie auf der Airflow-Weboberfläche in der Grafikansicht des DAG, ob Aufgabeninstanzen fehlgeschlagen sind.

    Bereich Schlüssel Wert
    webserver dag_orientation LR, TB, RL oder BT

Fehlerbehebung bei Operatorfehlern

So beheben Sie Operatorfehler:

  1. Suchen Sie nach aufgabenspezifischen Fehlern.
  2. Prüfen Sie die Airflow-Logs.
  3. Machen Sie sich mit Cloud Monitoring vertraut.
  4. Prüfen Sie dieoperatorspezifischen Logs.
  5. Beheben Sie die Fehler.
  6. Laden Sie den DAG in den Ordner dags/ hoch.
  7. Auf der Airflow-Weboberfläche löschen Sie die vergangenen Status für den DAG.
  8. Setzen Sie den DAG fort oder führen Sie ihn aus.

Fehlerbehebung bei der Ausführung von Aufgaben

Airflow ist ein verteiltes System mit vielen Entitäten wie Planer, Executor und Workern, die über eine Aufgabenwarteschlange und die Airflow-Datenbank miteinander kommunizieren und Signale senden (wie SIGTERM). Das folgende Diagramm zeigt eine Übersicht der Verbindungen zwischen Airflow-Komponenten.

Interaktion zwischen Airflow-Komponenten
Abbildung 1. Interaktion zwischen Airflow-Komponenten (zum Vergrößern klicken)

In einem verteilten System wie Airflow kann es zu Problemen mit der Netzwerkverbindung kommen oder in der zugrunde liegenden Infrastruktur treten zeitweise Probleme auf. Dies kann zu Situationen führen, in denen Aufgaben fehlschlagen und neu geplant werden können oder Aufgaben nicht erfolgreich abgeschlossen werden (z. B. Zombie-Aufgaben oder Aufgaben, die in der Ausführung hängen geblieben sind). Airflow verfügt über Mechanismen, mit denen solche Situationen bewältigt werden können und automatisch wieder normal funktionieren. In den folgenden Abschnitten werden häufige Probleme bei der Ausführung von Aufgaben durch Airflow erläutert: Zombie-Aufgaben, Giftpillen und SIGTERM-Signale.

Fehlerbehebung bei Zombie-Aufgaben

Airflow erkennt zwei Arten von Diskrepanzen zwischen einer Aufgabe und einem Prozess, der die Aufgabe ausführt:

  • Zombie-Aufgaben sind Aufgaben, die bereits ausgeführt werden sollten, aber nicht ausgeführt werden. Dies kann passieren, wenn der Prozess der Aufgabe beendet wurde oder nicht reagiert, der Airflow-Worker den Aufgabenstatus nicht rechtzeitig gemeldet hat, weil er überlastet ist, oder wenn die VM, auf der die Aufgabe ausgeführt wird, heruntergefahren wurde. Airflow findet solche Aufgaben regelmäßig und schlägt je nach Einstellungen der Aufgabe entweder fehl oder wiederholt sie.

    Entdecke die Zombie-Aufgaben

    resource.type="cloud_composer_environment"
resource.labels.environment_name="ENVIRONMENT_NAME"
log_id("airflow-scheduler")
textPayload:"Detected zombie job"

  • Untote Aufgaben sind Aufgaben, die normalerweise nicht ausgeführt werden. Airflow sucht regelmäßig nach solchen Aufgaben und beendet sie.

Die häufigsten Gründe und Lösungen für Zombie-Aufgaben sind unten aufgeführt.

Nicht genügend Arbeitsspeicher für Airflow-Worker

Jeder Airflow-Worker kann bis zu [celery]worker_concurrency Aufgabeninstanzen gleichzeitig ausführen. Wenn die kumulative Arbeitsspeichernutzung dieser Aufgabeninstanzen das Arbeitsspeicherlimit für einen Airflow-Worker überschreitet, wird ein zufälliger Prozess beendet, um Ressourcen freizugeben.

Ereignisse mit unzureichendem Arbeitsspeicher in Airflow-Workern ermitteln

resource.type="k8s_node"
resource.labels.cluster_name="GKE_CLUSTER_NAME"
log_id("events")
jsonPayload.message:("Killed process" "airflow task")

Lösungen:

Airflow-Worker wurde entfernt

Pod-Bereinigungen sind ein normaler Teil der Ausführung von Arbeitslasten in Kubernetes. GKE entfernt Pods, wenn kein Speicherplatz mehr vorhanden ist, oder um Ressourcen für Arbeitslasten mit höherer Priorität freizugeben.

Bereinigungen von Airflow-Workern entdecken

resource.type="k8s_pod"
resource.labels.cluster_name="GKE_CLUSTER_NAME"
resource.labels.pod_name:"airflow-worker"
log_id("events")
jsonPayload.reason="Evicted"

Lösungen:

  • Wenn eine Bereinigung durch zu wenig Speicherplatz verursacht wird, können Sie die Speichernutzung reduzieren oder temporäre Dateien entfernen, sobald sie nicht benötigt werden. Alternativ können Sie den verfügbaren Speicher erhöhen oder Arbeitslasten mit KubernetesPodOperator in einem dedizierten Pod ausführen.

Airflow-Worker wurde beendet

Airflow-Worker können extern entfernt werden. Wenn derzeit ausgeführte Aufgaben während eines ordnungsgemäßen Beendigungszeitraums nicht abgeschlossen werden, werden sie unterbrochen und können als Zombies erkannt werden.

Beendigungen von Airflow-Worker-Pods ermitteln

resource.type="k8s_cluster"
resource.labels.cluster_name="GKE_CLUSTER_NAME"
protoPayload.methodName:"pods.delete"
protoPayload.response.metadata.name:"airflow-worker"

Mögliche Szenarien und Lösungen:

  • Airflow-Worker werden bei Umgebungsänderungen wie Upgrades oder Paketinstallationen neu gestartet:

    Änderungen an der Composer-Umgebung erkennen

    resource.type="cloud_composer_environment"
resource.labels.environment_name="ENVIRONMENT_NAME"
log_id("cloudaudit.googleapis.com%2Factivity")

    Sie können solche Vorgänge ausführen, wenn keine kritischen Aufgaben ausgeführt werden, oder Aufgabenwiederholungen aktivieren.

  • Verschiedene Komponenten sind während Wartungsvorgängen möglicherweise vorübergehend nicht verfügbar:

    GKE-Wartungsvorgänge entdecken

    resource.type="gke_nodepool"
resource.labels.cluster_name="GKE_CLUSTER_NAME"
protoPayload.metadata.operationType="UPGRADE_NODES"

    Sie können Wartungsfenster angeben, um Überschneidungen mit der Ausführung kritischer Aufgaben zu minimieren.

  • In Cloud Composer 2-Versionen vor 2.4.5 kann ein beendender Airflow-Worker das SIGTERM-Signal ignorieren und weiterhin Aufgaben ausführen:

    Herunterskalieren durch Composer-Autoscaling kennenlernen

    resource.type="cloud_composer_environment"
resource.labels.environment_name="ENVIRONMENT_NAME"
log_id("airflow-worker-set")
textPayload:"Workers deleted"

    Sie können ein Upgrade auf eine neuere Cloud Composer-Version durchführen, in der dieses Problem behoben ist.

Airflow-Worker war stark ausgelastet

Die für einen Airflow-Worker verfügbare Menge an CPU- und Arbeitsspeicherressourcen ist durch die Konfiguration der Umgebung begrenzt. Wenn sich eine Auslastung den Limits nähert, kommt es zu einem Ressourcenkonflikt und unnötigen Verzögerungen bei der Ausführung der Aufgabe. Im Extremfall, wenn über längere Zeit Ressourcen fehlen, kann dies zu Zombie-Aufgaben führen.

Lösungen:

Airflow-Datenbank war stark ausgelastet

Eine Datenbank wird von verschiedenen Airflow-Komponenten für die Kommunikation untereinander und insbesondere zum Speichern von Heartbeats von Aufgabeninstanzen verwendet. Ein Ressourcenmangel in der Datenbank führt zu längeren Abfragezeiten und kann die Ausführung von Aufgaben beeinträchtigen.

Lösungen:

Airflow-Datenbank war vorübergehend nicht verfügbar

Ein Airflow-Worker kann Zeit in Anspruch nehmen, um zeitweise auftretende Fehler wie temporäre Verbindungsprobleme zu erkennen und zu beheben. Er kann den standardmäßigen Schwellenwert für die Zombieerkennung überschreiten.

Airflow-Heartbeat-Timeouts

resource.type="cloud_composer_environment"
resource.labels.environment_name="ENVIRONMENT_NAME"
log_id("airflow-worker")
textPayload:"Heartbeat time limit exceeded"

Lösungen:

  • Erhöhen Sie das Zeitlimit für Zombie-Aufgaben und überschreiben Sie den Wert der Airflow-Konfigurationsoption [scheduler]scheduler_zombie_task_threshold:

    Bereich Schlüssel Wert Notes
    scheduler scheduler_zombie_task_threshold Neues Timeout (in Sekunden) Der Standardwert ist 300.

Fehlerbehebung bei der Giftpille

Poison Pill ist ein Mechanismus, mit dem Airflow-Aufgaben beendet werden.

Airflow verwendet Giftpille in folgenden Situationen:

  • Wenn ein Planer eine Aufgabe beendet, die nicht rechtzeitig abgeschlossen wurde.
  • Wenn eine Aufgabe das Zeitlimit überschreitet oder zu lange ausgeführt wird.

Wenn Airflow Poison Pill verwendet, können Sie in den Logs eines Airflow-Workers, der die Aufgabe ausgeführt hat, die folgenden Logeinträge sehen:

  INFO - Subtask ... WARNING - State of this instance has been externally set
  to success. Taking the poison pill.
  INFO - Subtask ... INFO - Sending Signals.SIGTERM to GPID <X>
  INFO - Subtask ... ERROR - Received SIGTERM. Terminating subprocesses.

Mögliche Lösungen:

  • Überprüfen Sie den Aufgabencode auf Fehler, die eine zu lange Ausführungszeit verursachen können.
  • (Cloud Composer 2) Erhöhen Sie die CPU und den Arbeitsspeicher für Airflow-Worker, damit Aufgaben schneller ausgeführt werden.

  • Erhöhen Sie den Wert der Airflow-Konfigurationsoption [celery_broker_transport_options]visibility-timeout.

    Daher wartet der Planer länger auf den Abschluss einer Aufgabe, bevor er sie als Zombie-Aufgabe betrachtet. Diese Option ist besonders nützlich für zeitaufwendige Aufgaben, die viele Stunden dauern. Wenn der Wert zu niedrig ist (z. B. 3 Stunden), betrachtet der Planer Aufgaben, die 5 oder 6 Stunden lang ausgeführt werden, als „angehalten“ (Zombie-Aufgaben).

  • Erhöhen Sie den Wert der Airflow-Konfigurationsoption [core]killed_task_cleanup_time.

    Ein längerer Wert bietet Airflow-Workern mehr Zeit für die ordnungsgemäße Ausführung ihrer Aufgaben. Wenn der Wert zu niedrig ist, werden Airflow-Aufgaben möglicherweise abrupt unterbrochen, ohne dass Sie genügend Zeit haben, um ihre Arbeit ordnungsgemäß abzuschließen.

Fehlerbehebung bei SIGTERM-Signalen

SIGTERM-Signale werden von Linux, Kubernetes, Airflow-Planer und Celery verwendet, um Prozesse zu beenden, die für die Ausführung von Airflow-Workern oder Airflow-Aufgaben zuständig sind.

Es kann mehrere Gründe geben, warum SIGTERM-Signale in einer Umgebung gesendet werden:

  • Eine Aufgabe wurde zur Zombieaufgabe und muss gestoppt werden.

  • Der Planer hat ein Duplikat einer Aufgabe entdeckt und sendet die Signale „Poison Pill“ und „SIGTERM“ an die Aufgabe, um sie zu stoppen.

  • Beim horizontalen Pod-Autoscaling sendet die GKE-Steuerungsebene SIGTERM-Signale, um nicht mehr benötigte Pods zu entfernen.

  • Der Planer kann SIGTERM-Signale an den DagFileProcessorManager-Prozess senden. Solche SIGTERM-Signale werden vom Planer zur Verwaltung des DagFileProcessorManager-Prozesslebenszyklus verwendet und können ignoriert werden.

    Beispiel:

    Launched DagFileProcessorManager with pid: 353002
Sending Signals.SIGTERM to group 353002. PIDs of all processes in the group: []
Sending the signal Signals.SIGTERM to group 353002
Sending the signal Signals.SIGTERM to process 353002 as process group is missing.

  • Race-Bedingung zwischen dem Heartbeat-Callback und den Exit-Callbacks im „local_task_job“, der die Ausführung der Aufgabe überwacht. Wenn der Heartbeat erkennt, dass eine Aufgabe als erfolgreich markiert wurde, kann er nicht unterscheiden, ob die Aufgabe selbst erfolgreich war oder ob Airflow angewiesen wurde, die Aufgabe als erfolgreich zu betrachten. Trotzdem wird ein Aufgaben-Runner beendet, ohne auf dessen Beenden zu warten.

    Solche SIGTERM-Signale können ignoriert werden. Die Aufgabe befindet sich bereits im Status „Erfolgreich“ und die Ausführung des gesamten DAG ist nicht betroffen.

    Der Logeintrag Received SIGTERM. ist der einzige Unterschied zwischen dem regulären Beenden und dem Beenden der Aufgabe im Status „Erfolgreich“.

    Race-Bedingung zwischen Heartbeat und Exit-Callbacks
    Abbildung 2. Race-Bedingung zwischen Heartbeat und Exit-Callback (zum Vergrößern klicken)

  • Eine Airflow-Komponente nutzt mehr Ressourcen (CPU, Arbeitsspeicher) als vom Clusterknoten zulässig.

  • Der GKE-Dienst führt Wartungsvorgänge aus und sendet SIGTERM-Signale an Pods, die auf einem Knoten ausgeführt werden, der demnächst aktualisiert werden soll. Wenn eine Aufgabeninstanz mit SIGTERM beendet wird, sehen Sie in den Logs eines Airflow-Workers, der die Aufgabe ausgeführt hat, die folgenden Logeinträge:

{local_task_job.py:211} WARNING - State of this instance has been externally
set to queued. Terminating instance. {taskinstance.py:1411} ERROR - Received
SIGTERM. Terminating subprocesses. {taskinstance.py:1703} ERROR - Task failed
with exception

Mögliche Lösungen:

Dieses Problem tritt auf, wenn eine VM, auf der die Aufgabe ausgeführt wird, nicht genügend Arbeitsspeicher hat. Dies bezieht sich nicht auf Airflow-Konfigurationen, sondern auf die Größe des für die VM verfügbaren Arbeitsspeichers.

Das Erhöhen des Arbeitsspeichers hängt von der verwendeten Cloud Composer-Version ab. Beispiel:

  • In Cloud Composer 2 können Sie Airflow-Workern mehr CPU- und Arbeitsspeicherressourcen zuweisen.

  • Bei Cloud Composer 1 können Sie die Umgebung mit einem Maschinentyp mit mehr Leistung neu erstellen.

  • In beiden Versionen von Cloud Composer können Sie den Wert der Airflow-Konfigurationsoption für Nebenläufigkeit [celery]worker_concurrency verringern. Mit dieser Option wird festgelegt, wie viele Aufgaben von einem bestimmten Airflow-Worker gleichzeitig ausgeführt werden.

Weitere Informationen zum Optimieren der Cloud Composer 2-Umgebung finden Sie unter Umgebungsleistung und -kosten optimieren.

Cloud Logging-Abfragen zur Ermittlung von Gründen für Pod-Neustarts oder -Bereinigungen

Die Umgebungen von Cloud Composer verwenden GKE-Cluster als Computing-Infrastrukturebene. In diesem Abschnitt finden Sie nützliche Abfragen, mit denen sich Gründe für Neustarts oder Bereinigungen von Airflow-Workern oder Airflow-Planern ermitteln lassen.

Die unten dargestellten Abfragen könnten folgendermaßen optimiert werden:

  • Sie können in Cloud Logging einen für Sie interessanten Zeitraum angeben, z. B. die letzten 6 Stunden oder 3 Tage, oder einen benutzerdefinierten Zeitraum definieren.

  • Sie sollten die CLUSTER_NAME von Cloud Composer angeben.

  • Sie können die Suche auch auf einen bestimmten Pod beschränken. Fügen Sie dazu den Parameter POD_NAME hinzu.

Neu gestartete Container ermitteln

    resource.type="k8s_node"
    log_id("kubelet")
    jsonPayload.MESSAGE:"will be restarted"
    resource.labels.cluster_name="CLUSTER_NAME"

Alternative Abfrage, um die Ergebnisse auf einen bestimmten Pod zu beschränken:

    resource.type="k8s_node"
    log_id("kubelet")
    jsonPayload.MESSAGE:"will be restarted"
    resource.labels.cluster_name="CLUSTER_NAME"
    "POD_NAME"

Erkennen, dass Container aufgrund eines Ereignisses des unzureichenden Arbeitsspeichers heruntergefahren wurden

    resource.type="k8s_node"
    log_id("events")
    (jsonPayload.reason:("OOMKilling" OR "SystemOOM")
      OR jsonPayload.message:("OOM encountered" OR "out of memory"))
    severity=WARNING
    resource.labels.cluster_name="CLUSTER_NAME"

Alternative Abfrage, um die Ergebnisse auf einen bestimmten Pod zu beschränken:

    resource.type="k8s_node"
    log_id("events")
    (jsonPayload.reason:("OOMKilling" OR "SystemOOM")
      OR jsonPayload.message:("OOM encountered" OR "out of memory"))
    severity=WARNING
    resource.labels.cluster_name="CLUSTER_NAME"
    "POD_NAME"

Container ermitteln, die nicht mehr ausgeführt werden

    resource.type="k8s_node"
    log_id("kubelet")
    jsonPayload.MESSAGE:"ContainerDied"
    severity=DEFAULT
    resource.labels.cluster_name="CLUSTER_NAME"

Alternative Abfrage, um die Ergebnisse auf einen bestimmten Pod zu beschränken:

    resource.type="k8s_node"
    log_id("kubelet")
    jsonPayload.MESSAGE:"ContainerDied"
    severity=DEFAULT
    resource.labels.cluster_name="CLUSTER_NAME"
    "POD_NAME"

Auswirkungen von Aktualisierungs- oder Upgradevorgängen auf die Ausführung von Airflow-Aufgaben

Aktualisierungs- oder Upgradevorgänge unterbrechen die aktuell ausgeführte Ausführung von Airflow-Aufgaben, es sei denn, eine Aufgabe wird im zurückgestellten Modus ausgeführt.

Wir empfehlen, diese Vorgänge durchzuführen, wenn Sie minimale Auswirkungen auf die Ausführung von Airflow-Aufgaben erwarten. Richten Sie außerdem entsprechende Wiederholungsmechanismen in Ihren DAGs und Aufgaben ein.

Fehlerbehebung bei KubernetesExecutor-Aufgaben

CeleryKubernetesExecutor ist ein Executor-Typ in Cloud Composer 3, der CeleryExecutor und KubernetesExecutor gleichzeitig verwenden kann.

Weitere Informationen zur Fehlerbehebung bei Aufgaben, die mit KubernetesExecutor ausgeführt werden, finden Sie auf der Seite CeleryKubernetesExecutor verwenden.

Allgemeine Probleme

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

Airflow-Task wurde von Negsignal.SIGKILL unterbrochen

Manchmal beansprucht Ihre Aufgabe mehr Arbeitsspeicher, als dem Airflow-Worker zugewiesen ist. In einer solchen Situation kann es zu einer Unterbrechung durch Negsignal.SIGKILL kommen. Das System sendet dieses Signal, um einen weiteren Arbeitsspeicherverbrauch zu vermeiden, der die Ausführung anderer Airflow-Aufgaben beeinträchtigen könnte. Das Log des Airflow-Workers enthält möglicherweise den folgenden Logeintrag:

{local_task_job.py:102} INFO - Task exited with return code Negsignal.SIGKILL

Negsignal.SIGKILL kann auch als Code -9 angezeigt werden.

Mögliche Lösungen:

  • Niedrigere worker_concurrency von Airflow-Workern.

  • Erhöhen Sie im Fall von Cloud Composer 2 den Arbeitsspeicher der Airflow-Worker.

  • Führen Sie für Cloud Composer 1 ein Upgrade auf einen größeren Maschinentyp aus, der im Cloud Composer-Cluster verwendet wird.

  • Optimieren Sie Ihre Aufgaben, um weniger Arbeitsspeicher zu verbrauchen.

  • Verwalten Sie ressourcenintensive Aufgaben in Cloud Composer mit KubernetesPodOperator oder GKEStartPodOperator zur Aufgabenisolierung und benutzerdefinierten Ressourcenzuweisung.

Aufgabe schlägt fehl, ohne Logs aufgrund von DAG-Parsing-Fehlern auszugeben

Manchmal können subtile DAG-Fehler vorliegen, die dazu führen, dass ein Airflow-Planer und ein DAG-Prozessor in der Lage sind, Aufgaben zur Ausführung zu planen und eine DAG-Datei zu parsen. Der Airflow-Worker kann Aufgaben aus einem solchen DAG jedoch nicht ausführen, da die Python-DAG-Datei Programmierfehler enthält. Dies kann dazu führen, dass eine Airflow-Aufgabe als Failed markiert ist und es kein Log ihrer Ausführung gibt.

Lösungen:

  • Prüfen Sie in den Airflow-Worker-Logs, dass keine vom Airflow-Worker ausgelösten Fehler aufgrund fehlender DAG- oder DAG-Parsing-Fehler vorliegen.

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

    • Erhöhen Sie dagbag-import-timeout auf mindestens 120 Sekunden oder mehr, falls erforderlich.

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

  • Weitere Informationen finden Sie unter DAG-Prozessorlogs überprüfen.

Aufgabe schlägt fehl, ohne dass Logs aufgrund von Ressourcenmangel ausgegeben werden

Symptom: Während der Ausführung einer Aufgabe wird der für die Ausführung der Airflow-Aufgabe verantwortliche Unterprozess des Airflow-Workers plötzlich unterbrochen. Der im Log des Airflow-Workers angezeigte Fehler sieht in etwa so aus:

...
File "/opt/python3.8/lib/python3.8/site-packages/celery/app/trace.py", line 412, in trace_task    R = retval = fun(*args, **kwargs)  File "/opt/python3.8/lib/python3.8/site-packages/celery/app/trace.py", line 704, in __protected_call__    return self.run(*args, **kwargs)  File "/opt/python3.8/lib/python3.8/site-packages/airflow/executors/celery_executor.py", line 88, in execute_command    _execute_in_fork(command_to_exec)  File "/opt/python3.8/lib/python3.8/site-packages/airflow/executors/celery_executor.py", line 99, in _execute_in_fork
raise AirflowException('Celery command failed on host: ' + get_hostname())airflow.exceptions.AirflowException: Celery command failed on host: airflow-worker-9qg9x
...

Lösung:

Aufgabe schlägt fehl, ohne dass Logs aufgrund der Pod-Bereinigung ausgegeben werden

Google Kubernetes Engine-Pods unterliegen dem Kubernetes-Pod-Lebenszyklus und der Pod-Bereinigung. Spitzen bei Aufgaben und gemeinsame Planung von Workern sind zwei der häufigsten Ursachen für die Bereinigung von Pods in Cloud Composer.

Eine Pod-Bereinigung kann auftreten, wenn ein bestimmter Pod Ressourcen eines Knotens relativ zu den konfigurierten Erwartungen in Sachen Ressourcenverbrauch für den Knoten übernutzt. Die Bereinigung kann beispielsweise auftreten, wenn mehrere speicherintensive Aufgaben in einem Pod ausgeführt werden und deren kombinierte Last dazu führt, dass der Knoten, auf dem dieser Pod ausgeführt wird, das Limit für die Arbeitsspeichernutzung überschreitet.

Wenn ein Airflow-Worker-Pod bereinigt wird, werden alle auf diesem Pod ausgeführten Aufgabeninstanzen unterbrochen und später von Airflow als fehlgeschlagen markiert.

Logs werden zwischengespeichert. Wenn ein Worker-Pod entfernt wird, bevor der Zwischenspeicher geleert wurde, werden keine Logs ausgegeben. Ein Aufgabenfehler ohne Logs ist ein Hinweis darauf, dass die Airflow-Worker aufgrund von zu wenig Arbeitsspeicher neu gestartet werden. In Cloud Logging sind möglicherweise einige Logs vorhanden, obwohl die Airflow-Logs nicht ausgegeben wurden.

So können die Logs angezeigt werden:

  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.

  4. Unter Alle Logs -> Airflow-Logs -> Worker -> (einzelner Worker) können Sie Logs einzelner Worker ansehen.

Die DAG-Ausführung ist arbeitsspeicherbegrenzt. Jede Ausführung einer Aufgabe beginnt mit zwei Airflow-Prozessen: Ausführung und Monitoring. Jeder Knoten kann bis zu 6 Aufgaben gleichzeitig ausführen (ungefähr 12 Prozesse mit Airflow-Modulen). Je nach Art des DAG wird möglicherweise mehr Arbeitsspeicher belegt.

Symptom:

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

    Zu Arbeitslasten

  2. Wenn airflow-worker-Pods vorhanden sind, die Evicted anzeigen, klicken Sie auf die einzelnen bereinigten Pods und suchen Sie oben im Fenster nach folgender Meldung: The node was low on resource: memory.

Lösung:

  • Erstellen Sie in Cloud Composer 1 eine neue Cloud Composer-Umgebung mit einem größeren Maschinentyp als dem aktuellen Maschinentyp.
  • In Cloud Composer 2 erhöhen Sie die Arbeitsspeicherlimits für Airflow-Worker.
  • Prüfen Sie die Logs von airflow-worker-Pods auf mögliche Ursachen für die Bereinigung. Weitere Informationen zum Abrufen von Logs aus einzelnen Pods finden Sie unter Fehlerbehebung bei bereitgestellten Arbeitslasten.
  • Achten Sie darauf, dass die Aufgaben im DAG idempotent und wiederholbar sind.

  • Vermeiden Sie das Herunterladen unnötiger Dateien auf das lokale Dateisystem der Airflow-Worker.

    Airflow-Worker haben eine begrenzte lokale Dateisystemkapazität. In Cloud Composer 2 kann ein Worker beispielsweise 1 GB bis 10 GB Speicherplatz haben. Wenn der Speicherplatz aufgebraucht ist, wird der Airflow-Worker-Pod von der GKE-Steuerungsebene entfernt. Dadurch schlagen alle Aufgaben fehl, die der entfernte Worker ausgeführt hat.

    Beispiele für problematische Vorgänge:

    • Dateien oder Objekte herunterladen und lokal in einem Airflow-Worker speichern Speichern Sie diese Objekte stattdessen direkt in einem geeigneten Dienst, z. B. in einem Cloud Storage-Bucket.
    • Über einen Airflow-Worker auf große Objekte im Ordner /data zugreifen Der Airflow-Worker lädt das Objekt in sein lokales Dateisystem herunter. Implementieren Sie stattdessen Ihre DAGs, damit große Dateien außerhalb des Airflow-Worker-Pods verarbeitet werden.

Zeitüberschreitung beim Laden des DAG

Symptom:

  • Auf der Airflow-Weboberfläche wird oben auf der Seite mit der DAGs-Liste in einem roten Benachrichtigungsfeld Broken DAG: [/path/to/dagfile] Timeout angezeigt.

  • In Cloud Monitoring: Die airflow-scheduler-Logs enthalten Einträge wie die Folgenden:

    • ERROR - Process timed out
    • ERROR - Failed to import: /path/to/dagfile
    • AirflowTaskTimeout: Timeout

Lösung:

Überschreiben Sie die Airflow-Konfigurationsoption dag_file_processor_timeout und lassen Sie mehr Zeit für das DAG-Parsing:

Bereich Schlüssel Wert
core dag_file_processor_timeout Neuer Wert für die Zeitüberschreitung

DAG-Ausführung endet nicht innerhalb der erwarteten Zeit

Symptom:

Manchmal endet eine DAG-Ausführung nicht, weil Airflow-Aufgaben hängen bleiben und die DAG-Ausführung länger als erwartet dauert. Unter normalen Bedingungen bleiben Airflow-Aufgaben nicht unbegrenzt in der Warteschlange oder werden ausgeführt, da Airflow Zeitüberschreitungs- und Bereinigungsverfahren bietet, um diese Situation zu vermeiden.

Lösung:

  • Verwenden Sie für die DAGs den Parameter dagrun_timeout. Beispiel: dagrun_timeout=timedelta(minutes=120). Daher muss jede DAG-Ausführung innerhalb des Zeitlimits für die DAG-Ausführung abgeschlossen werden. Nicht abgeschlossene Aufgaben werden als Failed oder Upstream Failed gekennzeichnet. Weitere Informationen zum Airflow-Taskstatus finden Sie in der Apache Airflow-Dokumentation.

  • Verwenden Sie den Parameter Zeitlimit für Aufgabenausführung, um ein Standardzeitlimit für Aufgaben zu definieren, die auf der Grundlage von Apache Airflow-Operatoren ausgeführt werden.

DAG-Ausführungen nicht ausgeführt

Symptom:

Wenn ein geplantes Datum 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.

  • Frühere 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.

Lösung:

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

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

  • Vermeiden Sie die Verwendung von datetime.now() oder days_ago(<number of days>), es sei denn, Sie sind sich der Nebenwirkungen dieses Ansatzes bewusst.

Erhöhter Netzwerktraffic zur und von der Airflow-Datenbank

Die Menge des Netzwerk-Traffics zwischen dem GKE-Cluster Ihrer Umgebung und der Airflow-Datenbank hängt von der Anzahl der DAGs, der Anzahl der Aufgaben in DAGs und der Art des Zugriffs der DAGs auf Daten in der Airflow-Datenbank ab. Folgende Faktoren können sich auf die Netzwerknutzung auswirken:

  • Abfragen an die Airflow-Datenbank Wenn Ihre DAGs viele Abfragen ausführen, generieren sie große Mengen an Traffic. Beispiele: Status prüfen, bevor andere Aufgaben ausgeführt werden, XCom-Tabelle abfragen und Inhalt der Airflow-Datenbank auslesen.

  • Große Anzahl von Aufgaben Je mehr Aufgaben geplant sind, desto mehr Netzwerktraffic wird generiert. Dieser Aspekt gilt sowohl für die Gesamtzahl der Aufgaben in Ihren DAGs als auch für die Planungshäufigkeit. Wenn der Airflow-Planer DAG-Ausführungen plant, sendet er Abfragen an die Airflow-Datenbank und generiert Traffic.

  • Die Airflow-Weboberfläche generiert Netzwerkverkehr, da sie Abfragen an die Airflow-Datenbank sendet. Die intensive Verwendung von Seiten mit Grafiken, Aufgaben und Diagrammen kann große Mengen an Netzwerkverkehr generieren.

DAG führt zum Absturz des Airflow-Webservers oder verursacht einen 502 gateway timeout-Fehler

Webserverausfälle können aus verschiedenen Gründen auftreten. Prüfen Sie die airflow-webserver-Logs in Cloud Logging, um die Ursache des Fehlers 502 gateway timeout zu ermitteln.

Komplexe Berechnungen

Dieser Abschnitt gilt nur für Cloud Composer 1.

Führen Sie keine komplexen Berechnungen während der DAG-Analyse durch.

Im Gegensatz zu den Worker- und Planerknoten, deren Maschinentypen so angepasst werden können, dass sie eine größere CPU- und Arbeitsspeicherkapazität haben, verwendet der Webserver einen festen Maschinentyp. Dies kann zu DAG-Parsing-Fehlern führen, wenn die Parse-Zeit-Berechnung zu komplex ist.

Beachten Sie, dass der Webserver 2 vCPUs und 2 GB Arbeitsspeicher hat. Der Standardwert für core-dagbag_import_timeout beträgt 30 Sekunden. Dieser Zeitlimitwert ist die Obergrenze für den Zeitraum, den Airflow für das Laden eines Python-Moduls in den Ordner dags/ benötigt.

Falsche Berechtigungen

Dieser Abschnitt gilt nur für Cloud Composer 1.

Der Webserver wird nicht unter demselben Dienstkonto ausgeführt wie die Worker und der Planer. Worker und der Planer können somit gegebenenfalls auf vom Nutzer verwaltete Ressourcen zugreifen, auf die der Webserver keinen Zugriff hat.

Vermeiden Sie den Zugriff auf nicht öffentliche Ressourcen während der DAG-Analyse. Sollte dies nicht möglich sein, müssen Sie dem Dienstkonto des Webservers Berechtigungen erteilen. Der Name des Dienstkontos wird von der Webserverdomain abgeleitet. Lautet die Domain beispielsweise example-tp.appspot.com, ist das Dienstkonto example-tp@appspot.gserviceaccount.com.

DAG-Fehler

Dieser Abschnitt gilt nur für Cloud Composer 1.

Der Webserver wird in App Engine ausgeführt und ist vom GKE-Cluster Ihrer Umgebung getrennt. Der Webserver parst die DAG-Definitionsdateien. Wenn dabei Fehler im DAG auftreten, kann es zum 502 gateway timeout kommen. Airflow funktioniert normal ohne einen funktionierenden Webserver, wenn der problematische DAG keine in GKE ausgeführten Prozesse abbricht. In diesem Fall können Sie mit gcloud composer environments run Details aus Ihrer Umgebung abrufen und um das Problem umgehen, wenn der Webserver nicht mehr verfügbar ist.

In anderen Fällen können Sie die DAG-Analyse in GKE ausführen und nach DAGs suchen, die schwerwiegende Python-Ausnahmen oder eine Zeitüberschreitung auslösen (Standardeinstellung: 30 Sekunden). Stellen Sie zur Fehlerbehebung eine Verbindung zu einer Remote-Shell in einem Airflow-Worker-Container her und testen Sie, ob Syntaxfehler vorliegen. Weitere Informationen finden Sie unter DAGs testen.

Eine große Anzahl von DAGs und Plug-ins in dags- und Plug-in-Ordnern verarbeiten

Der Inhalt der Ordner /dags und /plugins wird aus dem Bucket Ihrer Umgebung mit lokalen Dateisystemen von Airflow-Workern und -Planern synchronisiert.

Je mehr Daten in diesen Ordnern gespeichert sind, desto länger dauert die Synchronisierung. Gehen Sie dazu folgendermaßen vor:

  • Begrenzen Sie die Anzahl der Dateien in den Ordnern „/dags“ und „/plugins“. Speichern Sie nur die Mindestanzahl der erforderlichen Dateien.

  • Erhöhen Sie nach Möglichkeit den Speicherplatz für Airflow-Planer und -Worker.

  • Erhöhen Sie nach Möglichkeit die CPU und den Arbeitsspeicher der Airflow-Planer und -Worker, damit die Synchronisierung schneller durchgeführt wird.

  • Bei einer sehr großen Anzahl von DAGs teilen Sie DAGs in Batches auf, komprimieren Sie sie in ZIP-Archiven und stellen Sie diese Archive im Ordner /dags bereit. Dieser Ansatz beschleunigt den Synchronisierungsprozess von DAGs. Airflow-Komponenten dekomprimieren ZIP-Archive vor der Verarbeitung von DAGs.

  • Das Generieren von DAGs in einer Programmatik kann auch eine Methode zum Begrenzen der Anzahl der im Ordner /dags gespeicherten DAG-Dateien sein. Im Abschnitt zu programmatischen DAGs erfahren Sie, wie Sie Probleme bei der Planung und Ausführung von programmatisch generierten DAGs vermeiden können.

Programmatisch generierte DAGs nicht gleichzeitig planen

Das programmatische Generieren von DAG-Objekten aus einer DAG-Datei ist eine effiziente Methode zum Erstellen vieler ähnlicher DAGs, die nur geringe Unterschiede aufweisen.

Es ist wichtig, die Ausführung all dieser DAGs nicht sofort einzuplanen. Es ist sehr wahrscheinlich, dass Airflow-Worker nicht genügend CPU- und Arbeitsspeicherressourcen haben, um alle gleichzeitig geplanten Aufgaben auszuführen.

So vermeiden Sie Probleme bei der Planung programmatischer DAGs:

  • Erhöhen Sie die Nebenläufigkeit von Workern und skalieren Sie Ihre Umgebung, damit sie mehr Aufgaben gleichzeitig ausführen kann.
  • Generieren Sie DAGs so, dass sie ihre Zeitpläne gleichmäßig über einen Zeitraum verteilen, um zu vermeiden, dass Hunderte von Aufgaben gleichzeitig geplant werden. So haben die Airflow-Worker Zeit, alle geplanten Aufgaben auszuführen.

Fehler 504 beim Zugriff auf den Airflow-Webserver

Siehe Fehler 504 beim Zugriff auf die Airflow-UI.

Die Lost connection to Postgres server during query-Ausnahme wird während der Ausführung der Aufgabe oder unmittelbar danach ausgelöst.

Lost connection to Postgres server during query-Ausnahmen treten häufig auf, wenn die folgenden Bedingungen erfüllt sind:

  • Ihr DAG verwendet PythonOperator oder einen benutzerdefinierten Operator.
  • Ihr DAG stellt Abfragen an die Airflow-Datenbank.

Wenn mehrere Abfragen von einer aufrufbaren Funktion gesendet werden, verweisen Tracebacks möglicherweise fälschlicherweise auf die Zeile self.refresh_from_db(lock_for_update=True) im Airflow-Code. Sie ist die erste Datenbankabfrage nach der Aufgabenausführung. Die tatsächliche Ursache der Ausnahme tritt davor auf, wenn eine SQLAlchemy-Sitzung nicht ordnungsgemäß geschlossen wird.

SQLAlchemy-Sitzungen sind auf einen Thread beschränkt und werden in einer aufrufbaren Funktionssitzung erstellt, die später innerhalb des Airflow-Codes fortgesetzt werden kann. Wenn zwischen Abfragen innerhalb einer Sitzung erhebliche Verzögerungen auftreten, wurde die Verbindung möglicherweise bereits vom Postgres-Server getrennt. Das Zeitlimit für die Verbindung in Cloud Composer-Umgebungen ist auf etwa zehn Minuten festgelegt.

Lösung:

  • Verwenden Sie den Decorator airflow.utils.db.provide_session. Dieser Decorator stellt eine gültige Sitzung für die Airflow-Datenbank im session-Parameter bereit und schließt die Sitzung am Ende der Funktion ordnungsgemäß.
  • Verwenden Sie keine einzelne Funktion mit langer Ausführungszeit. Verschieben Sie stattdessen alle Datenbankabfragen in separate Funktionen, sodass es mehrere Funktionen mit dem Decorator airflow.utils.db.provide_session gibt. In diesem Fall werden Sitzungen nach dem Abrufen von Abfrageergebnissen automatisch geschlossen.

Ausführungszeit von DAGs, Aufgaben und parallelen Ausführungen desselben DAG steuern

Mit dem DAG-Parameter dagrun_timeout können Sie steuern, wie lange eine einzelne DAG-Ausführung für einen bestimmten DAG dauert. Wenn beispielsweise eine einzelne DAG-Ausführung nicht länger als eine Stunde dauern darf (unabhängig davon, ob die Ausführung erfolgreich oder fehlgeschlagen ist), sollten Sie diesen Parameter auf 3.600 Sekunden festlegen.

Sie können auch festlegen, wie lange eine einzelne Airflow-Aufgabe dauern soll. Dazu können Sie execution_timeout verwenden.

Wenn Sie steuern möchten, wie viele aktive DAG-Ausführungen für einen bestimmten DAG erforderlich sind, können Sie dafür die Airflow-Konfigurationsoption [core]max-active-runs-per-dag verwenden.

Wenn nur eine einzige Instanz eines DAG gleichzeitig ausgeführt werden soll, setzen Sie den Parameter max-active-runs-per-dag auf 1.

Probleme bei der Synchronisierung von DAGs und Plug-ins mit Planern, Workern und Webservern

Cloud Composer synchronisiert den Inhalt der Ordner /dags und /plugins mit Planern und Workern. Bestimmte Objekte in den Ordnern /dags und /plugins können verhindern, dass diese Synchronisierung richtig funktioniert, oder sie verlangsamen.

  • Der Ordner „/dags“ wird mit Planern und Workern synchronisiert. Dieser Ordner wird nicht mit Webservern in Cloud Composer 2 synchronisiert oder wenn Sie DAG Serialization in Cloud Composer 1 aktivieren.

  • Der Ordner „/plugins“ wird mit Planern, Workern und Webservern synchronisiert.

Die folgenden Probleme können auftreten:

  • Sie haben mit GZIP komprimierte Dateien mit Komprimierungstranscodierung in die Ordner /dags und /plugins hochgeladen. Dies geschieht normalerweise, wenn Sie Daten mit dem Befehl gsutil cp -Z in den Bucket hochladen.

    Lösung: Löschen Sie das Objekt, das die Komprimierungscodierung verwendet hat, und laden Sie es noch einmal in den Bucket hoch.

  • Eines der Objekte heißt ".", das heißt, ein Objekt wird nicht mit Planern und Workern synchronisiert und wird daher möglicherweise überhaupt nicht mehr synchronisiert.

    Lösung: Benennen Sie das problematische Objekt um.

  • Ein Ordner und eine Python-Datei in DAG haben denselben Namen, z. B. a.py. In diesem Fall wird die DAG-Datei nicht richtig mit den Airflow-Komponenten synchronisiert.

    Lösung: Entfernen Sie den Ordner, der denselben Namen wie eine DAG-Python-Datei hat.

  • Eines der Objekte in den Ordnern /dags oder /plugins enthält am Ende des Objektnamens das Symbol /. Solche Objekte können den Synchronisierungsprozess in die Irre führen, da das Symbol / bedeutet, dass ein Objekt ein Ordner und keine Datei ist.

    Lösung: Entfernen Sie das Symbol / aus dem Namen des problematischen Objekts.

  • Speichern Sie keine unnötigen Dateien in den Ordnern /dags und /plugins.

    Manchmal werden den von Ihnen implementierten DAGs und Plug-ins zusätzliche Dateien hinzugefügt, z. B. Dateien zum Speichern von Tests für diese Komponenten. Diese Dateien werden mit Workern und Planern synchronisiert und wirken sich auf den Zeitaufwand für das Kopieren dieser Dateien in Planer, Worker und Webserver aus.

    Lösung: Speichern Sie keine zusätzlichen und unnötigen Dateien in den Ordnern /dags und /plugins.

Done [Errno 21] Is a directory: '/home/airflow/gcs/dags/...' Fehler wird von Planern und Workern generiert

Dieses Problem tritt auf, weil Objekte in Cloud Storage einen überlappenden Namespace haben können, während Planer und Worker gleichzeitig herkömmliche Dateisysteme verwenden. Beispielsweise ist es möglich, dem Bucket einer Umgebung sowohl einen Ordner als auch ein Objekt mit demselben Namen hinzuzufügen. Wenn der Bucket mit den Planern und Workern der Umgebung synchronisiert wird, wird dieser Fehler generiert, der zu Aufgabenfehlern führen kann.

Achten Sie darauf, dass sich im Bucket der Umgebung keine überlappenden Namespaces befinden, um dieses Problem zu beheben. Wenn sich beispielsweise sowohl /dags/misc (eine Datei) als auch /dags/misc/example_file.txt (eine andere Datei) in einem Bucket befinden, wird vom Planer ein Fehler generiert.

Vorübergehende Unterbrechungen beim Herstellen einer Verbindung zur Airflow-Metadatendatenbank

Cloud Composer wird auf einer verteilten Cloud-Infrastruktur ausgeführt. Dies bedeutet, dass gelegentlich vorübergehende Probleme auftreten können, die die Ausführung Ihrer Airflow-Aufgaben beeinträchtigen können.

In solchen Situationen werden in den Logs der Airflow-Worker möglicherweise die folgenden Fehlermeldungen angezeigt:

"Can't connect to Postgres server on 'airflow-sqlproxy-service.default.svc.cluster.local' (111)"

oder

"Can't connect to Postgres server on 'airflow-sqlproxy-service.default.svc.cluster.local' (104)"

Solche vorübergehenden Probleme können auch durch Wartungsvorgänge verursacht werden, die in Ihren Cloud Composer-Umgebungen ausgeführt werden.

Normalerweise treten solche Fehler zeitweise auf. Wenn Ihre Airflow-Aufgaben idempotent sind und Sie Wiederholungsversuche konfiguriert haben, sollten Sie dagegen immun sein. Sie können auch Wartungsfenster definieren.

Ein weiterer Grund für solche Fehler kann ein Mangel an Ressourcen im Cluster Ihrer Umgebung sein. In solchen Fällen können Sie Ihre Umgebung skalieren oder optimieren, wie in den Anleitungen Umgebungen skalieren oder Umgebung optimieren beschrieben.

Eine DAG-Ausführung ist als erfolgreich markiert, enthält aber keine ausgeführten Aufgaben

Wenn die DAG-Ausführung execution_date vor der start_date des DAG liegt, werden möglicherweise DAG-Ausführungen angezeigt, die keine Aufgabenausführungen haben, aber dennoch als erfolgreich markiert sind.

Eine erfolgreiche DAG-Ausführung ohne ausgeführte Aufgaben
Abbildung 3. Erfolgreiche DAG-Ausführung ohne ausgeführte Aufgaben (zum Vergrößern klicken)

Ursache

Diese Situation kann in einem der folgenden Fälle auftreten:

  • Eine Abweichung wird durch den Zeitzonenunterschied zwischen execution_date und start_date des DAG verursacht. Das kann beispielsweise passieren, wenn Sie pendulum.parse(...) verwenden, um start_date festzulegen.

  • Der start_date des DAG ist auf einen dynamischen Wert festgelegt, z. B. airflow.utils.dates.days_ago(1)

Lösung

  • Achten Sie darauf, dass execution_date und start_date dieselbe Zeitzone verwenden.

  • Geben Sie einen statischen start_date an und kombinieren Sie ihn mit catchup=False, um zu verhindern, dass DAGs mit vergangenen Startdaten ausgeführt werden.

Ein DAG ist in der Airflow-UI oder der DAG-UI nicht sichtbar und wird vom Planer nicht geplant

Der DAG-Prozessor parst jeden DAG, bevor er vom Planer geplant werden kann und bevor ein DAG in der Airflow-UI oder der DAG-UI sichtbar wird.

Die folgenden Airflow-Konfigurationsoptionen definieren Zeitlimits für das Parsen von DAGs:

Wenn ein DAG in der Airflow- oder DAG-UI nicht sichtbar ist:

  • Prüfen Sie die DAG-Prozessorlogs, ob der DAG-Prozessor Ihren DAG ordnungsgemäß verarbeiten kann. Bei Problemen werden möglicherweise die folgenden Logeinträge in den DAG-Prozessor- oder Planerlogs angezeigt:
[2020-12-03 03:06:45,672] {dag_processing.py:1334} ERROR - Processor for
/usr/local/airflow/dags/example_dag.py with PID 21903 started at
2020-12-03T03:05:55.442709+00:00 has timed out, killing it.
  • Prüfen Sie die Planerlogs, um festzustellen, ob sie ordnungsgemäß funktionieren. Bei Problemen werden möglicherweise die folgenden Logeinträge in den Planerlogs angezeigt:
DagFileProcessorManager (PID=732) last sent a heartbeat 240.09 seconds ago! Restarting it
Process timed out, PID: 68496

Lösungen:

  • Beheben Sie alle DAG-Parsing-Fehler. Der DAG-Prozessor parst mehrere DAGs. In seltenen Fällen können Parsing-Fehler eines DAG das Parsen anderer DAGs beeinträchtigen.

  • Wenn das Parsen des DAG länger dauert als die in [core]dagrun_import_timeout definierten Sekunden, erhöhen Sie dieses Zeitlimit.

  • Wenn das Parsen aller DAGs mehr als die in [core]dag_file_processor_timeout definierten Sekunden dauert, erhöhen Sie dieses Zeitlimit.

  • Wenn das Parsen des DAG lange dauert, kann dies auch bedeuten, dass er nicht optimal implementiert wurde. Beispielsweise werden viele Umgebungsvariablen gelesen oder externe Dienste oder Airflow-Datenbanken aufgerufen. Vermeiden Sie es nach Möglichkeit, solche Vorgänge in globalen Abschnitten von DAGs auszuführen.

  • Erhöhen Sie die CPU- und Arbeitsspeicherressourcen für den Scheduler, damit er schneller arbeitet.

  • Anzahl der Planer anpassen

  • Erhöhen Sie die Anzahl der DAG-Prozessorprozesse, damit das Parsen schneller erfolgen kann. Erhöhen Sie dazu den Wert von [scheduler]parsing_process.

  • Verringern Sie die Häufigkeit des DAG-Parsings.

  • Belastung der Airflow-Datenbank verringern

Symptome einer stark ausgelasteten Airflow-Datenbank

Weitere Informationen finden Sie unter Symptome einer Airflow-Datenbank, die unter Lasten steht.

Nächste Schritte