Résoudre les problèmes liés aux jobs de traitement en flux continu lents ou bloqués

Cette page explique comment résoudre les causes courantes de jobs Dataflow par flux lentes ou bloquées.

Si vous remarquez les symptômes suivants, votre job Dataflow par flux peut s'exécuter lentement ou être bloqué :

Utilisez les informations fournies dans les sections suivantes pour identifier et diagnostiquer le problème.

Identifier la cause principale

  1. Vérifiez les métriques de fraîcheur des données et d'octets en attente.

    • Si les deux métriques augmentent de façon monotone, cela signifie que le pipeline est bloqué et ne progresse pas.
    • Si la fraîcheur des données augmente, mais que les octets en attente restent normaux, cela signifie qu'un ou plusieurs éléments de travail sont bloqués dans le pipeline.

    Recherchez les étapes où ces métriques augmentent pour identifier les étapes problématiques et les opérations effectuées dans ces étapes.

  2. Consultez le graphique de traitement en parallèle pour voir si une étape est bloquée en raison d'un parallélisme excessif ou insuffisant. Consultez la section Résoudre les problèmes de parallélisme.

  3. Consultez les journaux de job pour détecter des problèmes tels que des limites de quota, des problèmes de rupture de stock ou un épuisement des adresses IP.

  4. Consultez les journaux des nœuds de calcul pour détecter les avertissements et les erreurs.

    • Si les journaux des nœuds de calcul contiennent des erreurs, affichez la trace de la pile. Déterminez si l'erreur est due à un bug dans votre code.
    • Recherchez les erreurs Dataflow. Consultez la page Résoudre les erreurs Dataflow.
    • Recherchez les erreurs indiquant que le job a dépassé une limite, par exemple la taille maximale des messages Pub/Sub.
    • Recherchez les erreurs de mémoire insuffisante, qui peuvent entraîner le blocage d'un pipeline. Si vous voyez des erreurs de mémoire insuffisante, suivez les étapes décrites dans la section Résoudre les erreurs Dataflow de mémoire insuffisante.
    • Pour identifier une étape lente ou bloquée, recherchez les messages Operation ongoing dans les journaux des nœuds de calcul. Affichez la trace de la pile pour voir où l'étape passe du temps. Pour en savoir plus, consultez la section Traitement bloqué ou opération en cours.

  5. Si un élément de travail est bloqué sur un nœud de calcul spécifique, redémarrez la VM de ce nœud de calcul.

  6. Si vous n'utilisez pas Streaming Engine, consultez les journaux du shuffler pour détecter les avertissements et les erreurs. Si vous voyez une erreur de délai d'attente RPC sur le port 12345 ou 12346, il est possible qu'il manque une règle de pare-feu à votre job. Consultez la section Règles de pare-feu pour Dataflow.

  7. Recherchez les clés chaudes.

  8. Si Portable Runner est activé, recherchez les erreurs dans les journaux du harness. Pour en savoir plus, consultez la section Résoudre les problèmes liés à Portable Runner.

Examiner les échecs répétés

Dans un job par flux, certains échecs sont relancés indéfiniment. Ces nouvelles tentatives empêchent le pipeline de progresser. Pour identifier les échecs répétés, consultez les exceptions dans les journaux des nœuds de calcul.

  • Si l'exception concerne le code utilisateur, déboguez et corrigez le problème dans le code ou dans les données.
  • Pour éviter que des défaillances inattendues ne bloquent l'exécution de votre pipeline, mettez en œuvre une file d'attente de lettres mortes. Pour un exemple de mise en œuvre, reportez-vous à la page Modèles BigQuery dans la documentation Apache Beam.
  • Si l'exception est une erreur de mémoire insuffisante (OOM), consultez la page Résoudre les erreurs de mémoire insuffisante Dataflow.
  • Pour les autres exceptions, consultez la page Résoudre les erreurs Dataflow.

