Cette page a été traduite par l'API Cloud Translation.
Switch to English

Entrée distribuée

Voir sur TensorFlow.org Exécuter dans Google Colab Afficher la source sur GitHub Télécharger le cahier

Les API tf.distribute offrent aux utilisateurs un moyen simple de faire évoluer leur formation d'une seule machine vers plusieurs machines. Lors de la mise à l'échelle de leur modèle, les utilisateurs doivent également distribuer leur entrée sur plusieurs appareils. tf.distribute fournit des API à l'aide desquelles vous pouvez automatiquement distribuer votre entrée sur les appareils.

Ce guide vous montrera les différentes façons dont vous pouvez créer un ensemble de données et des itérateurs distribués à l'aide des API tf.distribute . De plus, les sujets suivants seront abordés:

Ce guide ne couvre pas l'utilisation de l'entrée distribuée avec les API Keras.

Ensembles de données distribués

Pour utiliser les API tf.distribute à l'échelle, il est recommandé aux utilisateurs d'utiliser tf.data.Dataset pour représenter leur entrée. tf.distribute a été conçu pour fonctionner efficacement avec tf.data.Dataset (par exemple, la prélecture automatique des données sur chaque accélérateur) avec des optimisations de performances régulièrement intégrées dans l'implémentation. Si vous avez un cas d'utilisation pour utiliser autre chose que tf.data.Dataset , veuillez vous reporter à une section ultérieure de ce guide. Dans une boucle d'apprentissage non distribuée, les utilisateurs créent d'abord une instance tf.data.Dataset , puis tf.data.Dataset les éléments. Par exemple:

 import tensorflow as tf

# Helper libraries
import numpy as np
import os

print(tf.__version__)
 
2.3.0

 global_batch_size = 16
# Create a tf.data.Dataset object.
dataset = tf.data.Dataset.from_tensors(([1.], [1.])).repeat(100).batch(global_batch_size)

@tf.function
def train_step(inputs):
  features, labels = inputs
  return labels - 0.3 * features

# Iterate over the dataset using the for..in construct.
for inputs in dataset:
  print(train_step(inputs))

 
tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(4, 1), dtype=float32)

Pour permettre aux utilisateurs d'utiliser la stratégie tf.distribute avec des modifications minimales du code existant d'un utilisateur, deux API ont été introduites qui distribueraient une instance tf.data.Dataset et retourneraient un objet d'ensemble de données distribué. Un utilisateur peut ensuite parcourir cette instance de jeu de données distribué et entraîner son modèle comme auparavant. Examinons maintenant les deux API - tf.distribute.Strategy.experimental_distribute_dataset et tf.distribute.Strategy.experimental_distribute_datasets_from_function plus en détail:

tf.distribute.Strategy.experimental_distribute_dataset

Usage

Cette API prend une instance de tf.data.Dataset comme entrée et renvoie une instance de tf.distribute.DistributedDataset . Vous devez regrouper l'ensemble de données d'entrée avec une valeur égale à la taille globale du lot. Cette taille de lot globale correspond au nombre d'échantillons que vous souhaitez traiter sur tous les appareils en 1 étape. Vous pouvez parcourir cet ensemble de données distribué de manière pythonique ou créer un itérateur en utilisant iter . L'objet renvoyé n'est pas une instance de tf.data.Dataset et ne prend en charge aucune autre API qui transforme ou inspecte l'ensemble de données de quelque manière que ce soit. Il s'agit de l'API recommandée si vous ne disposez pas de méthodes spécifiques pour répartir votre entrée sur différents réplicas.

 global_batch_size = 16
mirrored_strategy = tf.distribute.MirroredStrategy()

dataset = tf.data.Dataset.from_tensors(([1.], [1.])).repeat(100).batch(global_batch_size)
# Distribute input using the `experimental_distribute_dataset`.
dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)
# 1 global batch of data fed to the model in 1 step.
print(next(iter(dist_dataset)))
 
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)
WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.6/site-packages/tensorflow/python/data/ops/multi_device_iterator_ops.py:601: get_next_as_optional (from tensorflow.python.data.ops.iterator_ops) is deprecated and will be removed in a future version.
Instructions for updating:
Use `tf.data.Iterator.get_next_as_optional()` instead.
(<tf.Tensor: shape=(16, 1), dtype=float32, numpy=
array([[1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.]], dtype=float32)>, <tf.Tensor: shape=(16, 1), dtype=float32, numpy=
array([[1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.],
       [1.]], dtype=float32)>)

