Fehlerbehebung bei langsamen oder hängenden Streamingjobs

Auf dieser Seite wird erläutert, wie Sie häufige Ursachen von langsamen oder hängenden Dataflow-Streamingjobs beheben können.

Wenn Sie die folgenden Symptome bemerken, wird Ihr Dataflow-Streamingjob möglicherweise langsam ausgeführt oder ist hängen geblieben:

  • Die Pipeline liest keine Daten aus der Quelle. Pub/Sub hat beispielsweise einen wachsenden Rückstand.
  • Die Pipeline schreibt keine Daten in die Senke.
  • Der Datenaktualitätsmesswert nimmt zu.
  • Der Systemlatenzmesswert nimmt zu.

Verwenden Sie die Informationen in den folgenden Abschnitten, um das Problem zu identifizieren und zu diagnostizieren.

Ursache ermitteln

  1. Prüfen Sie die Messwerte für Datenaktualität und Rückstand in Byte.

    • Wenn beide Messwerte monoton ansteigen, ist die Pipeline hängen geblieben und kommt nicht voran.
    • Wenn die Datenaktualität zunimmt, der Rückstand in Byte aber normal bleibt, sind ein oder mehrere Arbeitselemente in der Pipeline hängen geblieben.

    Suchen Sie nach den Phasen, in denen diese Messwerte ansteigen, um Phasen mit Problemen und die in diesen Phasen ausgeführten Vorgänge zu identifizieren.

  2. Prüfen Sie im Diagramm Parallelverarbeitung ob eine Phase aufgrund übermäßiger oder unzureichender Parallelität hängen geblieben ist. Weitere Informationen finden Sie unter Fehlerbehebung bei Parallelität.

  3. Prüfen Sie die Joblogs auf Probleme wie Kontingentlimits, Probleme mit der Verfügbarkeit oder erschöpfte IP-Adressen.

  4. Prüfen Sie die Worker-Logs auf Warnungen und Fehler.

    • Wenn die Worker-Logs Fehler enthalten, sehen Sie sich den Stacktrace an. Prüfen Sie, ob der Fehler durch einen Bug in Ihrem Code verursacht wird.
    • Suchen Sie nach Dataflow-Fehlern. Weitere Informationen finden Sie unter Dataflow-Fehler beheben.
    • Suchen Sie nach Fehlern, die darauf hinweisen, dass der Job ein Limit überschritten hat, z. B. die maximale Pub/Sub-Nachrichtengröße.
    • Suchen Sie nach Fehlern aufgrund von ungenügendem Arbeitsspeicher, die dazu führen können, dass die Pipeline hängen bleibt. Wenn Fehler aufgrund von ungenügendem Arbeitsspeicher auftreten, führen Sie die Schritte unter Fehlerbehebung bei Dataflow-Fehlern aufgrund von ungenügendem Arbeitsspeicheraus.
    • Prüfen Sie die Worker-Logs auf Operation ongoing-Meldungen, um einen langsamen oder hängen gebliebenen Schritt zu identifizieren. Sehen Sie sich den Stacktrace an, um zu sehen, wo der Schritt Zeit verbringt. Weitere Informationen finden Sie unter Verarbeitung hängen geblieben oder Vorgang läuft.

  5. Wenn ein Arbeitselement auf einem bestimmten Worker hängen bleibt, starten Sie die VM dieses Workers neu.

  6. Wenn Sie Streaming Engine nicht verwenden, prüfen Sie die Shuffler-Logs auf Warnungen und Fehler. Wenn ein RPC-Zeitüberschreitungsfehler an Port 12345 oder 12346 auftritt, fehlt Ihrem Job möglicherweise eine Firewallregel. Weitere Informationen finden Sie unter Firewallregeln für Dataflow.

  7. Prüfen Sie auf „heiße“ Schlüssel.

  8. Wenn der Portable Runner aktiviert ist, prüfen Sie die Harness-Logs auf Fehler. Weitere Informationen finden Sie unter Fehlerbehebung beim Portable Runner.

Wiederholte Fehler untersuchen

Bei einem Streamingjob werden bei einigen Fehlern ohne Begrenzung neue Versuche unternommen. Diese Wiederholungen verhindern den Fortschritt der Pipeline. Prüfen Sie die Worker-Logs auf Ausnahmen, um wiederholte Fehler zu identifizieren.

Fehlerhafte Worker ermitteln

Wenn die Worker, die den Streamingjob verarbeiten, fehlerhaft sind, ist der Job möglicherweise langsam oder scheint festzuhängen. So identifizieren Sie fehlerhafte Worker:

Nachzügler identifizieren

Ein Nachzügler ist ein Arbeitselement, das relativ zu anderen Arbeitselementen in der Phase langsam ist. Informationen zum Identifizieren und Korrigieren von Nachzüglern finden Sie unter Fehlerbehebung bei Nachzüglern in Streamingjobs.

Fehlerbehebung bei Parallelität

Aus Gründen der Skalierbarkeit und Effizienz führt Dataflow die Phasen Ihrer Pipeline parallel auf mehreren Workern aus. Die kleinste Einheit der parallelen Verarbeitung in Dataflow ist ein Schlüssel. Eingehende Nachrichten für jede zusammengeführte Phase sind einem Schlüssel zugeordnet. Der Schlüssel wird auf eine der folgenden Arten definiert:

  • Der Schlüssel wird implizit durch die Attribute der Quelle definiert, z. B. Kafka-Partitionen.
  • Der Schlüssel wird explizit durch Aggregationslogik in der Pipeline definiert, z. B. GroupByKey.

In Dataflow sind Worker-Threads für die Verarbeitung von Arbeitspaketen (Nachrichten) für einen Schlüssel zuständig. Die Anzahl der verfügbaren Threads zum Verarbeiten der Schlüssel des Jobs entspricht num_of_workers * threads_per_worker. Die Anzahl der Threads pro Worker wird anhand des SDK (Java, Python oder Go) und des Jobtyps (Batch oder Streaming) bestimmt.

Wenn die Pipeline für eine bestimmte Phase nicht genügend Schlüssel hat, begrenzt das die parallele Verarbeitung. Diese Phase kann zum Engpass werden.

Wenn die Pipeline eine sehr große Anzahl von Schlüsseln für eine bestimmte Phase verwendet, kann dies den Durchsatz der Phase begrenzen und zu Rückständen in den Upstream-Phasen führen, da pro Schlüssel ein gewisser Overhead entsteht. Der Overhead kann die Kommunikation des Back-Ends mit den Workern, externe RPCs zu einer Senke wie BigQuery und andere Verarbeitungsvorgänge umfassen. Wenn beispielsweise die Verarbeitung eines Schlüssels mit einer Nachricht 100 ms dauert, kann die Verarbeitung von 1.000 Nachrichten in diesem Schlüsselpaket auch etwa 100 ms dauern.

Phasen mit niedriger Parallelität identifizieren

Prüfen Sie die CPU-Auslastungsmesswerte, um festzustellen, ob die Langsamkeit der Pipeline durch eine niedrige Parallelität verursacht wird. Wenn die CPU-Auslastung niedrig, aber gleichmäßig auf die Worker verteilt ist, hat Ihr Job möglicherweise eine unzureichende Parallelität. Wenn Ihr Job Streaming Engine verwendet, sehen Sie sich auf dem Tab Jobmesswerte die Parallelitätsmesswerte an, um festzustellen, ob eine Phase eine niedrige Parallelität hat. So beheben Sie das Problem:

  • Prüfen Sie in der Google Cloud Console auf der Seite Jobinformationen den Tab Autoscaling, um zu sehen, ob Probleme beim Hochskalieren vorliegen. Wenn das Problem beim Autoscaling liegt, finden Sie weitere Informationen unter Fehlerbehebung beim Dataflow-Autoscaling.
  • Mit der Jobgrafik können Sie die Schritte in der Phase prüfen. Wenn die Phase aus einer Quelle liest oder in eine Senke schreibt, lesen Sie die Dokumentation für den Dienst der Quelle oder Senke. Ermitteln Sie anhand der Dokumentation, ob dieser Dienst für ausreichende Skalierbarkeit konfiguriert ist.

Phasen mit hoher Parallelität identifizieren

Eine Kombination aus niedriger Systemlatenz, steigender Datenaktualität und zunehmendem Rückstand sowie unterlasteten Worker-CPUs deutet darauf hin, dass die Pipeline aufgrund einer großen Anzahl von Schlüsseln gedrosselt wird. Prüfen Sie im Diagramm zur Parallelverarbeitung , ob es Phasen mit einer großen Anzahl von Schlüsseln gibt.

Transformationen wie Reshuffle können Millionen von Schlüsseln generieren, wenn Sie withNumBuckets nicht explizit angeben. Eine große Anzahl von Schlüsseln kann zur Erstellung zahlreicher kleinerer Arbeitspakete führen, für die jeweils ein eigener Arbeitsthread zur Verarbeitung erforderlich ist. Da die verfügbaren Worker-Threads begrenzt sind, kann dies zu einem erheblichen Rückstand an Verarbeitungsschlüsseln führen, was zu Verzögerungen führt, da sie auf Ressourcen warten. Daher werden die Worker-Threads nicht effizient genutzt.