Identifier les nœuds de calcul non opérationnels

Si les nœuds de calcul qui traitent le job par flux ne sont pas opérationnels, le job peut être lent ou sembler bloqué. Pour identifier les nœuds de calcul non opérationnels :

Identifier les retardataires

Un retardataire est un élément de travail lent par rapport aux autres éléments de travail de l'étape. Pour en savoir plus sur l'identification et la correction des retardataires, consultez la page Résoudre les problèmes de retardataires dans les jobs par flux.

Résoudre les problèmes de parallélisme

Pour plus d'évolutivité et d'efficacité, Dataflow exécute les étapes de votre pipeline en parallèle sur plusieurs nœuds de calcul. La plus petite unité de traitement parallèle dans Dataflow est une clé. Les messages entrants pour chaque étape fusionnée sont associés à une clé. La clé est définie de l'une des manières suivantes :

  • La clé est implicitement définie par les propriétés de la source, (par exemple, les partitions Kafka).
  • La clé est explicitement définie par la logique d'agrégation dans le pipeline (par exemple, GroupByKey).

Dans Dataflow, les threads de nœud de calcul sont responsables du traitement des groupes de travail (messages) pour une clé. Le nombre de threads disponibles pour traiter les clés du job est égal à num_of_workers * threads_per_worker. Le nombre de threads par nœud de calcul est déterminé en fonction du SDK (Java, Python ou Go) et du type de mission (par lot ou par flux).

Si le pipeline ne dispose pas de suffisamment de clés pour une étape donnée, il limite le traitement en parallèle. Cette étape peut devenir un goulot d'étranglement.

Si le pipeline utilise un très grand nombre de clés pour une étape donnée, il peut limiter le débit de l'étape et accumuler des tâches en attente dans les étapes en amont, car il existe une surcharge par clé. La surcharge peut inclure la communication backend avec les nœuds de calcul, les RPC externes vers un récepteur tel que BigQuery et d'autres traitements. Par exemple, si le traitement d'une clé avec un message prend 100 ms, le traitement de 1 000 messages dans ce groupe de clés peut également prendre environ 100 ms.

Identifier les étapes avec un faible parallélisme

Pour déterminer si le ralentissement du pipeline est causé par un parallélisme faible, consultez les métriques d'utilisation du processeur. Si l'utilisation est faible, mais répartie uniformément entre les nœuds de calcul, votre job peut présenter un parallélisme insuffisant. Si votre job utilise Streaming Engine, pour voir si une étape présente un parallélisme faible, consultez les métriques de parallélisme dans l'onglet Métriques de job. Pour résoudre ce problème, procédez comme suit :

  • Dans la Google Cloud console, sur la page Informations sur le job, utilisez l' onglet Autoscaling pour voir si le job rencontre des problèmes pour effectuer un scaling à la hausse. Si l'autoscaling est le problème, consultez la page Résoudre les problèmes liés à l'autoscaling Dataflow.
  • Utilisez le graphique de job pour vérifier les étapes de la phase. Si l'étape lit depuis une source ou écrit dans un récepteur, consultez la documentation du service de la source ou du récepteur. Consultez la documentation pour déterminer si ce service est configuré pour une évolutivité suffisante.

Identifier les étapes avec un parallélisme élevé

Une combinaison de faible latence du système, de fraîcheur croissante des données, d'augmentation des tâches en attente et de sous-utilisation des processeurs des nœuds de calcul suggère que le pipeline est limité en raison d'un grand nombre de clés. Consultez le graphique de traitement en parallèle pour identifier les étapes comportant un grand nombre de clés.

Les transformations telles que Reshuffle peuvent générer des millions de clés si vous ne spécifiez pas explicitement withNumBuckets. Un grand nombre de clés peut entraîner la création de nombreux groupes de travail plus petits, chacun nécessitant un thread de travail dédié pour le traitement. Étant donné que le nombre de threads de nœud de calcul disponibles est limité, cela peut entraîner un nombre important de clés de traitement en attente, ce qui provoque des retards en attendant les ressources. Par conséquent, les threads de nœud de calcul ne sont pas utilisés efficacement.