Propriétés

Traitement par lots

tf.distribute rebatche l'instance tf.data.Dataset entrée avec une nouvelle taille de lot égale à la taille de lot globale divisée par le nombre de répliques synchronisées. Le nombre de répliques synchronisées est égal au nombre d'appareils qui participent au gradient allreduce pendant l'entraînement. Lorsqu'un utilisateur appelle le next sur l'itérateur distribué, une taille de lot de données par réplica est renvoyée sur chaque réplica. La cardinalité du jeu de données rebatché sera toujours un multiple du nombre de répliques. Voici quelques exemples:

  • tf.data.Dataset.range(6).batch(4, drop_remainder=False)

    Sans distribution:

    Lot 1: [0, 1, 2, 3]

    Lot 2: [4, 5]

    Avec distribution sur 2 répliques:

    Lot 1: Réplique 1: [0, 1] Réplique 2: [2, 3]

    Lot 2: réplique 2: [4] réplique 2: [5]

    Le dernier lot ([4, 5]) est divisé entre 2 répliques.

  • tf.data.Dataset.range(4).batch(4)

    Sans distribution:

    Lot 1: [[0], [1], [2], [3]]

    Avec distribution sur 5 répliques:

    Lot 1: Réplique 1: [0] Réplique 2: [1] Réplique 3: [2] Réplique 4: [3] Réplique 5: []

  • tf.data.Dataset.range(8).batch(4)

    Sans distribution:

    Lot 1: [0, 1, 2, 3]

    Lot 2: [4, 5, 6, 7]

    Avec distribution sur 3 répliques:

    Lot 1: Réplique 1: [0, 1] Réplique 2: [2, 3] Réplique 3: []

    Lot 2: Réplique 1: [4, 5] Réplique 2: [6, 7] Réplique 3: []

Le rebatching du jeu de données a une complexité spatiale qui augmente linéairement avec le nombre de répliques. Cela signifie que pour le cas d'utilisation de la formation multi-ouvriers, le pipeline d'entrée peut rencontrer des erreurs MOO.

Sharding

tf.distribute également le partage automatique de l'ensemble de données d'entrée dans la formation multi-ouvriers. Chaque ensemble de données est créé sur le périphérique CPU du travailleur. Le partage automatique d'un ensemble de données sur un ensemble de nœuds de calcul signifie que chaque collaborateur se voit attribuer un sous-ensemble de l'ensemble de données complet (si le bon tf.data.experimental.AutoShardPolicy est défini). Ceci permet de garantir qu'à chaque étape, une taille de lot globale d'éléments de jeu de données ne se chevauchant pas sera traitée par chaque travailleur. Le partage automatique a quelques options différentes qui peuvent être spécifiées à l'aide de tf.data.experimental.DistributeOptions .

 dataset = tf.data.Dataset.from_tensors(([1.],[1.])).repeat(64).batch(16)
options = tf.data.Options()
options.experimental_distribute.auto_shard_policy = tf.data.experimental.AutoShardPolicy.DATA
dataset = dataset.with_options(options)
 