Wir empfehlen, die Anzahl der Schlüssel zu begrenzen, indem Sie die Option withNumBuckets in der Transformation Reshuffle festlegen. Der Wert sollte die Gesamtzahl der Threads aller Worker nicht überschreiten. Die Ausrichtung auf (threads_per_worker * max_workers) Schlüssel in der Pipeline ist möglicherweise nicht optimal. Manchmal sind weniger Schlüssel und größere Pakete möglich, die von Dataflow aufgrund der Verwendung weniger Worker effizienter verarbeitet werden. Eine geringere Anzahl von Schlüsseln führt zu größeren Arbeitspaketen, wodurch die Worker-Threads effizient genutzt und der Durchsatz der Phase erhöht wird.

Wenn es mehrere Reshuffle-Schritte in der Pipeline gibt, teilen Sie die Gesamtzahl der Threads durch die Anzahl der Reshuffle-Schritte, um withNumBuckets zu berechnen.

Auf „heiße“ Schlüssel prüfen

Wenn Aufgaben ungleichmäßig auf Worker verteilt sind und die Worker-Auslastung sehr ungleichmäßig ist, hat Ihre Pipeline möglicherweise einen „heißen“ Schlüssel. Ein „heißer“ Schlüssel ist ein Schlüssel, der in den Daten überrepräsentiert ist.

Prüfen Sie, ob Ihre Pipeline „heiße“ Schlüssel hat, indem Sie den Engpassstatus für den langsamen Schritt ansehen. Möglicherweise haben Sie „heiße“ Schlüssel, wenn der Grund für den Engpass „Heiße Schlüssel oder unzureichende Schlüsselparallelität“ lautet. Allgemeine Informationen zu Engpässen finden Sie unter Fehlerbehebung bei Engpässen.

Standardmäßig protokolliert Dataflow nur das Vorhandensein eines „heißen“ Schlüssels, nicht den spezifischen Schlüsselwert. Wenn Sie den für Menschen lesbaren Schlüssel in den Logs ausgeben lassen möchten, wenn ein „heißer“ Schlüssel erkannt wird, legen Sie beim Ausführen der Pipeline die folgende Pipelineoption für die Fehlerbehebung fest:

Java

--hotKeyLoggingEnabled=true

Python

--dataflow_service_options=enable_hot_key_logging

Wenn Sie einen Flex-Vorlagenjob ausführen, finden Sie unter Pipelineoptionen angeben weitere Informationen zur Verwendung von --additional-pipeline-options zum Übergeben der Option.

Prüfen Sie dann mit dem folgenden Logfilter auf „heiße“ Schlüssel:

  resource.type="dataflow_step"
  resource.labels.job_id=JOB_ID
  jsonPayload.logger="org.apache.beam.runners.dataflow.worker.HotKeyLogger"

Ersetzen Sie JOB_ID durch die ID Ihres Jobs.

Sie können auch logbasierte Benachrichtigungen konfigurieren, um benachrichtigt zu werden, wenn in Ihrer Pipeline „heiße“ Schlüssel erkannt werden. Verwenden Sie den Beispielfilter als Start.

Führen Sie einen oder mehrere der folgenden Schritte aus, um dieses Problem zu beheben:

  • Daten nochmal zur Verfügung stellen. Wenden Sie eine ParDo-Transformation an, um neue Schlüssel/Wert-Paare auszugeben. Weitere Informationen finden Sie auf der Java ParDo Transformationsseite oder auf der Python ParDo Transformationsseite in der Apache Beam-Dokumentation.
  • Verwenden Sie .withFanout in Combine-Transformationen. Weitere Informationen finden Sie in der Combine.PerKey Klasse im Java SDK oder im with_hot_key_fanout Vorgang im Python SDK.
  • Wenn Sie eine Java-Pipeline haben, die unbegrenzte PCollections mit hohem Volumen verarbeitet, empfehlen wir Folgendes:
    • Verwenden Sie Combine.Globally.withFanout anstelle von Combine.Globally.
    • Verwenden Sie Combine.PerKey.withHotKeyFanout anstelle von Count.PerKey.

Auf unzureichendes Kontingent prüfen

Achten Sie darauf, dass Ihr Kontingent für die Quelle und die Senke ausreicht. Wenn Ihre Pipeline beispielsweise Eingaben aus Pub/Sub oder BigQuery liest, hat Ihr Google Cloud Projekt möglicherweise nicht genügend Kontingent. Weitere Informationen zu Kontingentlimits für diese Dienste finden Sie unter Pub/Sub-Kontingente oder BigQuery-Kontingente.

