DAG-Probleme aufgrund von unzureichendem Arbeitsspeicher und Speicherplatz beheben

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

In dieser Anleitung erfahren Sie, wie Sie einen fehlgeschlagenen Airflow-DAG in Cloud Composer beheben und mithilfe von Protokollen und Umgebungsmonitoring Probleme mit Worker-Ressourcen wie fehlendem Arbeitsspeicher oder Speicherplatz diagnostizieren.

Einführung

In diesem Tutorial liegt der Schwerpunkt auf ressourcenbezogenen Problemen, um zu zeigen, wie Sie einen DAG debuggen können.

Ein Mangel an zugewiesenen Worker-Ressourcen führt zu DAG-Fehlern. Wenn bei einer Airflow-Aufgabe der Arbeitsspeicher oder Speicherplatz aufgebraucht ist, wird möglicherweise eine Airflow-Ausnahme angezeigt, z. B.:

WARNING airflow.exceptions.AirflowException: Task received SIGTERM signal
INFO - Marking task as FAILED.

oder

Task exited with return code Negsignal.SIGKILL

In solchen Fällen wird in der Regel empfohlen, die Ressourcen der Airflow-Worker zu erhöhen oder die Anzahl der Aufgaben pro Worker zu reduzieren. Da Airflow-Ausnahmen jedoch allgemeiner Natur sein können, ist es manchmal schwierig, die jeweilige Ressource zu ermitteln, die das Problem verursacht.

In dieser Anleitung wird erläutert, wie Sie die Ursache für einen DAG-Fehler diagnostizieren und die Art der Ressource ermitteln können, die Probleme verursacht. Dazu werden zwei Beispiel-DAGs mit Fehlern aufgrund von unzureichendem Arbeitsspeicher und Speicherplatz behoben.

Lernziele

• Beispiel-DAGs ausführen, die aus den folgenden Gründen fehlschlagen:

  • Zu wenig Arbeitsspeicher für Worker
  • Zu wenig Arbeitsspeicher

• Fehlerursachen ermitteln

• Zugewiesene Worker-Ressourcen erhöhen

• DAGs mit neuen Ressourcenlimits testen

Kosten

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

• Cloud Composer (siehe zusätzliche Kosten)
• Cloud Monitoring

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

Hinweis

In diesem Abschnitt werden die Schritte beschrieben, die Sie ausführen müssen, bevor Sie mit der Anleitung beginnen.

Projekt erstellen und konfigurieren

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

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

   Zur Projektauswahl

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

3. Der Google Cloud Nutzer Ihres Projekts muss die folgenden Rollen haben, um die erforderlichen Ressourcen zu erstellen:

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

Die APIs für Ihr Projekt aktivieren

Enable the Cloud Composer API.

Enable the API

Cloud Composer-Umgebung erstellen

Erstellen Sie eine Cloud Composer 2-Umgebung.

Wichtig: Verwenden Sie die Voreinstellung für eine kleine Umgebung, um die Fehler bei Verwendung der Beispiele weiter unten in dieser Anleitung zu reproduzieren. Dies ist die Standardeinstellung beim Erstellen einer Umgebung über die Google Cloud Console.

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

Limits für Worker-Ressourcen prüfen

Prüfen Sie die Ressourcenlimits für Airflow-Worker in Ihrer Umgebung:

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

   Zur Seite Umgebungen

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

3. Rufen Sie den Tab Umgebungskonfiguration auf.

4. Gehen Sie zu Ressourcen > Konfiguration der Arbeitslasten > Worker.

5. Prüfen Sie, ob die Werte 0,5 vCPUs, 1,875 GB Arbeitsspeicher und 1 GB Speicherplatz sind. Das sind die Airflow-Nutzerressourcenlimits, mit denen Sie in den nächsten Schritten dieser Anleitung arbeiten.

Beispiel: Probleme mit unzureichendem Arbeitsspeicher diagnostizieren

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

