Questa pagina è stata tradotta dall'API Cloud Translation.
Switch to English

Input distribuito

Visualizza su TensorFlow.org Esegui in Google Colab Visualizza sorgente su GitHub Scarica notebook

Le API tf.distribute forniscono agli utenti un modo semplice per scalare la formazione da una singola macchina a più macchine. Quando si ridimensiona il proprio modello, gli utenti devono anche distribuire i propri input su più dispositivi. tf.distribute fornisce API che consentono di distribuire automaticamente l'input tra i dispositivi.

Questa guida ti mostrerà i diversi modi in cui puoi creare set di dati distribuiti e iteratori utilizzando tf.distribute API tf.distribute . Inoltre, verranno trattati i seguenti argomenti:

Questa guida non copre l'utilizzo dell'input distribuito con le API di Keras.

Set di dati distribuiti

Per utilizzare tf.distribute API tf.distribute per la scalabilità, si consiglia agli utenti di utilizzare tf.data.Dataset per rappresentare il proprio input. tf.distribute è stato fatto per funzionare in modo efficiente con tf.data.Dataset (ad esempio, prelettura automatica dei dati su ciascun dispositivo acceleratore) con ottimizzazioni delle prestazioni regolarmente incorporate nell'implementazione. Se hai un caso d'uso per usare qualcosa di diverso da tf.data.Dataset , fai riferimento a una sezione successiva di questa guida. In un ciclo di addestramento non distribuito, gli utenti creano prima un'istanza tf.data.Dataset e quindi tf.data.Dataset gli elementi. Per esempio:

 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)

Per consentire agli utenti di utilizzare la strategia tf.distribute con modifiche minime al codice esistente di un utente, sono state introdotte due API che distribuiscono un'istanza tf.data.Dataset e restituiscono un oggetto dataset distribuito. Un utente potrebbe quindi eseguire l'iterazione su questa istanza del set di dati distribuito e addestrare il proprio modello come prima. Vediamo ora le due API - tf.distribute.Strategy.experimental_distribute_dataset e tf.distribute.Strategy.experimental_distribute_datasets_from_function in modo più dettagliato:

tf.distribute.Strategy.experimental_distribute_dataset

uso

Questa API accetta un'istanza tf.data.Dataset come input e restituisce un'istanza tf.distribute.DistributedDataset . È necessario eseguire il batch del set di dati di input con un valore uguale alla dimensione del batch globale. Questa dimensione globale del batch è il numero di campioni che si desidera elaborare su tutti i dispositivi in ​​1 passaggio. Puoi iterare su questo set di dati distribuito in modo pitonico o creare un iteratore usando iter . L'oggetto restituito non è un'istanza di tf.data.Dataset e non supporta altre API che trasformano o ispezionano il set di dati in alcun modo. Questa è l'API consigliata se non si dispone di modi specifici in cui si desidera suddividere l'input su repliche diverse.

 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)>)

Proprietà

Dosaggio

tf.distribute riassegna l'istanza di input tf.data.Dataset con una nuova dimensione del batch che è uguale alla dimensione del batch globale divisa per il numero di repliche sincronizzate. Il numero di repliche sincronizzate è uguale al numero di dispositivi che stanno prendendo parte al gradiente ridotto durante l'allenamento. Quando un utente chiama next sull'iteratore distribuito, su ogni replica viene restituita una dimensione batch di dati per replica. La cardinalità del set di dati rifiutato sarà sempre un multiplo del numero di repliche. Qui ci sono un paio di esempi:

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

    Senza distribuzione:

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

    Lotto 2: [4, 5]

    Con distribuzione su 2 repliche:

    Lotto 1: Replica 1: [0, 1] Replica 2: [2, 3]

    Lotto 2: Replica 2: [4] Replica 2: [5]

    L'ultimo batch ([4, 5]) è diviso tra 2 repliche.

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

    Senza distribuzione:

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

    Con distribuzione su 5 repliche:

    Batch 1: Replica 1: [0] Replica 2: [1] Replica 3: [2] Replica 4: [3] Replica 5: []

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

    Senza distribuzione:

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

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

    Con distribuzione su 3 repliche:

    Batch 1: Replica 1: [0, 1] Replica 2: [2, 3] Replica 3: []

    Lotto 2: Replica 1: [4, 5] Replica 2: [6, 7] Replica 3: []

