Airflow-Triggerprobleme beheben

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

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

Vorgänge im Trigger blockieren

Asynchrone Aufgaben werden in Triggern gelegentlich blockiert. In den meisten Fällen sind die Probleme auf unzureichende Triggerressourcen oder Probleme mit benutzerdefiniertem asynchronem Operatorcode zurückzuführen.

Triggerprotokolle enthalten Warnmeldungen, anhand derer Sie die Ursachen für eine geringere Triggerleistung ermitteln können. Es gibt zwei nach wichtigen Warnungen suchen.

  1. Asynchroner Thread blockiert

    Triggerer's async thread was blocked for 1.2 seconds, likely due to the highly utilized environment.

    Diese Warnung weist auf Leistungsprobleme hin, die auf eine große Anzahl asynchroner Aufgaben zurückzuführen sind.

    Lösung: Um dieses Problem zu beheben, weisen Sie den Auslösern mehr Ressourcen zu, reduzieren Sie die Anzahl der gleichzeitig ausgeführten verschobenen Aufgaben oder erhöhen Sie die Anzahl der Auslöser in Ihrer Umgebung. Auch wenn Trigger aufgeschobene Aufgaben verarbeiten, ist es die dafür verantwortlich sind, die Erledigung der einzelnen Aufgaben. Wenn Sie die Anzahl der Trigger anpassen, sollten Sie auch die Anzahl der Worker-Instanzen skalieren.

  2. Eine bestimmte Aufgabe hat den asynchronen Thread blockiert.

    WARNING - Executing <Task finished coro=<TriggerRunner.run_trigger() done, defined at /opt/***/***/jobs/my-custom-code.py:609> result=None> took 0.401 second

    Diese Warnung verweist auf einen bestimmten Teil des Operatorcodes, der von Cloud Composer Aus Gründen des Designs sollten Trigger die asyncio-Bibliothek für die Ausführung von Vorgängen im Hintergrund verwenden. Eine benutzerdefinierte Implementierung eines Triggers kann nicht ordnungsgemäß den asyncio-Verträgen entsprechen, z. B. aufgrund einer falschen Verwendung der Keywords await und async im Python-Code.

    Lösung: Prüfen Sie den Code, der in der Warnung gemeldet wurde, und prüfen Sie, ob der asynchrone Vorgang richtig implementiert ist.

Zu viele Trigger

Die Anzahl der verschobenen Aufgaben ist im Messwert task_count sichtbar: wird auch im Monitoring-Dashboard Ihrer Umgebung angezeigt. Jeder Trigger erzeugt Ressourcen wie Verbindungen zu externen Ressourcen, die Speicherplatz verbrauchen.

Zurückgestellte Aufgaben, die im Monitoring-Dashboard angezeigt werden
Abbildung 1. Ausgesetzte Aufgaben im Monitoring-Dashboard (zum Vergrößern anklicken)

Grafiken zur Arbeitsspeicher- und CPU-Nutzung zeigen, dass unzureichende Ressourcen wird neu gestartet, da die Aktivitätsprüfung aufgrund fehlender Heartbeats fehlschlägt:

Trigger wird aufgrund unzureichender Ressourcen neu gestartet
Abbildung 2. Trigger wird aufgrund unzureichender Ressourcen neu gestartet (zum Vergrößern anklicken)

Lösung: So beheben Sie das Problem: den Triggern mehr Ressourcen zuweisen, die Anzahl der ausgesetzten Aufgaben zu reduzieren, die gleichzeitig ausgeführt werden, oder erhöhen Sie die Anzahl der Trigger in Ihrer Umgebung.

Absturz eines Airflow-Workers während der Ausführung des Callbacks

Nachdem der Trigger die Ausführung beendet hat, kehrt die Steuerung zu einem Airflow-Worker zurück, der eine Rückrufmethode mit einem Ausführungsslot ausführt. Diese Phase umfasst die von Celery Executor gesteuert werden, und daher die entsprechende Konfiguration und Es gelten Ressourcenlimits (z. B. parallelism oder worker_concurrency).

Wenn die Callback-Methode im Airflow-Worker fehlschlägt, der Worker fehlschlägt oder der Worker, der die Methode ausführt, neu gestartet wird, wird die Aufgabe als FAILED gekennzeichnet. In In diesem Fall führt der Wiederholungsvorgang die gesamte Aufgabe neu aus, nicht nur die .

Endlosschleife in einem Trigger

Es ist möglich, einen benutzerdefinierten Trigger-Operator so zu implementieren, blockiert die Haupttriggerer-Schleife vollständig, sodass nur der eine defekte Trigger die zu diesem Zeitpunkt ausgeführt wurden. In diesem Fall wird in den Trigger-Logs eine Warnung generiert, nachdem der fehlerhafte Trigger beendet wurde.

Triggerklasse nicht gefunden

Da der DAGs-Ordner nicht mit dem Airflow-Triggerer synchronisiert wird, Der Inline-Triggercode fehlt, wenn der Trigger ausgeführt wird. Der Fehler wird in den Protokollen der fehlgeschlagenen Aufgabe generiert:

ImportError: Module "PACKAGE_NAME" does not define a "CLASS_NAME" attribute/
class

Lösung: Importieren Sie den fehlenden Code von PyPI.

Warnmeldung zum Trigger in der Airflow-Benutzeroberfläche

In einigen Fällen wird nach dem Deaktivieren des Auslösers möglicherweise die folgende Warnmeldung in der Airflow-Benutzeroberfläche angezeigt:

The triggerer does not appear to be running. Last heartbeat was received
4 hours ago. Triggers will not run, and any deferred operator will remain
deferred until it times out or fails.

Diese Meldung kann angezeigt werden, da unvollständige Trigger in Airflow verbleiben. Datenbank. Diese Meldung bedeutet in der Regel, dass der Auslöser deaktiviert wurde, bevor alle Trigger in Ihrer Umgebung abgeschlossen wurden.

Alle Trigger, die in der Umgebung ausgeführt werden, finden Sie auf der Seite Durchsuchen > Trigger in der Airflow-Benutzeroberfläche. Sie benötigen dazu die Rolle Admin.

Lösungen:

Aufgaben verbleiben im Status „Zurückgestellt“, nachdem der Trigger deaktiviert wurde

Wenn der Auslöser deaktiviert ist, bleiben Aufgaben, die sich bereits im Status „Ausgesetzt“ befinden, in diesem Status, bis das Zeitlimit erreicht ist. Diese Zeitüberschreitung kann unendlich, abhängig von der Airflow- und DAG-Konfiguration.

Verwenden Sie eine der folgenden Lösungen:

  • Aufgaben manuell als fehlgeschlagen markieren.
  • Aktivieren Sie den Auslöser, um die Aufgaben auszuführen.

Wir empfehlen, den Auslöser nur zu deaktivieren, wenn in Ihrer Umgebung keine verzögerten Operatoren oder Aufgaben ausgeführt werden und alle verzögerten Aufgaben abgeschlossen sind.

Nächste Schritte