Dieser DAG enthält eine Aufgabe, die die folgenden Schritte ausführt:

1. Erstellt eine leere Liste s.
2. Führt einen Zyklus aus, um der Liste den String More anzuhängen.
3. Gibt an, wie viel Arbeitsspeicher die Liste verbraucht, und wartet bei jeder Iteration von einer Minute 1 Sekunde.

Warnung: Dieser DAG simuliert ein Problem mit Arbeitsspeichermangel, das den Betrieb in Ihrer Umgebung beeinträchtigen kann.

import time

import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import sys
from datetime import timedelta

default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 0,
    'retry_delay': timedelta(minutes=10)
}

dag = DAG(
    'create_list_with_many_strings',
    default_args=default_args,
    schedule_interval=None)


def consume():
    s = []
    for i in range(120):
        for j in range(1000000):
            s.append("More")
        print(f"i={i}; size={sys.getsizeof(s) / (1000**3)}GB")
        time.sleep(1)


t1 = PythonOperator(
    task_id='task0',
    python_callable=consume,
    dag=dag,
    depends_on_past=False,
    retries=0
)

Beispiel-DAG auslösen

Beispiel-DAG create_list_with_many_strings auslösen:

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

   Zur Seite Umgebungen

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

3. Klicken Sie in der Airflow-Weboberfläche auf der Seite DAGs in der Spalte Links für Ihren DAG auf die Schaltfläche Trigger DAG.

4. Klickbasierter Trigger

5. Klicken Sie auf der Seite DAGs auf die ausgelöste Aufgabe und prüfen Sie die Ausgabeprotokolle, um sicherzustellen, dass der DAG gestartet wurde.

Während der Ausführung des Tasks wird in den Ausgabeprotokollen die Arbeitsspeichergröße in GB ausgegeben, die vom DAG verwendet wird.

Nach einigen Minuten schlägt die Aufgabe fehl, da sie das Arbeitsspeicherlimit von 1,875 GB für Airflow-Worker überschreitet.

Hinweis: Wenn Sie den DAG, der das Arbeitsspeicherlimit überschreitet, gleichzeitig mit einem anderen Beispiel-DAG auslösen, sehen Sie, dass andere laufende Aufgaben ebenfalls fehlschlagen, wenn eine der Aufgaben das Arbeitsspeicherlimit überschreitet und fehlschlägt, auch wenn sie keinen Arbeitsspeicher verbrauchen.

Fehlgeschlagenen DAG diagnostizieren

Wenn Sie zum Zeitpunkt des Fehlers mehrere Aufgaben ausgeführt haben, sollten Sie nur eine Aufgabe ausführen und während dieser Zeit den Ressourcendruck diagnostizieren, um herauszufinden, welche Aufgaben den Ressourcendruck verursachen und welche Ressourcen Sie erhöhen müssen.

Airflow-Aufgaben-Logs prüfen

Die Aufgabe aus dem create_list_with_many_strings-DAG hat den Status Failed.

Prüfen Sie die Logs der Aufgabe. Sie sehen den folgenden Logeintrag:

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

`Netsignal.SIGKILL` might be an indication of your task using more memory
than the Airflow worker is allocated. The system sends
the `Negsignal.SIGKILL` signal to avoid further memory consumption.

Arbeitslasten überprüfen

Prüfen Sie die Arbeitslasten, um sicherzustellen, dass die Auslastung Ihrer Aufgabe nicht dazu führt, dass der Knoten, auf dem der Pod ausgeführt wird, das Limit für die Arbeitsspeichernutzung überschreitet:

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

   Zur Seite Umgebungen

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

3. Rufen Sie den Tab Umgebungskonfiguration auf.

4. Klicken Sie unter Ressourcen > GKE-Cluster > Arbeitslasten auf Clusterarbeitslasten ansehen.