Il rebatch del set di dati ha una complessità dello spazio che aumenta linearmente con il numero di repliche. Ciò significa che per il caso d'uso della formazione multi-lavoratore la pipeline di input può incorrere in errori OOM.

sharding

tf.distribute anche la registrazione automatica del set di dati di input nella formazione multi worker. Ogni set di dati viene creato sul dispositivo CPU del lavoratore. Il partizionamento automatico di un set di dati su un set di worker significa che a ciascun worker viene assegnato un sottoinsieme dell'intero set di dati (se è impostato il tf.data.experimental.AutoShardPolicy corretto). Questo per garantire che in ogni fase, una dimensione batch globale di elementi del set di dati non sovrapposti verrà elaborata da ciascun lavoratore. La suddivisione automatica ha un paio di diverse opzioni che possono essere specificate usando 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)
 

Sono disponibili tre diverse opzioni che è possibile impostare per tf.data.experimental.AutoShardPolicy :

  • AUTO: questa è l'opzione predefinita, il che significa che verrà fatto un tentativo di frammentazione da FILE. Il tentativo di frammentare con FILE fallisce se non viene rilevato un set di dati basato su file. tf.distribute tornerà quindi al sharding di DATA. Se il set di dati di input è basato su file ma il numero di file è inferiore al numero di lavoratori, verrà generato un errore.
  • FILE: questa è l'opzione se vuoi condividere i file di input su tutti i lavoratori. Se il numero di file è inferiore al numero di lavoratori, verrà generato un errore. È necessario utilizzare questa opzione se il numero di file di input è molto maggiore del numero di worker e i dati nei file sono distribuiti in modo uniforme. Il rovescio della medaglia di questa opzione è avere lavoratori inattivi se i dati nei file non sono distribuiti uniformemente. Ad esempio, distribuiamo 2 file su 2 worker con 1 replica ciascuno. Il file 1 contiene [0, 1, 2, 3, 4, 5] e il file 2 contiene [6, 7, 8, 9, 10, 11]. Lascia che il numero totale di repliche sincronizzate sia 2 e la dimensione del batch globale 4.

    • Lavoratore 0:

    Batch 1 = Replica 1: [0, 1]

    Lotto 2 = Replica 1: [2, 3]

    Batch 3 = Replica 1: [4]

    Batch 4 = Replica 1: [5]

    • Lavoratore 1:

    Batch 1 = Replica 2: [6, 7]

    Lotto 2 = Replica 2: [8, 9]

    Batch 3 = Replica 2: [10]

    Batch 4 = Replica 2: [11]

  • DATI: Questo autoshard gli elementi in tutti i lavoratori. Ciascuno dei lavoratori leggerà l'intero set di dati ed elaborerà solo il frammento assegnato. Tutti gli altri frammenti verranno scartati. Viene generalmente utilizzato se il numero di file di input è inferiore al numero di lavoratori e si desidera una migliore condivisione dei dati tra tutti i lavoratori. Il rovescio della medaglia è che l'intero set di dati verrà letto su ciascun lavoratore. Ad esempio, lasciamo distribuire 1 file su 2 lavoratori. Il file 1 contiene [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]. Lascia che il numero totale di repliche sincronizzate sia 2.

    • Lavoratore 0:

    Lotto 1 = Replica 1: [0, 1]

    Batch 2 = Replica 1: [4, 5]

    Batch 3 = Replica 1: [8, 9]

    • Lavoratore 1:

    Batch 1 = Replica 2: [2, 3]

    Lotto 2 = Replica 2: [6, 7]

    Batch 3 = Replica 2: [10, 11]

  • OFF: se disattivi l'autosharding, ogni worker elaborerà tutti i dati. Ad esempio, distribuiamo 1 file su 2 worker. Il file 1 contiene [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]. Lascia che il numero totale di repliche sincronizzate sia 2. Quindi ogni worker vedrà la seguente distribuzione:

    • Lavoratore 0:

    Lotto 1 = Replica 1: [0, 1]

    Lotto 2 = Replica 1: [2, 3]

    Batch 3 = Replica 1: [4, 5]

    Lotto 4 = Replica 1: [6, 7]

    Batch 5 = Replica 1: [8, 9]

    Batch 6 = Replica 1: [10, 11]

    • Lavoratore 1:

    Lotto 1 = Replica 2: [0, 1]

    Lotto 2 = Replica 2: [2, 3]

    Batch 3 = Replica 2: [4, 5]

    Batch 4 = Replica 2: [6, 7]

    Lotto 5 = Replica 2: [8, 9]

    Batch 6 = Replica 2: [10, 11]