Vous pouvez définir trois options différentes pour le tf.data.experimental.AutoShardPolicy :

  • AUTO: il s'agit de l'option par défaut, ce qui signifie qu'une tentative de partitionnement sera effectuée par FILE. La tentative de partitionnement par FILE échoue si un ensemble de données basé sur un fichier n'est pas détecté. tf.distribute reviendra alors au partitionnement par DATA. Notez que si l'ensemble de données d'entrée est basé sur un fichier mais que le nombre de fichiers est inférieur au nombre de nœuds de calcul, une erreur sera générée.
  • FICHIER: c'est l'option si vous voulez partager les fichiers d'entrée sur tous les travailleurs. Si le nombre de fichiers est inférieur au nombre de nœuds de calcul, une erreur sera générée. Vous devez utiliser cette option si le nombre de fichiers d'entrée est beaucoup plus grand que le nombre de nœuds de calcul et que les données des fichiers sont uniformément réparties. L'inconvénient de cette option est d'avoir des travailleurs inactifs si les données des fichiers ne sont pas uniformément réparties. Par exemple, distribuons 2 fichiers sur 2 nœuds de calcul avec 1 réplique chacun. Le fichier 1 contient [0, 1, 2, 3, 4, 5] et le fichier 2 contient [6, 7, 8, 9, 10, 11]. Supposons que le nombre total de répliques synchronisées soit de 2 et la taille globale du lot de 4.

    • Travailleur 0:

    Lot 1 = Réplique 1: [0, 1]

    Lot 2 = Réplique 1: [2, 3]

    Lot 3 = Réplique 1: [4]

    Lot 4 = réplique 1: [5]

    • Travailleur 1:

    Lot 1 = Réplique 2: [6, 7]

    Lot 2 = Réplique 2: [8, 9]

    Lot 3 = Réplique 2: [10]

    Lot 4 = Réplique 2: [11]

  • DONNÉES: Cela va autoshard les éléments sur tous les travailleurs. Chacun des nœuds de calcul lira l'intégralité du jeu de données et ne traitera que le fragment qui lui est attribué. Tous les autres fragments seront supprimés. Ceci est généralement utilisé si le nombre de fichiers d'entrée est inférieur au nombre de nœuds de calcul et que vous souhaitez un meilleur partage des données sur tous les nœuds de calcul. L'inconvénient est que l'ensemble de données sera lu sur chaque travailleur. Par exemple, distribuons 1 fichiers sur 2 travailleurs. Le fichier 1 contient [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]. Soit le nombre total de répliques synchronisées égale à 2.

    • Travailleur 0:

    Lot 1 = Réplique 1: [0, 1]

    Lot 2 = Réplique 1: [4, 5]

    Lot 3 = Réplique 1: [8, 9]

    • Travailleur 1:

    Lot 1 = réplique 2: [2, 3]

    Lot 2 = Réplique 2: [6, 7]

    Lot 3 = Réplique 2: [10, 11]

  • OFF: Si vous désactivez le partage automatique, chaque collaborateur traitera toutes les données. Par exemple, distribuons 1 fichiers sur 2 travailleurs. Le fichier 1 contient [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]. Soit le nombre total de réplicas synchronisés à 2. Ensuite, chaque worker verra la distribution suivante:

    • Travailleur 0:

    Lot 1 = Réplique 1: [0, 1]

    Lot 2 = Réplique 1: [2, 3]

    Lot 3 = Réplique 1: [4, 5]

    Lot 4 = Réplique 1: [6, 7]

    Lot 5 = Réplique 1: [8, 9]

    Lot 6 = Réplique 1: [10, 11]

    • Travailleur 1:

    Lot 1 = Réplique 2: [0, 1]

    Lot 2 = Réplique 2: [2, 3]

    Lot 3 = Réplique 2: [4, 5]

    Lot 4 = Réplique 2: [6, 7]

    Lot 5 = Réplique 2: [8, 9]

    Lot 6 = Réplique 2: [10, 11]

Prérécupération

Par défaut, tf.distribute ajoute une transformation de prélecture à la fin de l'instance tf.data.Dataset fournie par l'utilisateur. L'argument de la transformation de prélecture qui est buffer_size est égal au nombre de répliques synchronisées.

tf.distribute.Strategy.experimental_distribute_datasets_from_function

Usage

Cette API prend une fonction d'entrée et renvoie une instance de tf.distribute.DistributedDataset . La fonction d'entrée que les utilisateurs transmettent a un argument tf.distribute.InputContext et doit renvoyer une instance de tf.data.Dataset . Avec cette API, tf.distribute n'apporte aucune autre modification à l'instance tf.data.Dataset l'utilisateur renvoyée par la fonction d'entrée. Il est de la responsabilité de l'utilisateur de regrouper et de partager l'ensemble de données. tf.distribute appelle la fonction d'entrée sur le périphérique CPU de chacun des travailleurs. En plus de permettre aux utilisateurs de spécifier leur propre logique de traitement par lots et de partitionnement, cette API démontre également une meilleure évolutivité et des performances par rapport à tf.distribute.Strategy.experimental_distribute_dataset lorsqu'elle est utilisée pour la formation multi-ouvriers.

 mirrored_strategy = tf.distribute.MirroredStrategy()

