Régler l'autoscaling horizontal pour les pipelines de traitement par flux

Dans les pipelines de traitement en flux continu avec un volume élevé de données d'entrée, il existe généralement un compromis entre coût et latence. Pour conserver une latence faible, Dataflow doit ajouter des nœuds de calcul à mesure que le volume de trafic augmente. Un autre facteur est la vitesse à laquelle le pipeline doit effectuer un scaling à la hausse ou à la baisse en réponse aux modifications du débit de données d'entrée.

L'autoscaler Dataflow dispose de paramètres par défaut adaptés à de nombreuses charges de travail. Toutefois, vous souhaiterez peut-être ajuster ce comportement en fonction de votre scénario. Par exemple, une latence moyenne plus élevée peut être acceptable pour réduire les coûts, mais vous pouvez également souhaiter que Dataflow effectue un scaling à la hausse plus rapidement afin de répondre aux pics de trafic.

Pour optimiser l'autoscaling horizontal, vous pouvez ajuster les paramètres suivants :

Définir la plage d'autoscaling

Lorsque vous créez un job de traitement en flux continu, vous pouvez définir le nombre initial de nœuds de calcul et le nombre maximal de nœuds de calcul. Pour ce faire, spécifiez les options de pipeline suivantes :

Java

  • --numWorkers : nombre initial de nœuds de calcul disponibles au début de l'exécution du pipeline
  • --maxNumWorkers : nombre maximal de nœuds de calcul disponibles pour votre pipeline

Python

  • --num_workers : nombre initial de nœuds de calcul disponibles au début de l'exécution du pipeline
  • --max_num_workers : nombre maximal de nœuds de calcul disponibles pour votre pipeline

Accéder

  • --num_workers : nombre initial de nœuds de calcul disponibles au début de l'exécution du pipeline
  • --max_num_workers : nombre maximal de nœuds de calcul disponibles pour votre pipeline

Pour les jobs de traitement en flux continu qui utilisent Streaming Engine, l'option --maxNumWorkers est facultative. La valeur par défaut est 100. Pour les jobs de traitement en flux continu qui n'utilisent pas Streaming Engine, --maxNumWorkers est requis lorsque l'autoscaling horizontal est activé.

La valeur de départ de --maxNumWorkers détermine également le nombre de disques persistants alloués au job. Les pipelines sont déployés avec un pool fixe de disques persistants, d'un nombre égal à --maxNumWorkers. Lors du streaming, les disques persistants sont redistribués de manière à ce que chaque nœud de calcul ait un nombre égal de disques associés.

Si vous définissez --maxNumWorkers, assurez-vous que la valeur fournit suffisamment de disques pour votre pipeline. Tenez compte de la croissance future lorsque vous définissez la valeur initiale. Pour en savoir plus sur les performances de Persistent Disk, consultez la section Configurer Persistent Disk et vos VM. Dataflow facture l'utilisation de Persistent Disk et dispose de quotas Compute Engine, y compris de quotas Persistent Disk.

Par défaut, le nombre minimal de nœuds de calcul est de 1 pour les jobs de traitement en flux continu qui utilisent Streaming Engine et de (maxNumWorkers/15) arrondi pour les jobs qui n'utilisent pas Streaming Engine.

Mettre à jour la plage d'autoscaling

Pour les jobs qui utilisent Streaming Engine, vous pouvez ajuster le nombre minimal et maximal de nœuds de calcul, sans arrêter ni remplacer le job. Pour ajuster ces valeurs, utilisez une mise à jour de job en cours. Mettez à jour les options de tâches suivantes :

  • --min-num-workers : nombre minimal de nœuds de calcul.

  • --max-num-workers : nombre maximal de nœuds de calcul.

gcloud

Exécutez la commande gcloud dataflow jobs update-options :

gcloud dataflow jobs update-options \
  --region=REGION \
  --min-num-workers=MINIMUM_WORKERS \
  --max-num-workers=MAXIMUM_WORKERS \
  JOB_ID