prefetching

Per impostazione predefinita, tf.distribute aggiunge una trasformazione di prefetch alla fine dell'utente fornito tf.data.Dataset . L'argomento della trasformazione di prefetch che è buffer_size è uguale al numero di repliche sincronizzate.

tf.distribute.Strategy.experimental_distribute_datasets_from_function

uso

Questa API utilizza una funzione di input e restituisce un'istanza tf.distribute.DistributedDataset . La funzione di input che gli utenti passano ha un argomento tf.distribute.InputContext e dovrebbe restituire un'istanza tf.data.Dataset . Con questa API, tf.distribute non apporta ulteriori modifiche all'istanza tf.data.Dataset dell'utente restituita dalla funzione di input. È responsabilità dell'utente raggruppare e frammentare il set di dati. tf.distribute chiama la funzione di input sul dispositivo CPU di ciascuno dei lavoratori. Oltre a consentire agli utenti di specificare la propria logica di batch e tf.distribute.Strategy.experimental_distribute_dataset , questa API dimostra anche una migliore scalabilità e prestazioni rispetto a tf.distribute.Strategy.experimental_distribute_dataset quando viene utilizzata per la formazione multi worker.

 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',)

Proprietà

Dosaggio

L'istanza tf.data.Dataset che è il valore restituito della funzione di input deve essere raggruppata in batch utilizzando la dimensione batch per replica. La dimensione del batch per replica è la dimensione del batch globale divisa per il numero di repliche che partecipano alla formazione di sincronizzazione. Questo perché tf.distribute chiama la funzione di input sul dispositivo CPU di ciascuno dei worker. Il set di dati creato su un determinato worker dovrebbe essere pronto per essere utilizzato da tutte le repliche su quel worker.

sharding

L'oggetto tf.distribute.InputContext che viene passato implicitamente come argomento alla funzione di input dell'utente viene creato da tf.distribute sotto il cofano. Contiene informazioni sul numero di lavoratori, sull'ID lavoratore corrente ecc. Questa funzione di input può gestire il sharding secondo le politiche impostate dall'utente usando queste proprietà che fanno parte dell'oggetto tf.distribute.InputContext .

prefetching

tf.distribute non aggiunge una trasformazione di precaricamento alla fine del tf.data.Dataset restituito dalla funzione di input fornita dall'utente.

Iteratori distribuiti

Simile alle istanze tf.data.Dataset non distribuite, sarà necessario creare un iteratore sulle istanze tf.distribute.DistributedDataset per iterare su di esse e accedere agli elementi in tf.distribute.DistributedDataset . I seguenti sono i modi in cui puoi creare un tf.distribute.DistributedIterator e usarlo per addestrare il tuo modello:

usi

Usa un Pythonic for loop costrutto