Wenn der Job eine hohe Anzahl von Fehlern des Typs 429 (Rate Limit Exceeded) generiert, ist das Kontingent möglicherweise nicht ausreichend. Führen Sie die folgenden Schritte aus, um auf Fehler zu prüfen:

  1. Rufen Sie die Google Cloud Console auf.
  2. Klicken Sie im Navigationsbereich auf APIs &Dienste.
  3. Klicken Sie im Menü auf Bibliothek.
  4. Geben Sie in das Suchfeld Pub/Sub ein.
  5. Klicken Sie auf Cloud Pub/Sub API.
  6. Klicken Sie auf Verwalten.
  7. Suchen Sie im Diagramm Traffic nach Antwortcode nach (4xx)-Clientfehlercodes.

Sie können auch Metrics Explorer verwenden, um die Kontingentnutzung zu prüfen. Wenn Ihre Pipeline eine BigQuery-Quelle oder -Senke verwendet, verwenden Sie die BigQuery Storage API-Messwerte, um Kontingentprobleme zu beheben. So erstellen Sie beispielsweise ein Diagramm mit der Anzahl gleichzeitiger BigQuery-Verbindungen:

  1. Wählen Sie in der Google Cloud Console Monitoring aus:

    Zu Monitoring

  2. Wählen Sie im Navigationsbereich Metrics Explorer aus.

  3. Filtern Sie im Bereich Messwert auswählen für Messwert zu BigQuery-Projekt >Schreiben >Anzahl gleichzeitiger Verbindungen.

Eine Anleitung zum Aufrufen von Pub/Sub-Messwerten finden Sie unter Kontingentnutzung überwachen in "Pub/Sub in Cloud Monitoring überwachen". Eine Anleitung zum Aufrufen von BigQuery-Messwerten finden Sie unter Kontingentnutzung und -limits aufrufen in "Dashboards, Diagramme und Benachrichtigungen erstellen".

Tools zur Fehlerbehebung

Wenn Sie eine langsame oder hängende Pipeline haben, können Sie mit den folgenden Tools versuchen, das Problem zu diagnostizieren.

  • Verwenden Sie Cloud Monitoring für Dataflow, um Vorfälle zu korrelieren und Engpässe zu identifizieren.
  • Verwenden Sie zum Überwachen der Pipelineleistung Cloud Profiler.
  • Einige Transformationen eignen sich besser für Pipelines mit hohem Volumen als andere. Lognachrichten können eine hängen gebliebene Nutzertransformation in Batch- oder Streamingpipelines identifizieren.
  • Weitere Informationen zu hängen gebliebenen Jobs finden Sie unter Dataflow-Jobmesswerte. Die folgende Liste enthält nützliche Messwerte:
    • Der Messwert Rückstand in Byte (backlog_bytes) misst die Menge der nicht verarbeiteten Eingabe in Byte pro Phase. Verwenden Sie diesen Messwert, um einen zusammengeführten Schritt zu finden, der keinen Durchsatz hat. Der Messwert der Rückstandselemente (backlog_elements) misst die Anzahl der nicht verarbeiteten Eingabeelemente für eine Phase.
    • Der Messwert Verarbeitungsparallelitätsschlüssel (processing_parallelism_keys) misst die Anzahl der parallelen Verarbeitungsschlüssel für eine bestimmte Phase der Pipeline in den letzten fünf Minuten. Verwenden Sie diesen Messwert, um Untersuchungen anzustellen:
      • Sie können das Problem auf bestimmte Phasen eingrenzen und Warnungen zu „heißen“ Schlüsseln wie A hot key ... was detected bestätigen.
      • Suchen Sie nach Durchsatzengpässen aufgrund unzureichender Parallelität. Diese Engpässe können zu langsamen oder hängenden Pipelines führen.
    • Der Messwert Systemverzögerung (system_lag) und der Messwert für die verzögerte Systemverzögerung pro Phase (per_stage_system_lag) messen die maximale Zeitdauer, für die ein Datenelement verarbeitet wurde oder auf die Verarbeitung gewartet hat. Verwenden Sie diese Messwerte, um ineffiziente Phasen und Engpässe aus Datenquellen zu ermitteln.

Weitere Messwerte, die nicht in der Dataflow -Monitoring-Weboberfläche enthalten sind, finden Sie in der vollständigen Liste der Dataflow-Messwerte unter Google Cloud Messwerte.