5. Prüfen Sie, ob einige der Arbeitslast-Pods einen Status haben, der dem folgenden ähnelt:

   Error with exit code 137 and 1 more issue.
   ContainerStatusUnknown with exit code 137 and 1 more issue
   

   Exit code 137 bedeutet, dass ein Container oder Pod versucht, mehr Arbeitsspeicher zu verwenden, als zulässig ist. Der Prozess wird beendet, um die Arbeitsspeichernutzung zu verhindern.

Überwachung der Umgebungsintegrität und des Ressourcenverbrauchs prüfen

Prüfen Sie den Zustand der Umgebung und die Überwachung des Ressourcenverbrauchs:

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 und wählen Sie Übersicht aus.

4. Suchen Sie im Bereich Umgebungsübersicht nach dem Diagramm Umgebungszustand (Airflow-Monitoring-DAG). Er enthält einen roten Bereich, der der Zeit entspricht, zu der in den Protokollen Fehler ausgegeben wurden.

5. Wählen Sie Worker aus und suchen Sie das Diagramm Gesamte Arbeitsspeichernutzung der Worker. Sie sehen, dass die Linie Arbeitsspeichernutzung zu dem Zeitpunkt, als die Aufgabe ausgeführt wurde, einen Ausschlag aufweist.

Abbildung 1: Diagramm zur Gesamtarbeitsspeichernutzung der Worker (zum Vergrößern anklicken)

Auch wenn die Linie für die Arbeitsspeichernutzung im Diagramm das Limit nicht erreicht, müssen Sie bei der Diagnose der Fehlerursachen nur den zuweisbaren Arbeitsspeicher berücksichtigen. Die Linie Arbeitsspeicherlimit im Diagramm steht für den gesamten verfügbaren Arbeitsspeicher (einschließlich der von GKE reservierten Kapazität).

In diesem Beispiel ist das Arbeitsspeicherlimit für Worker auf 1, 875 GB festgelegt. GKE reserviert 25% der ersten 4 GiB Arbeitsspeicher. GKE reserviert außerdem einen zusätzlichen Bereinigungsschwellenwert: 100 MiB Arbeitsspeicher pro Knoten für die Kubelet-Bereinigung.

Der zuweisbare Arbeitsspeicher wird so berechnet:

ALLOCATABLE = CAPACITY - RESERVED - EVICTION-THRESHOLD

Wenn das Speicherlimit 1,875 GB beträgt, beträgt der tatsächlich zuweisbare Arbeitsspeicher:

1.75 GiB (1.875GB) - 0.44 (25% GiB reserved) - 0.1 = 1.21 GiB (~1.3 GB).

Wenn Sie dieses tatsächliche Limit an das Diagramm zur Arbeitsspeichernutzung anhängen, sehen Sie, dass der Anstieg der Arbeitsspeichernutzung der Aufgabe das tatsächliche Arbeitsspeicherlimit erreicht. Sie können dann daraus schließen, dass die Aufgabe aufgrund von unzureichendem Arbeitsspeicher des Workers fehlgeschlagen ist.

Hinweis :Wenn die Speichernutzung zu schnell ansteigt, wird die tatsächliche Ressourcennutzung möglicherweise nicht korrekt erfasst. Der airflow_monitoring-DAG ist eine lauffähige Systemdiagnose pro Umgebung, mit der der environment/healthy-Monitoringmesswert ausgefüllt wird. Mit diesem DAG können Sie Cloud Monitoring-Funktionen wie Messwertdiagramme verwenden oder Benachrichtigungen festlegen, wenn die Umgebung nicht mehr fehlerfrei funktioniert. Er wird alle 5 Minuten ausgeführt und die Messwerte werden erfasst. Die Ressourcennutzung zwischen den Ausführungen wird daher möglicherweise nicht erfasst.

Worker-Speicherlimit erhöhen