Nous vous recommandons de limiter le nombre de clés en définissant l'option withNumBuckets dans la transformation Reshuffle. La valeur ne doit pas dépasser le nombre total de threads sur tous les nœuds de calcul. Cibler (threads_per_worker * max_workers) clés dans le pipeline n'est peut-être pas optimal. Parfois, moins de clés et des groupes plus volumineux sont possibles et sont traités plus efficacement par Dataflow, car moins de nœuds de calcul sont utilisés. Un nombre plus petit de clés crée des groupes de travail plus volumineux, ce qui utilise efficacement les threads de nœud de calcul et augmente le débit de l'étape.

S'il existe plusieurs étapes Reshuffle dans le pipeline, divisez le nombre total de threads par le nombre d'étapes Reshuffle pour calculer withNumBuckets.

Rechercher les clés chaudes

Si les tâches sont réparties de manière inégale entre les nœuds de calcul et que l'utilisation est très inégale, votre pipeline peut avoir une clé chaude. Une clé chaude est une clé surreprésentée dans les données.

Vérifiez si votre pipeline comporte des clés chaudes en examinant l'état du goulot d'étranglement pour l'étape lente. Vous pouvez avoir des clés chaudes si la raison du goulot d'étranglement est "Clés chaudes ou parallélisme de clés insuffisant". Pour obtenir des informations générales sur les goulots d'étranglement, consultez la section Résoudre les problèmes de goulots d'étranglement.

Par défaut, Dataflow n'enregistre que la présence d'une clé chaude, et non la valeur de clé spécifique. Pour imprimer la clé lisible dans les journaux lorsqu'une clé chaude est détectée, définissez l'option de pipeline de débogage suivante lorsque vous exécutez votre pipeline :

Java

--hotKeyLoggingEnabled=true

Python

--dataflow_service_options=enable_hot_key_logging

Si vous exécutez un job de modèle Flex, consultez la section Spécifier les options de pipeline pour en savoir plus sur l'utilisation de --additional-pipeline-options afin de transmettre l'option.

Ensuite, recherchez les clés chaudes à l'aide du filtre de journal suivant :

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

Remplacez JOB_ID par l'ID de votre job.

Vous pouvez également configurer des alertes basées sur les journaux pour recevoir une notification lorsque des clés chaudes sont détectées dans votre pipeline. Utilisez l'exemple de filtre comme point de départ.

Pour résoudre ce problème, appliquez une ou plusieurs des étapes suivantes :

  • Saisissez de nouveau vos données. Pour générer de nouvelles paires clé/valeur, appliquez une transformation ParDo. Pour en savoir plus, consultez la page de la transformation Java ParDo ou la Python ParDo dans le Documentation Apache Beam.
  • Utilisez .withFanout dans vos transformations combinées. Pour en savoir plus, consultez la Combine.PerKey classe dans le SDK Java, ou l' with_hot_key_fanout opération dans le SDK Python.
  • Si vous disposez d'un pipeline Java qui traite des PCollections illimitées contenant de grands volumes de données, nous vous recommandons d'effectuer les opérations suivantes :
    • Utilisez Combine.Globally.withFanout à la place de Combine.Globally.
    • Utilisez Combine.PerKey.withHotKeyFanout à la place de Count.PerKey.

Vérifier si le quota est insuffisant

Assurez-vous de disposer d'un quota suffisant pour votre source et votre récepteur. Par exemple, si votre pipeline lit des entrées provenant de Pub/Sub ou de BigQuery, votre Google Cloud projet présente un quota insuffisant. Pour en savoir plus sur les limites de quota pour ces services, consultez les pages Quota Pub/Sub ou Quota BigQuery.