È possibile utilizzare un loop Pythonic intuitivo per tf.distribute.DistributedDataset il tf.distribute.DistributedDataset . Gli elementi restituiti da tf.distribute.DistributedIterator possono essere un singolo tf.Tensor o un tf.distribute.DistributedValues che contiene un valore per replica. Posizionare il loop all'interno di una funzione tf.function aumenterà le prestazioni. Tuttavia, break e return sono attualmente supportati se il ciclo è posto all'interno di una funzione tf.function .. Inoltre, non supportiamo l'inserimento del loop all'interno di una funzione tf.function quando si utilizzano strategie multi-lavoratore come tf.distribute.experimental.MultiWorkerMirroredStrategy e tf.distribute.TPUStrategy . Posizionamento del ciclo all'interno tf.function lavori per singolo lavoratore tf.distribute.TPUStrategy ma non quando si utilizza cialde 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)

Usa iter per creare un iteratore esplicito

Per tf.distribute.DistributedDataset gli elementi in un'istanza tf.distribute.DistributedDataset , puoi creare un tf.distribute.DistributedIterator utilizzando l'API iter su di esso. Con un iteratore esplicito, puoi iterare per un numero fisso di passaggi. Al fine di ottenere il prossimo elemento da un tf.distribute.DistributedIterator esempio dist_iterator , è possibile chiamare next(dist_iterator) , dist_iterator.get_next() , o dist_iterator.get_next_as_optional() . I primi due sono essenzialmente gli stessi:

 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)

Con next() o tf.distribute.DistributedIterator.get_next() , se tf.distribute.DistributedIterator ha raggiunto la fine, verrà generato un errore OutOfRange. Il client può rilevare l'errore sul lato python e continuare a svolgere altri lavori come il checkpoint e la valutazione. Tuttavia, questo non funzionerà se stai usando un ciclo di addestramento host (cioè, esegui più passaggi per funzione tf.function ), che assomiglia a:

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

train_fn contiene più passaggi avvolgendo il corpo del passaggio all'interno di un tf.range . In questo caso, diverse iterazioni nel ciclo senza dipendenza potrebbero iniziare in parallelo, quindi un errore OutOfRange può essere attivato in iterazioni successive prima che il calcolo delle iterazioni precedenti sia terminato. Una volta generato un errore OutOfRange, tutte le operazioni nella funzione verranno immediatamente interrotte. Se questo è un caso che si desidera evitare, un'alternativa che non genera un errore tf.distribute.DistributedIterator.get_next_as_optional() è tf.distribute.DistributedIterator.get_next_as_optional() . get_next_as_optional restituisce un tf.experimental.Optional che contiene l'elemento successivo o nessun valore se tf.distribute.DistributedIterator ha raggiunto la fine.

 # 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], [])

Utilizzando la proprietà element_spec

Se si passano gli elementi di un set di dati distribuito a una funzione tf.function e si desidera una garanzia tf.TypeSpec , è possibile specificare l'argomento input_signature della funzione tf.function . L'output di un set di dati distribuito è tf.distribute.DistributedValues che può rappresentare l'input per un singolo dispositivo o più dispositivi. Per ottenere il tf.TypeSpec corrispondente a questo valore distribuito è possibile utilizzare la proprietà element_spec del set di dati distribuito o dell'oggetto iteratore distribuito.

 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]])

Lotti parziali

I batch parziali vengono rilevati quando tf.data.Dataset istanze tf.data.Dataset create dagli utenti possono contenere dimensioni batch non divisibili in modo uniforme per il numero di repliche o quando la cardinalità dell'istanza del set di dati non è divisibile per la dimensione del batch. Ciò significa che quando il set di dati viene distribuito su più repliche, la chiamata next su alcuni iteratori comporterà un OutOfRangeError. Per gestire questo caso d'uso, tf.distribute restituisce batch fittizi di dimensione batch 0 sulle repliche che non hanno più dati da elaborare.