Hinweis: Wenn Sie die CPU oder den Arbeitsspeicher des Workers ändern, werden die Worker neu gestartet. Dies kann sich auf laufende Aufgaben auswirken. Wir empfehlen, dies zu tun, wenn keine DAGs ausgeführt werden.

Weisen Sie zusätzlichen Worker-Speicher zu, damit der Beispiel-DAG ausgeführt werden kann:

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

   Zur Seite Umgebungen

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

3. Rufen Sie den Tab Umgebungskonfiguration auf.

4. Klicken Sie unter Ressourcen > Arbeitslasten auf Bearbeiten.

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

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

DAG mit dem neuen Arbeitsspeicherlimit testen

Lösen Sie den create_list_with_many_strings-DAG noch einmal aus und warten Sie, bis die Ausführung abgeschlossen ist.

1. In den Ausgabeprotokollen der DAG-Ausführung wird Marking task as SUCCESS angezeigt und der Status der Aufgabe ist Erfolgreich.

2. Prüfen Sie auf dem Tab Monitoring im Bereich Umgebungsübersicht, ob es keine roten Bereiche gibt.

3. Klicken Sie auf den Bereich Worker und suchen Sie das Diagramm Gesamte Arbeitsspeichernutzung der Worker. Sie sehen, dass die Linie Speicherlimit die Änderung des Speicherlimits widerspiegelt und die Linie Speichernutzung weit unter dem tatsächlichen zuweisbaren Speicherlimit liegt.

Beispiel: Probleme mit nicht ausreichendem Speicherplatz diagnostizieren

In diesem Schritt laden Sie zwei DAGs hoch, die große Dateien erstellen. Der erste DAG erstellt eine große Datei. Der zweite DAG erstellt eine große Datei und simuliert einen langwierigen Vorgang.

Die Größe der Datei in beiden DAGs überschreitet das Standardspeicherlimit von 1 GB für Airflow-Worker. Der zweite DAG enthält jedoch eine zusätzliche Warteaufgabe, um seine Dauer künstlich zu verlängern.

In den nächsten Schritten untersuchen Sie die Unterschiede im Verhalten der beiden DAGs.

Hinweis: Mit den DAGs in dieser Anleitung werden lokale Dateien erstellt. Speichern Sie in einer Produktionsumgebung keine Dateien oder Konfigurationen im lokalen Dateisystem, da die nächste Aufgabe wahrscheinlich auf einem anderen Airflow-Worker ohne Zugriff auf die Datei ausgeführt wird.

DAG hochladen, durch den eine große Datei erstellt wird

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

Dieser DAG enthält eine Aufgabe, die die folgenden Schritte ausführt:

1. Schreibt eine 1,5 GB große localfile.txt-Datei in den Speicher des Airflow-Workers.
2. Damit wird die Größe der erstellten Datei mithilfe des Python-Moduls os ausgegeben.
3. Gibt alle 1 Minute die Dauer der DAG-Ausführung aus.

Warnung: Diese DAG simuliert ein Problem mit nicht ausreichendem Speicherplatz, das den Betrieb in Ihrer Umgebung beeinträchtigen kann.

import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import os
from datetime import timedelta
import time

default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 0,
    'retry_delay': timedelta(minutes=10)
}

dag = DAG(
    'create_large_txt_file_print_logs',
    default_args=default_args,
    schedule_interval=None)


def consume():
    size = 1000**2  # bytes in 1 MB
    amount = 100

    def create_file():
        print(f"Start creating a huge file")
        with open("localfile.txt", "ab") as f:
            for j in range(15):
                f.write(os.urandom(amount) * size)
        print("localfile.txt size:", os.stat("localfile.txt").st_size / (1000**3), "GB")

    create_file()
    print("Success!")


t1 = PythonOperator(
    task_id='create_huge_file',
    python_callable=consume,
    dag=dag,
    depends_on_past=False,
    retries=0)

DAG hochladen, der bei einem langwierigen Vorgang eine große Datei erstellt

