Restez organisé à l'aide des collections
Enregistrez et classez les contenus selon vos préférences.
La fonctionnalité de rééquilibrage dynamique des tâches du service Dataflow permet au service de repartitionner le travail de manière dynamique en fonction des conditions d'exécution. Ces conditions peuvent être les suivantes :
Déséquilibres dans les attributions de tâches
Nœuds de calcul nécessitant plus de temps que prévu pour terminer
Nœuds de calcul nécessitant moins de temps que prévu pour terminer
Le service Dataflow détecte automatiquement ces conditions et peut attribuer de manière dynamique des tâches à des nœuds de calcul inutilisés ou sous-utilisés afin de réduire le temps de traitement global de votre tâche.
Limites
Le rééquilibrage dynamique du travail a lieu uniquement lorsque le service Dataflow traite certaines données d'entrée en parallèle : lors de la lecture de données à partir d'une source d'entrée externe, lors de l'utilisation d'une classe PCollection intermédiaire matérialisée, ou lors de l'utilisation du résultat d'une agrégation, comme GroupByKey. Si de nombreuses étapes de votre tâche sont fusionnées, votre tâche comporte moins de classes PCollection intermédiaires, et le rééquilibrage dynamique du travail est limité à la nombre d'éléments dans la PCollection matérialisée source. Si vous voulez vous assurer que le rééquilibrage dynamique du travail peut être appliqué à une classe PCollection particulière de votre pipeline, vous pouvez empêcher la fusion de différentes manières afin de garantir un parallélisme dynamique.
Le rééquilibrage dynamique du travail ne peut pas remettre en parallèle des données plus fines qu'un seul enregistrement.
Si vos données contiennent des enregistrements individuels qui entraînent des temps de traitement importants, elles peuvent retarder votre tâche. Dataflow ne peut pas subdiviser et redistribuer un enregistrement "à chaud" individuel sur plusieurs nœuds de calcul.
Java
Si vous définissez un nombre fixe de segments pour la sortie finale de votre pipeline (par exemple, en écrivant des données à l'aide de TextIO.Write.withNumShards), Dataflow limite la parallélisation en fonction du nombre de segments que vous choisissez.
Python
Si vous définissez un nombre fixe de segments pour la sortie finale de votre pipeline (par exemple, en écrivant des données à l'aide de beam.io.WriteToText(..., num_shards=...)), Dataflow limite la parallélisation en fonction du nombre de segments que vous choisissez.
Go
Si vous définissez un nombre fixe de segments pour la sortie finale de votre pipeline, Dataflow limite la parallélisation en fonction du nombre de segments que vous choisissez.
Travailler avec des sources de données personnalisées
Java
Si votre pipeline utilise une source de données personnalisée que vous fournissez, vous devez mettre en œuvre la méthode splitAtFraction pour permettre à votre source d'utiliser la fonctionnalité de rééquilibrage dynamique du travail.
Si vous mettez en œuvre splitAtFraction de manière incorrecte, les enregistrements de votre source peuvent être dupliqués ou supprimés. Consultez les informations de référence de l'API sur RangeTracker pour obtenir de l'aide et des conseils sur la mise en œuvre de splitAtFraction.
Python
Si votre pipeline utilise une source de données personnalisée que vous fournissez, votre RangeTracker doit mettre en œuvre try_claim, try_split, position_at_fraction et fraction_consumed pour permettre à votre source d'utiliser la fonctionnalité de rééquilibrage dynamique du travail.
Si votre pipeline utilise une source de données personnalisée que vous fournissez, vous devez mettre en œuvre une méthode RTracker valide pour permettre à votre source d'utiliser la fonctionnalité de rééquilibrage dynamique du travail.
Le rééquilibrage dynamique du travail est activé par la valeur de retour de la méthode getProgress() de votre source personnalisée. L'implémentation par défaut de getProgress() renvoie null. Pour garantir l'activation de l'autoscaling, assurez-vous que votre source personnalisée remplace getProgress()afin de renvoyer une valeur appropriée.
Sauf indication contraire, le contenu de cette page est régi par une licence Creative Commons Attribution 4.0, et les échantillons de code sont régis par une licence Apache 2.0. Pour en savoir plus, consultez les Règles du site Google Developers. Java est une marque déposée d'Oracle et/ou de ses sociétés affiliées.
Dernière mise à jour le 2025/09/04 (UTC).
[[["Facile à comprendre","easyToUnderstand","thumb-up"],["J'ai pu résoudre mon problème","solvedMyProblem","thumb-up"],["Autre","otherUp","thumb-up"]],[["Difficile à comprendre","hardToUnderstand","thumb-down"],["Informations ou exemple de code incorrects","incorrectInformationOrSampleCode","thumb-down"],["Il n'y a pas l'information/les exemples dont j'ai besoin","missingTheInformationSamplesINeed","thumb-down"],["Problème de traduction","translationIssue","thumb-down"],["Autre","otherDown","thumb-down"]],["Dernière mise à jour le 2025/09/04 (UTC)."],[[["\u003cp\u003eThe Dataflow service's Dynamic Work Rebalancing feature automatically redistributes work among workers based on runtime conditions such as work imbalances or varying processing times.\u003c/p\u003e\n"],["\u003cp\u003eDynamic work rebalancing is limited to parallel data processing stages, like reading from external sources or working with materialized \u003ccode\u003ePCollection\u003c/code\u003es, and is restricted by the number of elements or shards in those stages.\u003c/p\u003e\n"],["\u003cp\u003eIf you have custom data sources, dynamic work rebalancing requires implementing specific methods in your data source, such as \u003ccode\u003esplitAtFraction\u003c/code\u003e in Java or \u003ccode\u003etry_split\u003c/code\u003e and \u003ccode\u003eposition_at_fraction\u003c/code\u003e in Python, in order to function correctly.\u003c/p\u003e\n"],["\u003cp\u003eDynamic work rebalancing cannot further divide and redistribute a single record that is processing slower than the rest, potentially causing delays.\u003c/p\u003e\n"],["\u003cp\u003eSetting a fixed number of shards for your pipeline's output limits the parallelization that Dataflow can perform, thereby impacting the effectiveness of dynamic work rebalancing.\u003c/p\u003e\n"]]],[],null,["# Dynamic work rebalancing\n\nThe Dynamic Work Rebalancing feature of the Dataflow service allows the\nservice to dynamically repartition work based on runtime conditions. These\nconditions might include the following:\n\n- Imbalances in work assignments\n- Workers taking longer than expected to finish\n- Workers finishing faster than expected\n\nThe Dataflow service automatically detects these conditions and\ncan dynamically assign work to unused or underused workers to decrease\nthe overall processing time of your job.\n\nLimitations\n-----------\n\nDynamic work rebalancing only happens when the Dataflow service is\nprocessing some input data in parallel: when reading data from an external input\nsource, when working with a materialized intermediate `PCollection`, or when\nworking with the result of an aggregation like `GroupByKey`. If a large number\nof steps in your job are\n[fused](/dataflow/docs/pipeline-lifecycle#fusion_optimization), your job has fewer\nintermediate `PCollection`s, and dynamic work rebalancing is\nlimited to the number of elements in the source materialized `PCollection`. If\nyou want to ensure that dynamic work rebalancing can be applied to a particular\n`PCollection` in your pipeline, you can\n[prevent fusion](/dataflow/docs/pipeline-lifecycle#preventing_fusion) in a few\ndifferent ways to ensure dynamic parallelism.\n\nDynamic work rebalancing cannot reparallelize data finer than a single record.\nIf your data contains individual records that cause large delays in processing\ntime, they might still delay your job. Dataflow can't\nsubdivide and redistribute an individual \"hot\" record to multiple workers. \n\n### Java\n\nIf you set a fixed number of shards for the final output of your pipeline (for\nexample, by writing data using `TextIO.Write.withNumShards`),\nDataflow limits parallelization based on the number of\nshards that you choose.\n\n### Python\n\nIf you set a fixed number of shards for the final output of your pipeline (for\nexample, by writing data using `beam.io.WriteToText(..., num_shards=...)`),\nDataflow limits parallelization based on the number of\nshards that you choose.\n\n### Go\n\nIf you set a fixed number of shards for the final output of your pipeline,\nDataflow limits parallelization based on the number of shards\nthat you choose.\n| **Note:** The fixed-shards limitation can be considered temporary, and might be subject to change in future releases of the Dataflow service.\n\nWorking with Custom Data Sources\n--------------------------------\n\n### Java\n\nIf your pipeline uses a custom data source that you provide, you must\nimplement the method `splitAtFraction` to allow your source to work with the\ndynamic work rebalancing feature.\n| **Caution:** Using dynamic work rebalancing with custom data sources is an advanced use case. If you choose to implement `splitAtFraction`, it's critical that you test your code extensively and with maximum code coverage.\n\nIf you implement `splitAtFraction` incorrectly, records from your source might\nappear to get duplicated or dropped. See the\n[API reference information on RangeTracker](https://beam.apache.org/documentation/sdks/javadoc/current/index.html?org/apache/beam/sdk/io/range/RangeTracker.html) for help and tips on\nimplementing `splitAtFraction`.\n\n### Python\n\nIf your pipeline uses a custom data source that you provide, your\n`RangeTracker` must implement `try_claim`, `try_split`,\n`position_at_fraction`, and `fraction_consumed` to allow your source to work\nwith the dynamic work rebalancing feature.\n\nSee the\n[API reference information on RangeTracker](https://beam.apache.org/documentation/sdks/pydoc/current/apache_beam.io.iobase.html#apache_beam.io.iobase.RangeTracker)\nfor more information.\n\n### Go\n\nIf your pipeline uses a custom data source that you provide, you must\nimplement a valid `RTracker` to allow your source to work with the dynamic\nwork rebalancing feature.\n\nFor more information, see the [RTracker API reference information](https://pkg.go.dev/github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf#RTracker).\n\nDynamic work rebalancing uses the return value of the `getProgress()`\nmethod of your custom source to activate. The default implementation for `getProgress()` returns\n`null`. To ensure autoscaling activates, make sure your custom source overrides\n`getProgress()` to return an appropriate value."]]