Si votre job génère un nombre élevé d'erreurs 429 (Rate Limit Exceeded), son quota peut être insuffisant. Pour rechercher les erreurs, procédez comme suit :

  1. Accédez à la Google Cloud console.
  2. Dans le volet de navigation, cliquez sur API et services.
  3. Dans le menu, cliquez sur Bibliothèque.
  4. Utilisez le champ de recherche pour rechercher Pub/Sub.
  5. Cliquez sur API Cloud Pub/Sub.
  6. Cliquez sur Gérer.
  7. Dans le graphique Trafic par code de réponse, recherchez des codes d'erreur client (4xx).

Vous pouvez également vérifier l'utilisation des quotas à l'aide de l'explorateur de métriques. Si votre pipeline utilise une source ou un récepteur BigQuery, vous pouvez utiliser les métriques de l'API BigQuery Storage pour résoudre les problèmes de quota. Par exemple, pour créer un graphique indiquant le nombre de connexions BigQuery simultanées, procédez comme suit :

  1. Dans la Google Cloud console, sélectionnez Monitoring:

    Accéder à Monitoring

  2. Dans le volet de navigation, sélectionnez Explorateur de métriques.

  3. Dans le volet Sélectionner une métrique, pour Métrique, filtrez sur Projet BigQuery > Écriture > Nombre de connexions simultanées.

Pour obtenir des instructions sur l'affichage des métriques Pub/Sub, consultez la section Surveiller l'utilisation des quotas dans "Surveiller Pub/Sub dans Cloud Monitoring". Pour obtenir des instructions sur l'affichage des métriques BigQuery, consultez la section Afficher l'utilisation et les limites de quota de la page "Créer des tableaux de bord, des graphiques et des alertes".

Outils de débogage

Lorsque votre pipeline est lent ou bloqué, les outils suivants peuvent vous aider à diagnostiquer le problème.

  • Pour mettre en corrélation des incidents et identifier les goulots d'étranglement, utilisez Cloud Monitoring pour Dataflow.
  • Pour surveiller les performances du pipeline, utilisez Cloud Profiler.
  • Certaines transformations sont mieux adaptées que d'autres aux pipelines traitant de gros volumes. Les messages de journal peuvent identifier une transformation utilisateur bloquée dans des pipelines de traitement par lot ou par flux.
  • Pour en savoir plus sur un job bloqué, utilisez les métriques de job Dataflow. La liste suivante inclut des métriques utiles :
    • La métrique Octets en attente (backlog_bytes) mesure la quantité d'entrées non traitées, exprimée en octets par étape. Utilisez cette métrique pour identifier une étape fusionnée qui n'a pas de débit. De même, la métrique d'éléments en attente (backlog_elements) mesure le nombre d'éléments d'entrée non traités pour une étape.
    • La métrique Traiter des clés en parallèle (processing_parallelism_keys) mesure le nombre de clés de traitement en parallèle par étape spécifique du pipeline au cours des cinq dernières minutes. Utilisez cette métrique pour examiner les éléments suivants :
      • Filtrez le problème sur des étapes spécifiques et confirmez les avertissements de clé chaude, tels que A hot key ... was detected.
      • Identifiez les goulots d'étranglement de débit causés par un parallélisme insuffisant. Ces goulots d'étranglement peuvent ralentir les pipelines ou les bloquer.
    • La métrique Retard du système (system_lag) et la métrique de latence du système par étape (per_stage_system_lag) mesurent la durée maximale de traitement ou d'attente de traitement d'un élément de données. Utilisez ces métriques pour identifier les étapes et les goulots d'étranglement inefficaces à partir de sources de données.

Pour obtenir d'autres métriques qui ne sont pas incluses dans l'interface Web de surveillance Dataflow, consultez la liste complète des métriques Dataflow dans Google Cloud la section Métriques.