Wenn Sie einen lang laufenden DAG simulieren und die Auswirkungen der Aufgabendauer auf den Endstatus untersuchen möchten, laden Sie den zweiten Beispiel-DAG in Ihre Umgebung hoch. In dieser Anleitung heißt dieser DAG long_running_create_large_txt_file_print_logs.

Dieser DAG enthält eine Aufgabe, die die folgenden Schritte ausführt:

1. Schreibt eine 1,5 GB große localfile.txt-Datei in den Speicher des Airflow-Workers.
2. Damit wird die Größe der erstellten Datei mithilfe des Python-Moduls os ausgegeben.
3. Es wird 1 Stunde und 15 Minuten gewartet, um die Zeit zu simulieren, die für Vorgänge mit der Datei erforderlich ist, z. B. das Lesen aus der Datei.
4. Gibt alle 1 Minute die Dauer der DAG-Ausführung aus.

Warnung: Diese DAG simuliert ein Problem mit nicht ausreichendem Speicherplatz, das den Betrieb in Ihrer Umgebung beeinträchtigen kann.

import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import os
from datetime import timedelta
import time

default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 0,
    'retry_delay': timedelta(minutes=10)
}

dag = DAG(
    'long_running_create_large_txt_file_print_logs',
    default_args=default_args,
    schedule_interval=None)


def consume():
    size = 1000**2  # bytes in 1 MB
    amount = 100

    def create_file():
        print(f"Start creating a huge file")
        with open("localfile.txt", "ab") as f:
            for j in range(15):
                f.write(os.urandom(amount) * size)
        print("localfile.txt size:", os.stat("localfile.txt").st_size / (1000**3), "GB")

    create_file()
    for k in range(75):
        time.sleep(60)
        print(f"{k+1} minute")

    print("Success!")


t1 = PythonOperator(
    task_id='create_huge_file',
    python_callable=consume,
    dag=dag,
    depends_on_past=False,
    retries=0)

Beispiel-DAGs auslösen

Lösen Sie den ersten DAG, create_large_txt_file_print_logs, aus:

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

   Zur Seite Umgebungen

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

3. Klicken Sie in der Airflow-Weboberfläche auf der Seite DAGs in der Spalte Links für Ihren DAG auf die Schaltfläche Trigger DAG.

4. Klickbasierter Trigger

5. Klicken Sie auf der Seite DAGs auf die ausgelöste Aufgabe und prüfen Sie die Ausgabeprotokolle, um sicherzustellen, dass der DAG gestartet wurde.

6. Warten Sie, bis die Aufgabe abgeschlossen ist, die Sie mit dem create_large_txt_file_print_logs-DAG erstellt haben. Das kann einige Minuten dauern.

   Hinweis :Möglicherweise stellen Sie fest, dass der verwendete Airflow-Arbeitsspeicher nach Abschluss der Aufgabe auf Null zurückgesetzt wurde. Cloud Composer verwendet Google Kubernetes Engine Autopilot, der die erforderlichen Ressourcen bereitstellt. Das Autoscaling in Cloud Composer ist messwertbasiert und steuert die Ressourcenzuweisung über die GKE-Pod-Verwaltung innerhalb eines bestimmten Kontingents.

7. Klicken Sie auf der Seite DAGs auf die DAG-Ausführung. Die Aufgabe hat den Status Success, obwohl das Speicherlimit überschritten wurde.

Prüfen Sie die Airflow-Logs der Aufgabe:

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

   Zur Seite Umgebungen

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

   2. Rufen Sie den Tab Protokolle und dann Alle Protokolle > Airflow-Protokolle > Worker > Im Log-Explorer ansehen auf.

   3. Filtern Sie die Protokolle nach Typ: Nur Fehler-Nachrichten anzeigen.

In den Protokollen werden Meldungen ähnlich der folgenden angezeigt:

Worker: warm shutdown (Main Process)