def dataset_fn(input_context):
  batch_size = input_context.get_per_replica_batch_size(global_batch_size)
  dataset = tf.data.Dataset.from_tensors(([1.],[1.])).repeat(64).batch(16)
  dataset = dataset.shard(
    input_context.num_input_pipelines, input_context.input_pipeline_id)
  dataset = dataset.batch(batch_size)
  dataset = dataset.prefetch(2) # This prefetches 2 batches per device.
  return dataset

dist_dataset = mirrored_strategy.experimental_distribute_datasets_from_function(dataset_fn)
 
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)

Propriétés

Traitement par lots

L'instance tf.data.Dataset qui est la valeur de retour de la fonction d'entrée doit être traitée par lots en utilisant la taille de lot par réplica. La taille de lot par réplica correspond à la taille globale du lot divisée par le nombre de réplicas participant à la formation à la synchronisation. Cela est dû au fait que tf.distribute appelle la fonction d'entrée sur le périphérique CPU de chacun des tf.distribute . L'ensemble de données créé sur un worker donné doit être prêt à être utilisé par tous les réplicas de ce worker.

Sharding

L'objet tf.distribute.InputContext qui est implicitement passé comme argument à la fonction d'entrée de l'utilisateur est créé par tf.distribute sous le capot. Elle contient des informations sur le nombre de nœuds de calcul, l'ID de travail actuel, etc. Cette fonction d'entrée peut gérer le partage selon les règles définies par l'utilisateur à l'aide de ces propriétés qui font partie de l'objet tf.distribute.InputContext .

Prérécupération

tf.distribute n'ajoute pas de transformation de prélecture à la fin du tf.data.Dataset renvoyé par la fonction d'entrée fournie par l'utilisateur.

Itérateurs distribués

Comme pour les instances de tf.data.Dataset non distribuées, vous devrez créer un itérateur sur les instances de tf.distribute.DistributedDataset pour effectuer une itération dessus et accéder aux éléments dans le tf.distribute.DistributedDataset . Voici les façons dont vous pouvez créer un tf.distribute.DistributedIterator et l'utiliser pour entraîner votre modèle:

Coutumes

Utiliser une construction de boucle Pythonic for

Vous pouvez utiliser une boucle Pythonic conviviale pour parcourir le tf.distribute.DistributedDataset . Les éléments renvoyés par le tf.distribute.DistributedIterator peuvent être un seul tf.Tensor ou un tf.distribute.DistributedValues qui contient une valeur par réplica. Placer la boucle dans une fonction tf.function augmentera les performances. Cependant, break et return sont actuellement pas pris en charge si la boucle est placée dans une fonction tf.function . Nous ne prenons pas non plus en charge le placement de la boucle dans un tf.function lors de l'utilisation de stratégies multi-ouvriers telles que tf.distribute.experimental.MultiWorkerMirroredStrategy et tf.distribute.TPUStrategy . Placer la boucle à l'intérieur de tf.function fonctionne pour le travailleur unique tf.distribute.TPUStrategy mais pas lors de l'utilisation de pods TPU.

 global_batch_size = 16
mirrored_strategy = tf.distribute.MirroredStrategy()

dataset = tf.data.Dataset.from_tensors(([1.],[1.])).repeat(100).batch(global_batch_size)
dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)

@tf.function
def train_step(inputs):
  features, labels = inputs
  return labels - 0.3 * features

for x in dist_dataset:
  # train_step trains the model using the dataset elements
  loss = mirrored_strategy.run(train_step, args=(x,))
  print("Loss is ", loss)
 
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(4, 1), dtype=float32)

Utilisez iter pour créer un itérateur explicite

Pour parcourir les éléments d'une instance de tf.distribute.DistributedDataset , vous pouvez créer un tf.distribute.DistributedIterator à l'aide de l'API iter . Avec un itérateur explicite, vous pouvez itérer pour un nombre fixe d'étapes. Afin d'obtenir l'élément suivant d'un tf.distribute.DistributedIterator exemple dist_iterator , vous pouvez appeler à next(dist_iterator) , dist_iterator.get_next() , ou dist_iterator.get_next_as_optional() . Les deux premiers sont essentiellement les mêmes:

 num_epochs = 10