Per il singolo caso di lavoro, se i dati non vengono restituiti dalla chiamata next sull'iteratore, vengono creati e utilizzati batch fittizi di dimensione batch 0 insieme ai dati reali nel set di dati. Nel caso di batch parziali, l'ultimo batch globale di dati conterrà dati reali insieme a batch fittizi di dati. La condizione di arresto per l'elaborazione dei dati ora controlla se una delle repliche ha dati. Se non sono presenti dati su nessuna delle repliche, viene generato un errore OutOfRange.

Per il caso multi worker, il valore booleano che rappresenta la presenza di dati su ciascuno dei worker viene aggregato utilizzando la comunicazione cross replica e questo viene utilizzato per identificare se tutti i worker hanno terminato l'elaborazione del dataset distribuito. Poiché ciò implica la comunicazione tra i lavoratori, è implicata una penalità di prestazione.

Avvertenze

  • Quando si utilizzano tf.distribute.Strategy.experimental_distribute_dataset API tf.distribute.Strategy.experimental_distribute_dataset con un'impostazione a più lavoratori, gli utenti passano un tf.data.Dataset che legge dai file. Se tf.data.experimental.AutoShardPolicy è impostato su AUTO o FILE , la dimensione batch effettiva per fase potrebbe essere inferiore alla dimensione batch globale definita dall'utente. Ciò può accadere quando gli elementi rimanenti nel file sono inferiori alla dimensione batch globale. Gli utenti possono esaurire il set di dati senza dipendere dal numero di passaggi da eseguire o impostare tf.data.experimental.AutoShardPolicy su DATA per tf.data.experimental.AutoShardPolicy .

  • Le trasformazioni del set di dati con stato non sono attualmente supportate con tf.distribute e tutte le operazioni con stato che il set di dati potrebbe avere sono attualmente ignorate. Ad esempio, se il tuo set di dati ha un map_fn che utilizza tf.random.uniform per ruotare un'immagine, allora hai un grafico del set di dati che dipende dallo stato (cioè il seme casuale) sulla macchina locale in cui viene eseguito il processo python.

  • tf.data.experimental.OptimizationOptions sperimentale Le tf.data.experimental.OptimizationOptions disabilitate per impostazione predefinita possono in alcuni contesti, ad esempio se utilizzate insieme a tf.distribute , causare un peggioramento delle prestazioni. È necessario abilitarli solo dopo aver verificato che avvantaggiano le prestazioni del carico di lavoro in un'impostazione di distribuzione.

  • L'ordine in cui i dati vengono elaborati dai lavoratori quando si utilizza tf.distribute.experimental_distribute_dataset o tf.distribute.experimental_distribute_datasets_from_function non è garantito. Ciò è in genere necessario se si utilizza tf.distribute per scalare la previsione. È tuttavia possibile inserire un indice per ogni elemento nel batch e ordinare gli output di conseguenza. Il frammento di codice seguente è un esempio di come ordinare gli output.

 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}

Come distribuisco i miei dati se non utilizzo un'istanza canonica tf.data.Dataset?

A volte gli utenti non possono utilizzare un tf.data.Dataset per rappresentare il proprio input e successivamente le API sopra menzionate per distribuire il set di dati su più dispositivi. In tali casi è possibile utilizzare tensori grezzi o input da un generatore.

Utilizzare la funzione sperimentale_distributo_valori_da_per ingressi tensore arbitrari

strategy.run accetta tf.distribute.DistributedValues che è l'output di next(iterator) . Per passare i valori dei tensori, usa experimental_distribute_values_from_function per costruire tf.distribute.DistributedValues dai tensori grezzi.

 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)

Usa tf.data.Dataset.from_generator se il tuo input proviene da un generatore

Se si dispone di una funzione di generatore che si desidera utilizzare, è possibile creare un'istanza tf.data.Dataset utilizzando 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',)