oder

A worker pod was evicted at 2023-12-01T12:30:05Z with message: Pod ephemeral
local storage usage exceeds the total limit of containers 1023Mi.

Diese Protokolle zeigen, dass der Pod den Prozess „Warmes Herunterfahren“ gestartet hat, weil der genutzte Speicher das Limit überschritten hat und innerhalb einer Stunde entfernt wurde. Die DAG-Ausführung ist jedoch nicht fehlgeschlagen, da sie innerhalb des Kubernetes-Kündigungstoleranzzeitraums abgeschlossen wurde. Weitere Informationen dazu finden Sie in dieser Anleitung.

Das Konzept des Kulanzzeitraums für die Beendigung veranschaulicht das Ergebnis des zweiten Beispiel-DAG, long_running_create_large_txt_file_print_logs.

Zweiten DAG auslösen, long_running_create_large_txt_file_print_logs:

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

   Zur Seite Umgebungen

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

3. Klicken Sie in der Airflow-Weboberfläche auf der Seite DAGs in der Spalte Links für Ihren DAG auf die Schaltfläche Trigger DAG.

4. Klickbasierter Trigger

5. Klicken Sie auf der Seite DAGs auf die ausgelöste Aufgabe und prüfen Sie die Ausgabeprotokolle, um sicherzustellen, dass der DAG gestartet wurde.

6. Warten Sie, bis die Ausführung des long_running_create_large_txt_file_print_logs-DAG fehlschlägt. Das dauert etwa eine Stunde.

Ergebnisse der DAG-Ausführung prüfen:

1. Klicken Sie auf der Seite DAGs auf die DAG-Ausführung long_running_create_large_txt_file_print_logs. Sie sehen, dass die Aufgabe den Status Failed hat und dass die Ausführungsdauer genau 1 Stunde und 5 Minuten betrug. Das ist kürzer als die Wartezeit der Aufgabe von 1 Stunde und 15 Minuten.

2. Prüfen Sie die Logs der Aufgabe. Sobald der DAG die Datei localfile.txt im Container des Airflow-Workers erstellt hat, wird im Log ausgegeben, dass der DAG mit dem Warten begonnen hat. Die Ausführungsdauer wird alle 60 Sekunden in den Aufgabenprotokollen ausgegeben. In diesem Beispiel druckt der DAG das localfile.txt size:-Protokoll aus und die Größe der localfile.txt-Datei beträgt 1, 5 GB.

Sobald die Datei, die in den Container des Airflow-Workers geschrieben wird, das Speicherlimit überschreitet, sollte die DAG-Ausführung fehlschlagen. Die Aufgabe schlägt jedoch nicht sofort fehl, sondern wird so lange ausgeführt, bis ihre Dauer 1 Stunde und 5 Minuten erreicht. Das liegt daran, dass Kubernetes die Aufgabe nicht sofort beendet, sondern weiterlaufen lässt, um eine Wiederherstellungszeit von einer Stunde zu ermöglichen. Dieser Zeitraum wird als Kulanzzeitraum bezeichnet. Wenn einem Knoten die Ressourcen ausgehen, beendet Kubernetes den Pod nicht sofort, um die Beendigung ordnungsgemäß zu verwalten, damit die Auswirkungen auf den Endnutzer minimal sind.

Die Kulanzzeitraum für die Beendigung hilft Nutzern, Dateien nach Aufgabenausfällen wiederherzustellen. Bei der Diagnose von DAGs kann es jedoch zu Verwirrung führen. Wenn das Speicherlimit des Airflow-Arbeiters überschritten wird, hängt der Endtask-Status von der Dauer der DAG-Ausführung ab:

• Wenn der DAG-Lauf das Speicherlimit des Workers überschreitet, aber in weniger als einer Stunde abgeschlossen wird, erhält die Aufgabe den Status Success, da sie innerhalb der Kulanzzeitraums für die Beendigung abgeschlossen wurde. Kubernetes beendet jedoch den Pod und die geschriebene Datei wird sofort aus dem Container gelöscht.