steps_per_epoch = 5
for epoch in range(num_epochs):
  dist_iterator = iter(dist_dataset)
  for step in range(steps_per_epoch):
    # train_step trains the model using the dataset elements
    loss = mirrored_strategy.run(train_step, args=(next(dist_iterator),))
    # which is the same as
    # loss = mirrored_strategy.run(train_step, args=(dist_iterator.get_next(),))
    print("Loss is ", loss)
 
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)
Loss is  tf.Tensor(
[[0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]
 [0.7]], shape=(16, 1), dtype=float32)

Avec next() ou tf.distribute.DistributedIterator.get_next() , si le tf.distribute.DistributedIterator a atteint sa fin, une erreur OutOfRange sera générée. Le client peut détecter l'erreur côté python et continuer à effectuer d'autres tâches telles que le point de contrôle et l'évaluation. Cependant, cela ne fonctionnera pas si vous utilisez une boucle d'entraînement hôte (c'est-à-dire, exécutez plusieurs étapes par tf.function ), qui ressemble à:

 @tf.function
def train_fn(iterator):
  for _ in tf.range(steps_per_loop):
    strategy.run(step_fn, args=(next(iterator),))
 

train_fn contient plusieurs étapes en enveloppant le corps de l'étape dans un tf.range . Dans ce cas, différentes itérations dans la boucle sans dépendance pourraient démarrer en parallèle, de sorte qu'une erreur OutOfRange peut être déclenchée dans les itérations ultérieures avant la fin du calcul des itérations précédentes. Une fois qu'une erreur OutOfRange est levée, toutes les opérations de la fonction se terminent immédiatement. Si c'est un cas que vous souhaitez éviter, une alternative qui ne renvoie pas d'erreur tf.distribute.DistributedIterator.get_next_as_optional() est tf.distribute.DistributedIterator.get_next_as_optional() . get_next_as_optional renvoie un tf.experimental.Optional qui contient l'élément suivant ou aucune valeur si le tf.distribute.DistributedIterator a atteint la fin.

 # You can break the loop with get_next_as_optional by checking if the Optional contains value
global_batch_size = 4
steps_per_loop = 5
strategy = tf.distribute.MirroredStrategy(devices=["GPU:0", "CPU:0"])

dataset = tf.data.Dataset.range(9).batch(global_batch_size)
distributed_iterator = iter(strategy.experimental_distribute_dataset(dataset))

@tf.function
def train_fn(distributed_iterator):
  for _ in tf.range(steps_per_loop):
    optional_data = distributed_iterator.get_next_as_optional()
    if not optional_data.has_value():
      break
    per_replica_results = strategy.run(lambda x:x, args=(optional_data.get_value(),))
    tf.print(strategy.experimental_local_results(per_replica_results))
train_fn(distributed_iterator)
 
WARNING:tensorflow:There are non-GPU devices in `tf.distribute.Strategy`, not using nccl allreduce.
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0', '/job:localhost/replica:0/task:0/device:CPU:0')
([0 1], [2 3])
([4 5], [6 7])
([8], [])

Utilisation de la propriété element_spec

Si vous transmettez les éléments d'un ensemble de données distribué à un tf.function et que vous souhaitez une garantie tf.TypeSpec , vous pouvez spécifier l'argument input_signature de la tf.function . La sortie d'un ensemble de données distribué est tf.distribute.DistributedValues qui peut représenter l'entrée d'un seul ou de plusieurs périphériques. Pour obtenir le tf.TypeSpec correspondant à cette valeur distribuée, vous pouvez utiliser la propriété element_spec de l'ensemble de données distribué ou de l'objet itérateur distribué.

 global_batch_size = 16
epochs = 5
steps_per_epoch = 5
mirrored_strategy = tf.distribute.MirroredStrategy()

dataset = tf.data.Dataset.from_tensors(([1.],[1.])).repeat(100).batch(global_batch_size)
dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)

@tf.function(input_signature=[dist_dataset.element_spec])
def train_step(per_replica_inputs):
  def step_fn(inputs):
    return 2 * inputs
  
  return mirrored_strategy.run(step_fn, args=(per_replica_inputs,))