Remplacez les éléments suivants :

  • REGION : ID de région du point de terminaison régional de la tâche.
  • MINIMUM_WORKERS : nombre minimal d'instances Compute Engine.
  • MAXIMUM_WORKERS : nombre maximal d'instances Compute Engine.
  • JOB_ID : ID de la tâche à mettre à jour.

Vous pouvez également mettre à jour --min-num-workers et --max-num-workers individuellement.

REST

Utilisez la méthode projects.locations.jobs.update :

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?updateMask=runtime_updatable_params.max_num_workers,runtime_updatable_params.min_num_workers
{
  "runtime_updatable_params": {
    "min_num_workers": MINIMUM_WORKERS,
    "max_num_workers": MAXIMUM_WORKERS
  }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du Google Cloud projet du job Dataflow
  • REGION : identifiant de la région du point de terminaison régional de la tâche.
  • JOB_ID : ID de la tâche à mettre à jour.
  • MINIMUM_WORKERS : nombre minimal d'instances Compute Engine.
  • MAXIMUM_WORKERS : nombre maximal d'instances Compute Engine.

Vous pouvez également mettre à jour min_num_workers et max_num_workers individuellement. Spécifiez les paramètres à mettre à jour dans le paramètre de requête updateMask, puis incluez les valeurs mises à jour dans le champ runtimeUpdatableParams du corps de la requête. L'exemple suivant met à jour min_num_workers :

PUT https://dataflow.googleapis.com/v1b3/projects/my_project/locations/us-central1/jobs/job1?updateMask=runtime_updatable_params.min_num_workers
{
  "runtime_updatable_params": {
    "min_num_workers": 5
  }
}

Pour les jobs qui n'utilisent pas Streaming Engine, vous pouvez remplacer le job existant par une valeur mise à jour de maxNumWorkers.

Si vous mettez à jour un job de traitement en flux continu qui n'utilise pas Streaming Engine, l'autoscaling horizontal est désactivé par défaut sur le job mis à jour. Pour maintenir l'autoscaling activé, spécifiez --autoscalingAlgorithm et --maxNumWorkers pour le job mis à jour.

Définir l'optimisation de l'utilisation des nœuds de calcul

Dataflow utilise l'utilisation moyenne du processeur comme signal pour déterminer quand appliquer l'autoscaling horizontal. Par défaut, Dataflow définit un objectif d'utilisation du processeur de 0,8. Lorsque l'utilisation se situe en dehors de cette plage, Dataflow peut ajouter ou supprimer des nœuds de calcul.

Pour un contrôle plus précis du comportement de l'autoscaling, vous pouvez définir l'objectif d'utilisation du processeur sur une valeur comprise dans la plage [0,1 - 0,9].

  • Définissez une valeur d'utilisation du processeur inférieure si vous souhaitez atteindre des latences maximales plus faibles. Une valeur inférieure permet à Dataflow d'effectuer un scaling horizontal plus agressif en réponse à une utilisation croissante des nœuds de calcul, et de procéder à un scaling à la baisse plus prudent pour améliorer la stabilité. Une valeur inférieure offre également plus de marge de manœuvre lorsque le pipeline affiche un état stable, ce qui entraîne généralement une latence de queue plus faible. (La latence de queue mesure les temps d'attente les plus longs avant le traitement d'un nouvel enregistrement.)

  • Définissez une valeur plus élevée si vous souhaitez économiser les ressources et réduire les coûts lors des pics de trafic. Une valeur plus élevée évite un scaling à la hausse excessif, au prix d'une latence plus élevée.

Pour configurer l'optimisation de l'utilisation lorsque vous exécutez une tâche sans modèle, définissez l'option de service worker_utilization_hint. Pour une tâche de modèle, modifiez plutôt l'optimisation de l'utilisation, car les options de service ne sont pas prises en charge.

L'exemple suivant montre comment utiliser worker_utilization_hint :

Java

--dataflowServiceOptions=worker_utilization_hint=TARGET_UTILIZATION

Remplacez TARGET_UTILIZATION par une valeur comprise dans la plage [0.1 - 0.9].

Python

--dataflow_service_options=worker_utilization_hint=TARGET_UTILIZATION

Remplacez TARGET_UTILIZATION par une valeur comprise dans la plage [0.1 - 0.9].

Accéder

--dataflow_service_options=worker_utilization_hint=TARGET_UTILIZATION

Remplacez TARGET_UTILIZATION par une valeur comprise dans la plage [0.1 - 0.9].

Pour les nouveaux pipelines, nous vous recommandons d'effectuer les tests sous des charges réalistes en utilisant les paramètres par défaut. Évaluez ensuite le comportement de l'autoscaling tel qu'il s'applique à votre pipeline et apportez les ajustements nécessaires.

L'optimisation de l'utilisation n'est qu'un des facteurs que Dataflow prend en compte pour déterminer s'il convient d'effectuer le scaling des nœuds de calcul. D'autres facteurs, tels que le backlog et les clés disponibles, peuvent remplacer la valeur d'optimisation. De plus, l'optimisation n'est pas une cible stricte. L'autoscaler tente de maintenir l'utilisation du processeur dans la plage de la valeur d'optimisation, mais la métrique d'utilisation agrégée peut être supérieure ou inférieure. Pour en savoir plus, consultez la section Heuristique d'autoscaling de flux.

Mettre à jour l'optimisation de l'utilisation

Pour mettre à jour l'optimisation de l'utilisation pendant l'exécution d'une tâche, effectuez une mise à jour en cours comme suit :

gcloud

Exécutez la commande gcloud dataflow jobs update-options :

gcloud dataflow jobs update-options \
  --region=REGION \
  --worker-utilization-hint=TARGET_UTILIZATION \
  JOB_ID

Remplacez les éléments suivants :

  • REGION : ID de région du point de terminaison régional de la tâche.
  • JOB_ID : ID de la tâche à mettre à jour.
  • TARGET_UTILIZATION : valeur comprise dans la plage [0,1 - 0,9].

Pour rétablir la valeur par défaut de l'optimisation d'utilisation, exécutez la commande gcloud suivante :

gcloud dataflow jobs update-options \
  --unset-worker-utilization-hint \
  --region=REGION \
  --project=PROJECT_ID \
  JOB_ID

REST

Utilisez la méthode projects.locations.jobs.update :

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?updateMask=runtime_updatable_params.worker_utilization_hint
{
  "runtime_updatable_params": {
    "worker_utilization_hint": TARGET_UTILIZATION
  }
}

Remplacez les éléments suivants :

  • PROJECT_ID : Google Cloud ID du projet du job Dataflow.
  • REGION : identifiant de la région du point de terminaison régional de la tâche.
  • JOB_ID : ID de la tâche à mettre à jour.
  • TARGET_UTILIZATION : valeur comprise dans la plage [0,1 - 0,9].

Définir l'optimisation du parallélisme des nœuds de calcul

Pour gérer l'autoscaling avec des opérations longues qui dépendent moins des processeurs, telles que les charges de travail nécessitant une utilisation intensive du ML, vous pouvez définir l'optimisation du parallélisme des nœuds de calcul à l'aide des optimisations de ressources Apache Beam. Ces optimisations permettent de passer l'autoscaling à un autre mode optimisé pour les charges de travail nécessitant une utilisation intensive du GPU ou les transformations avec un long temps de traitement.

L'exemple suivant montre comment associer une optimisation du parallélisme à une transformation :

Java

pcoll.apply(MyCompositeTransform.of(...)
  .setResourceHints(
      ResourceHints.create()
          .withMaxActiveBundlesPerWorker(TARGET_PARALLELISM_PER_WORKER)))

Remplacez TARGET_PARALLELISM_PER_WORKER par une valeur adaptée à votre cas d'utilisation. Pour obtenir des conseils généraux, consultez la section comment choisir une bonne valeur de départ.

Python

pcoll | MyPTransform().with_resource_hints(
  max_active_bundles_per_worker="TARGET_PARALLELISM_PER_WORKER")

Remplacez TARGET_PARALLELISM_PER_WORKER par une valeur adaptée à votre cas d'utilisation. Pour obtenir des conseils généraux, consultez la section comment choisir une bonne valeur de départ.

Choisir la valeur d'optimisation du parallélisme des nœuds de calcul

Pour les cas d'utilisation du ML, une bonne valeur de départ est équivalente au nombre de modèles exécutés en parallèle dans chaque nœud de calcul. Cette valeur est limitée par la capacité des accélérateurs sur le nœud de calcul et la taille du modèle.

Pour les autres cas d'utilisation, le pipeline est lié à la mémoire ou au processeur. Pour les pipelines liés à la mémoire, utilisez la limite de mémoire pour calculer le traitement parallèle maximal. Pour les pipelines liés au processeur, il est recommandé de conserver la règle d'autoscaling par défaut au lieu de fournir une optimisation du parallélisme.

Il est possible d'affiner la valeur pour tenir compte des besoins de traitement des autres étapes, comme l'écriture dans un récepteur. Augmenter la valeur de 1 ou 2 lorsque le parallélisme de votre modèle est de 2 permet de reconnaître le temps de traitement plus rapide de l'écriture dans le récepteur en lui donnant plus de marge pour tenir compte du traitement effectué dans d'autres étapes. Si votre pipeline n'implique pas de shuffle et que les transformations sont fusionnées en une seule étape, il n'est pas nécessaire d'ajuster la valeur pour les autres transformations.

Cette valeur peut également être ajustée pour simuler les effets des retards de backlog acceptables. Par exemple, si vous acceptez un délai maximal de 10 minutes et que le temps de traitement moyen de votre modèle est de 1 minute, vous pouvez choisir d'augmenter la valeur de 1 en supposant que le nombre maximal de nœuds de calcul est défini sur 10.

Heuristique d'autoscaling nécessitant une utilisation intensive du GPU

Dans le paramètre nécessitant une utilisation intensive du GPU indiqué par la définition de l'optimisation du parallélisme, Dataflow prend en compte plusieurs facteurs lors de l'autoscaling. y compris les suivants :

  • Clés disponibles. Les clés sont l'unité de base du parallélisme dans Dataflow.
  • Nombre maximal de bundles actifs par nœud de calcul. Cela suggère le nombre idéal maximal de parallélisme de traitement dans le nœud de calcul.

L'idée générale derrière les décisions de scaling est de calculer les nœuds de calcul nécessaires pour gérer la charge actuelle, comme indiqué par les clés disponibles. Par exemple, si 100 clés sont disponibles pour être traitées et que le parallélisme maximal par nœud de calcul est de 10, vous devez viser un total de 10 nœuds de calcul.

Si votre pipeline est complexe et comporte à la fois une charge de travail nécessitant une utilisation intensive du GPU et de nombreuses transformations nécessitant une utilisation intensive du processeur, nous vous recommandons d'activer l'ajustement approprié. Cela permet au service de faire une bonne distinction entre le travail nécessitant une utilisation intensive du processeur et celui nécessitant une utilisation intensive du GPU, puis de mettre à l'échelle chaque pool de nœuds de calcul en conséquence.

Définir le niveau de latence

L'autoscaling horizontal Dataflow utilise la durée du backlog pour déterminer quand mettre à l'échelle les nœuds de calcul. Par défaut, Dataflow vise un backlog de moins de 15 secondes, ce qui convient aux charges de travail en temps réel, mais peut entraîner des coûts de ressources plus élevés.

Pour les jobs de traitement en flux continu qui utilisent Streaming Engine, vous pouvez configurer le seuil de backlog en choisissant un niveau de latence. Un niveau de latence correspond à une durée de backlog cible, ce qui vous permet de configurer la sensibilité de l'autoscaler à la durée du backlog. Une décision d'autoscaling n'est déclenchée que lorsque la durée du backlog dépasse le niveau sélectionné pendant plusieurs minutes.

Le tableau suivant décrit les niveaux de latence disponibles :

Niveau de latence Temps d'attente estimé autorisé Cas d'utilisation prévu
low_latency Moins de 15 secondes Privilégie un scaling à la hausse rapide pour maintenir le backlog au minimum. Valeur par défaut sélectionnée si aucun niveau n'est spécifié
medium_latency Moins de 2 minutes Compromis équilibré entre la latence et les coûts des ressources.
high_latency Moins de 10 minutes Privilégie les économies en permettant l'accumulation du backlog. Conçu pour les charges de travail tolérant les retards.

Pour configurer le niveau de latence lorsque vous exécutez un job, définissez l'option d'expérimentation enable_latency_tier_lightweight_update :

--experiments=enable_latency_tier_lightweight_update=TIER

Remplacez TIER par l'un des niveaux de latence disponibles (par exemple, medium_latency).

Mettre à jour le niveau de latence

Pour mettre à jour le niveau de latence pendant l'exécution d'un job, effectuez une mise à jour en cours à l'aide de la gcloud dataflow jobs update-options :

gcloud dataflow jobs update-options \
  --region=REGION \
  --latency-tier=TIER \
  JOB_ID

Remplacez les éléments suivants :

  • REGION : ID de région du point de terminaison régional de la tâche.
  • JOB_ID : ID de la tâche à mettre à jour.
  • TIER : niveau de latence cible. Au choix : low_latency, medium_latency ou high_latency.

Pour réinitialiser le niveau de latence et revenir au comportement par défaut de faible latence, utilisez la commande suivante :

gcloud dataflow jobs update-options \
  --unset-latency-tier \
  --region=REGION \
  JOB_ID

Heuristique d'autoscaling de flux

Pour les pipelines de traitement en flux continu, l'objectif de l'autoscaling horizontal est de minimiser les jobs en attente tout en optimisant l'utilisation et le débit des nœuds de calcul, ainsi que de réagir rapidement aux pics de charge.

Dataflow prend en compte plusieurs facteurs lors de l'autoscaling, y compris les suivants :

  • En attente. Le temps d'attente estimé est calculé à partir du débit et des octets en attente de traitement à partir de la source d'entrée. Un pipeline est considéré comme en retard lorsque le temps d'attente estimé est supérieur au seuil configuré. Par défaut, ce seuil est de 15 secondes. Pour en savoir plus, consultez la section Définir le niveau de latence.

  • Objectif d'utilisation du processeur. L'objectif par défaut pour l'utilisation moyenne du processeur est de 0,8. Vous pouvez remplacer cette valeur.

  • Clés disponibles. Les clés sont l'unité de base du parallélisme dans Dataflow.

Dans certains cas, Dataflow utilise les facteurs suivants dans les décisions d'autoscaling. Si ces facteurs sont utilisés pour votre job, vous pouvez consulter ces informations dans l'onglet des métriques Autoscaling.

  • La limitation basée sur les clés utilise le nombre de clés de traitement reçues par le job pour calculer la limite des nœuds de calcul utilisateur, car chaque clé ne peut être traitée que par un seul nœud de calcul à la fois.

  • Blocage du scaling à la baisse. Si Dataflow détecte que des décisions d'autoscaling instables ont été prises, il ralentit le taux de scaling à la baisse afin d'améliorer la stabilité.

  • Le scaling à la hausse basé sur le processeur utilise l'utilisation élevée du processeur comme critère de scaling à la hausse.

  • Pour les tâches de traitement en flux continu qui n'utilisent pas Streaming Engine, le scaling peut être limité par le nombre de disques persistants. Pour en savoir plus, consultez la section Définir la plage d'autoscaling.

  • Autoscaling nécessitant une utilisation intensive du GPU, s'il est activé en définissant l'optimisation du parallélisme des nœuds de calcul. Pour en savoir plus, consultez la section Heuristique d'autoscaling nécessitant une utilisation intensive du GPU.

Augmentation de la résolution. Si un pipeline de traitement en flux continu reste en retard avec un parallélisme suffisant sur les nœuds de calcul pendant plusieurs minutes, Dataflow effectue un scaling à la hausse. Dataflow tente de supprimer le traitement en attente dans les 150 secondes environ suivant le scaling à la hausse, en fonction du débit actuel par nœud de calcul. En cas de retard alors que le nœud de calcul n'a pas assez de parallélisme pour les nœuds de calcul supplémentaires, le pipeline n'effectue pas de scaling à la hausse. (Le scaling du nombre de nœuds de calcul au-delà du nombre de clés disponibles pour le traitement parallèle n'aide pas à traiter le backlog plus rapidement.)

Scaling à la baisse : lorsque l'autoscaler prend une décision concernant un scaling à la baisse, les tâches en attente est le facteur de priorité le plus élevé. L'autoscaler vise un backlog basé sur le niveau de latence. Si le temps de traitement des tâches en attente passe suffisamment en dessous de l'objectif pendant plusieurs minutes et que l'utilisation moyenne des nœuds de calcul est inférieure à l'objectif d'utilisation du processeur, Dataflow effectue un scaling à la baisse. Tant que le nombre de tâches en attente est acceptable, l'autoscaler tente de maintenir l'utilisation du processeur proche de l'utilisation cible du processeur. Toutefois, si l'utilisation est déjà suffisamment proche de l'objectif, l'autoscaler peut conserver même le nombre de nœuds de calcul, car chaque étape de scaling à la baisse a un coût.

Streaming Engine utilise également une technique d'autoscaling prédictif basée sur les timers en attente. Les données illimitées d'un pipeline de traitement en flux continu sont divisées en fenêtres regroupées par code temporel. À la fin d'une fenêtre, des timers (minuteurs) se déclenchent pour chaque clé en cours de traitement dans cette fenêtre. Le déclenchement d'un timer indique que la fenêtre a expiré pour une clé donnée. Streaming Engine peut mesurer les timers en attente et prédire combien d'entre eux se déclencheront à la fermeture d'une fenêtre. En utilisant les timers en attente comme signal, Dataflow peut estimer le volume de traitement qui doit avoir lieu lors du déclenchement des timers à venir. Dataflow s'adapte automatiquement et à l'avance suivant la charge future estimée afin de répondre à la demande prévue.

Métriques

Pour connaître les limites actuelles d'autoscaling d'une tâche, interrogez les métriques suivantes :

  • job/max_worker_instances_limit : Nombre maximal de nœuds de calcul
  • job/min_worker_instances_limit : Nombre minimal de nœuds de calcul

Pour obtenir des informations sur l'utilisation des nœuds de calcul, interrogez les métriques suivantes :

  • job/aggregated_worker_utilization : utilisation agrégée des nœuds de calcul.
  • job/worker_utilization_hint : optimisation de l'utilisation actuelle des nœuds de calcul.

Pour obtenir des informations sur le comportement de l'autoscaler, interrogez la métrique suivante :

  • job.worker_utilization_hint_is_actively_used : indique si l'autoscaler utilise activement l'optimisation d'utilisation des nœuds de calcul. Si d'autres facteurs remplacent la valeur d'optimisation lors de l'échantillonnage de cette métrique, la valeur est false.
  • job/horizontal_worker_scaling : décrit les décisions prises par l'autoscaler. Cette métrique contient les étiquettes suivantes :
    • direction : indique si l'autoscaler a effectué un scaling à la hausse, à la baisse ou n'a effectué aucune action.
    • rationale : spécifie la logique de la décision de l'autoscaler.

Pour en savoir plus, consultez les métriques Cloud Monitoring. Ces métriques sont également affichées dans les graphiques de surveillance de l'autoscaling.

Étapes suivantes