• Wenn der DAG das Speicherlimit des Workers überschreitet und länger als eine Stunde ausgeführt wird, läuft er eine Stunde lang weiter und kann das Speicherlimit um Tausende von Prozent überschreiten, bevor Kubernetes den Pod entfernt und Airflow die Aufgabe als Failed kennzeichnet.

Hinweis :Wenn Sie einen der Beispiel-DAGs, die das Speicherplatzlimit überschreiten, gleichzeitig mit einem anderen Beispiel-DAG auslösen, sehen Sie, dass alle anderen laufenden Aufgaben fehlschlagen, wenn eine der Aufgaben das Speicherplatzlimit überschreitet und fehlschlägt, auch wenn sie keinen Speicherplatz benötigen.

Fehlgeschlagenen DAG diagnostizieren

Wenn Sie zum Zeitpunkt des Fehlers mehrere Aufgaben ausgeführt haben, sollten Sie nur eine Aufgabe ausführen und während dieser Zeit den Ressourcendruck diagnostizieren, um herauszufinden, welche Aufgaben den Ressourcendruck verursachen und welche Ressourcen Sie erhöhen müssen.

Sehen Sie sich die Aufgabenprotokolle des zweiten DAG an, long_running_create_large_txt_file_print_logs:

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 Protokolle und dann Alle Protokolle > Airflow-Protokolle > Worker > Im Log-Explorer ansehen auf.

4. Filtern Sie die Protokolle nach Typ: Nur Fehler-Nachrichten anzeigen.

In den Protokollen werden Meldungen ähnlich der folgenden angezeigt:

Container storage usage of worker reached 155.7% of the limit.

This likely means that the total size of local files generated by your DAGs is
close to the storage limit of worker.

You may need to decrease the storage usage or increase the worker storage limit
in your Cloud Composer environment configuration.

oder

Pod storage usage of worker reached 140.2% of the limit.
A worker pod was evicted at 2023-12-01T12:30:05Z with message: Pod ephemeral
local storage usage exceeds the total limit of containers 1023Mi.

This eviction likely means that the total size of dags and plugins folders plus
local files generated by your DAGs exceeds the storage limit of worker.

Please decrease the storage usage or increase the worker storage limit in your
Cloud Composer environment configuration.

Diese Meldungen geben an, dass während der Ausführung der Aufgabe Fehler in den Airflow-Logs ausgegeben wurden, als die Größe der von Ihrem DAG generierten Dateien das Speicherlimit des Workers überschritten hat und die Kulanzzeitraum für die Beendigung begann. Während des Kulanzzeitraums für die Kündigung ist der Speicherverbrauch nicht auf das Limit zurückgekehrt. Dies hat dazu geführt, dass der Pod nach Ablauf des Kulanzzeitraums entfernt wurde.

Prüfen Sie den Zustand der Umgebung und die Überwachung des Ressourcenverbrauchs:

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 und wählen Sie Übersicht aus.

4. Suchen Sie im Bereich Umgebungsübersicht nach dem Diagramm Umgebungszustand (Airflow-Monitoring-DAG). Er enthält einen roten Bereich, der der Zeit entspricht, zu der in den Protokollen Fehler ausgegeben wurden.

5. Wählen Sie Worker aus und suchen Sie das Diagramm Gesamte Laufwerknutzung der Worker. Sie sehen, dass die Linie Laufwerksnutzung einen Ausschlag aufweist und die Linie Laufwerklimit zum Zeitpunkt der Ausführung Ihrer Aufgabe überschreitet.

Abbildung 2. Diagramm zur Gesamtlaufwerknutzung der Worker (zum Vergrößern anklicken)

Speicherlimit für Worker erhöhen