for _ in range(epochs):
  iterator = iter(dist_dataset)
  for _ in range(steps_per_epoch):
    output = train_step(next(iterator))
    tf.print(output)
 
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])
([[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]], [[1]
 [1]
 [1]
 ...
 [1]
 [1]
 [1]])

Lots partiels

Des lots partiels sont rencontrés lorsque les instances de tf.data.Dataset par les utilisateurs peuvent contenir des tailles de lot qui ne sont pas divisibles de manière égale par le nombre de répliques ou lorsque la cardinalité de l'instance de l'ensemble de données n'est pas divisible par la taille du lot. Cela signifie que lorsque l'ensemble de données est distribué sur plusieurs répliques, le next appel sur certains itérateurs entraînera une OutOfRangeError. Pour gérer ce cas d'utilisation, tf.distribute retourne des lots factices de taille de lot 0 sur les réplicas qui n'ont plus de données à traiter.

Pour le cas de travail unique, si les données ne sont pas renvoyées par le next appel de l'itérateur, des lots factices de taille de lot 0 sont créés et utilisés avec les données réelles de l'ensemble de données. Dans le cas de lots partiels, le dernier lot global de données contiendra des données réelles aux côtés de lots de données factices. La condition d'arrêt du traitement des données vérifie désormais si l'une des répliques contient des données. S'il n'y a aucune donnée sur l'un des réplicas, une erreur OutOfRange est générée.

Pour le cas multi-ouvriers, la valeur booléenne représentant la présence de données sur chacun des nœuds de calcul est agrégée à l'aide d'une communication entre répliques et elle est utilisée pour identifier si tous les nœuds de calcul ont terminé le traitement de l'ensemble de données distribué. Étant donné que cela implique une communication entre travailleurs, il y a une certaine pénalité de performance.

Mises en garde

  • Lors de l'utilisation des API tf.distribute.Strategy.experimental_distribute_dataset avec une configuration de travail multiple, les utilisateurs transmettent un tf.data.Dataset qui lit des fichiers. Si tf.data.experimental.AutoShardPolicy est défini sur AUTO ou FILE , la taille de lot réelle par étape peut être inférieure à la taille de lot globale définie par l'utilisateur. Cela peut se produire lorsque les éléments restants du fichier sont inférieurs à la taille globale du lot. Les utilisateurs peuvent soit épuiser l'ensemble de données sans dépendre du nombre d'étapes à exécuter, soit définir tf.data.experimental.AutoShardPolicy sur DATA pour le contourner.

  • Les transformations d'ensemble de données avec tf.distribute ne sont actuellement pas prises en charge avec tf.distribute et toutes les opérations avec état que l'ensemble de données peut avoir sont actuellement ignorées. Par exemple, si votre ensemble de données a un map_fn qui utilise tf.random.uniform pour faire pivoter une image, alors vous avez un graphique d'ensemble de données qui dépend de l'état (c'est-à-dire de la graine aléatoire) sur la machine locale où le processus python est exécuté.

  • Expérimental tf.data.experimental.OptimizationOptions qui sont désactivées par défaut peuvent dans certains contextes - comme lorsqu'elles sont utilisées avec tf.distribute - provoquer une dégradation des performances. Vous ne devez les activer qu'après avoir vérifié qu'ils profitent aux performances de votre charge de travail dans un paramètre de distribution.

  • L'ordre dans lequel les données sont traitées par les travailleurs lors de l'utilisation de tf.distribute.experimental_distribute_dataset ou tf.distribute.experimental_distribute_datasets_from_function n'est pas garanti. Ceci est généralement requis si vous utilisez tf.distribute pour tf.distribute à l'échelle la prédiction. Vous pouvez cependant insérer un index pour chaque élément du lot et ordonner les sorties en conséquence. L'extrait de code suivant est un exemple de commande des sorties.

 mirrored_strategy = tf.distribute.MirroredStrategy()
dataset_size = 24
batch_size = 6
dataset = tf.data.Dataset.range(dataset_size).enumerate().batch(batch_size)
dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)

def predict(index, inputs):
  outputs = 2 * inputs
  return index, outputs