Hinweis: Wenn Sie den Speicherplatz für Worker ändern, werden die Airflow-Worker neu gestartet. Dies kann sich auf laufende Aufgaben auswirken. Wir empfehlen, dies zu tun, wenn keine DAGs ausgeführt werden.

Weisen Sie zusätzlichen Airflow-Worker-Speicherplatz zu, damit der Beispiel-DAG ausgeführt werden kann:

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

   Zur Seite Umgebungen

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

3. Rufen Sie den Tab Umgebungskonfiguration auf.

4. Klicken Sie unter Ressourcen > Arbeitslasten auf Bearbeiten.

5. Geben Sie im Abschnitt Worker im Feld Speicher das neue Speicherlimit für Airflow-Worker an. In dieser Anleitung wird der Wert auf 2 GB festgelegt.

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

DAG mit der neuen Speicherplatzbegrenzung testen

Triggern Sie den long_running_create_large_txt_file_print_logs-DAG noch einmal und warten Sie 1 Stunde und 15 Minuten, bis er fertig ist.

1. In den Ausgabeprotokollen der DAG-Ausführung wird Marking task as SUCCESS angezeigt und der Status der Aufgabe ist Erfolg mit einer Dauer von 1 Stunde und 15 Minuten, was der im DAG-Code festgelegten Wartezeit entspricht.

2. Prüfen Sie auf dem Tab Monitoring im Bereich Umgebungsübersicht, ob es keine roten Bereiche gibt.

3. Klicken Sie auf den Bereich Worker (Worker) und suchen Sie das Diagramm Gesamte Laufwerknutzung der Worker. Sie sehen, dass die Linie Datenträgerlimit die Änderung des Speicherlimits widerspiegelt und dass die Linie Datenträgernutzung im zulässigen Bereich liegt.

Zusammenfassung

In dieser Anleitung haben Sie den Grund für einen DAG-Fehler ermittelt und die Art der Ressource identifiziert, die zu einer Überlastung führt. Dazu haben Sie zwei Beispiel-DAGs fehlerbehoben, die aufgrund von unzureichendem Arbeitsspeicher und Speicherplatz fehlgeschlagen sind. Nachdem Sie Ihren Workern mehr Arbeitsspeicher und Speicherplatz zugewiesen hatten, konnten Sie die DAGs erfolgreich ausführen. Es wird jedoch empfohlen, Ihre DAGs (Workflows) zu optimieren, um den Ressourcenverbrauch von Workern von vornherein zu reduzieren, da die Ressourcen nicht über einen bestimmten Grenzwert hinaus erhöht werden können.

Bereinigen

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

Projekt löschen

Achtung: Das Löschen von Projekten hat folgende Auswirkungen:

• Alle Inhalte des Projekts werden gelöscht. Wenn Sie für die Aufgaben in diesem Dokument ein bereits bestehendes Projekt verwendet haben und dieses löschen, werden auch alle anderen im Rahmen des Projekts erstellten Daten gelöscht.
• Benutzerdefinierte Projekt-IDs gehen verloren. Beim Erstellen dieses Projekts haben Sie möglicherweise eine benutzerdefinierte Projekt-ID erstellt, die Sie weiterhin verwenden möchten. Damit die URLs, die die Projekt-ID nutzen, z. B. eine appspot.com-URL, erhalten bleiben, sollten Sie ausgewählte Ressourcen innerhalb des Projekts löschen, statt das gesamte Projekt.

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

1. In the Google Cloud console, go to the Manage resources page.

   Go to Manage resources

2. In the project list, select the project that you want to delete, and then click Delete.
3. In the dialog, type the project ID, and then click Shut down to delete the project.

Einzelne Ressourcen löschen

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

Löschen Sie die Cloud Composer-Umgebung. Außerdem wird dabei der Bucket der Umgebung gelöscht.

Nächste Schritte

• Umgebungen skalieren
• Fehlerbehebung bei DAGs