result = {}
for index, inputs in dist_dataset:
  output_index, outputs = mirrored_strategy.run(predict, args=(index, inputs))
  indices = list(mirrored_strategy.experimental_local_results(output_index))
  rindices = []
  for a in indices:
    rindices.extend(a.numpy())
  outputs = list(mirrored_strategy.experimental_local_results(outputs))
  routputs = []
  for a in outputs:
    routputs.extend(a.numpy())
  for i, value in zip(rindices, routputs):
    result[i] = value

print(result)
 
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)
WARNING:tensorflow:Using MirroredStrategy eagerly has significant overhead currently. We will be working on improving this in the future, but for now please wrap `call_for_each_replica` or `experimental_run` or `experimental_run_v2` inside a tf.function to get the best performance.
WARNING:tensorflow:Using MirroredStrategy eagerly has significant overhead currently. We will be working on improving this in the future, but for now please wrap `call_for_each_replica` or `experimental_run` or `experimental_run_v2` inside a tf.function to get the best performance.
WARNING:tensorflow:Using MirroredStrategy eagerly has significant overhead currently. We will be working on improving this in the future, but for now please wrap `call_for_each_replica` or `experimental_run` or `experimental_run_v2` inside a tf.function to get the best performance.
WARNING:tensorflow:Using MirroredStrategy eagerly has significant overhead currently. We will be working on improving this in the future, but for now please wrap `call_for_each_replica` or `experimental_run` or `experimental_run_v2` inside a tf.function to get the best performance.
{0: 0, 1: 2, 2: 4, 3: 6, 4: 8, 5: 10, 6: 12, 7: 14, 8: 16, 9: 18, 10: 20, 11: 22, 12: 24, 13: 26, 14: 28, 15: 30, 16: 32, 17: 34, 18: 36, 19: 38, 20: 40, 21: 42, 22: 44, 23: 46}

Comment distribuer mes données si je n'utilise pas une instance canonique de tf.data.Dataset?

Parfois, les utilisateurs ne peuvent pas utiliser un tf.data.Dataset pour représenter leur entrée et par la suite les API mentionnées ci-dessus pour distribuer l'ensemble de données à plusieurs appareils. Dans de tels cas, vous pouvez utiliser des tenseurs bruts ou des entrées d'un générateur.

Utilisez experimental_distribute_values_from_function pour les entrées de tenseurs arbitraires

strategy.run accepte tf.distribute.DistributedValues qui est la sortie de next(iterator) . Pour transmettre les valeurs de tenseur, utilisez experimental_distribute_values_from_function pour construire tf.distribute.DistributedValues partir de tenseurs bruts.

 mirrored_strategy = tf.distribute.MirroredStrategy()
worker_devices = mirrored_strategy.extended.worker_devices

def value_fn(ctx):
  return tf.constant(1.0)

distributed_values = mirrored_strategy.experimental_distribute_values_from_function(value_fn)
for _ in range(4):
  result = mirrored_strategy.run(lambda x:x, args=(distributed_values,))
  print(result)
 
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)
WARNING:tensorflow:Using MirroredStrategy eagerly has significant overhead currently. We will be working on improving this in the future, but for now please wrap `call_for_each_replica` or `experimental_run` or `experimental_run_v2` inside a tf.function to get the best performance.
tf.Tensor(1.0, shape=(), dtype=float32)
tf.Tensor(1.0, shape=(), dtype=float32)
tf.Tensor(1.0, shape=(), dtype=float32)
tf.Tensor(1.0, shape=(), dtype=float32)

Utilisez tf.data.Dataset.from_generator si votre entrée provient d'un générateur

Si vous avez une fonction de générateur que vous souhaitez utiliser, vous pouvez créer une instance de tf.data.Dataset à l'aide de l'API from_generator .

 mirrored_strategy = tf.distribute.MirroredStrategy()
def input_gen():
  while True:
    yield np.random.rand(4)

# use Dataset.from_generator
dataset = tf.data.Dataset.from_generator(
    input_gen, output_types=(tf.float32), output_shapes=tf.TensorShape([4]))
dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)
iterator = iter(dist_dataset)
for _ in range(4):
  mirrored_strategy.run(lambda x:x, args=(next(iterator),))
 
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)