 Lihat di TensorFlow.org |  Jalankan di Google Colab |  Lihat sumber di GitHub |  Unduh buku catatan |
API tf.distribute menyediakan cara mudah bagi pengguna untuk menskalakan pelatihan mereka dari satu mesin ke beberapa mesin. Saat menskalakan model mereka, pengguna juga harus mendistribusikan input mereka di beberapa perangkat. tf.distribute menyediakan API yang dapat digunakan untuk mendistribusikan input Anda secara otomatis ke seluruh perangkat.
Panduan ini akan menunjukkan kepada Anda berbagai cara di mana Anda dapat membuat dataset dan iterator tf.distribute menggunakan API tf.distribute . Selain itu, topik berikut akan dibahas:
- Opsi penggunaan, sharding dan batching ketika menggunakan
tf.distribute.Strategy.experimental_distribute_datasetdantf.distribute.Strategy.experimental_distribute_datasets_from_function. - Berbagai cara untuk melakukan iterasi atas kumpulan data terdistribusi.
- Perbedaan antara
tf.distribute.Strategy.experimental_distribute_dataset/tf.distribute.Strategy.experimental_distribute_datasets_from_functionAPIs dantf.dataAPIs juga segala batasan yang mungkin ditemukan pengguna dalam penggunaannya.
Panduan ini tidak mencakup penggunaan input yang didistribusikan dengan API Keras.
Set Data Terdistribusi
Untuk menggunakan API tf.distribute untuk menskala, disarankan agar pengguna menggunakan tf.data.Dataset untuk mewakili input mereka. tf.distribute telah dibuat untuk bekerja secara efisien dengan tf.data.Dataset (misalnya, prefetch data otomatis ke setiap perangkat akselerator) dengan optimalisasi kinerja yang secara teratur dimasukkan ke dalam implementasi. Jika Anda memiliki use case untuk menggunakan sesuatu selain tf.data.Dataset , silakan lihat bagian selanjutnya dalam panduan ini. Dalam loop pelatihan yang tidak terdistribusi, pengguna pertama kali membuat instance tf.data.Dataset dan kemudian beralih ke elemen-elemen. Sebagai contoh:
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)
Untuk memungkinkan pengguna menggunakan strategi tf.distribute dengan perubahan minimal pada kode yang ada pengguna, dua API diperkenalkan yang akan mendistribusikan instance tf.data.Dataset dan mengembalikan objek dataset terdistribusi. Seorang pengguna kemudian dapat beralih dari instance dataset terdistribusi ini dan melatih model mereka seperti sebelumnya. Mari kita lihat dua API - tf.distribute.Strategy.experimental_distribute_dataset dan tf.distribute.Strategy.experimental_distribute_datasets_from_function secara lebih rinci:
tf.distribute.Strategy.experimental_distribute_dataset
Pemakaian
API ini mengambil instance tf.data.Dataset sebagai input dan tf.distribute.DistributedDataset instance tf.distribute.DistributedDataset . Anda harus mengumpulkan kumpulan data masukan dengan nilai yang sama dengan ukuran kumpulan global. Ukuran tumpukan global ini adalah jumlah sampel yang ingin Anda proses di semua perangkat dalam 1 langkah. Anda dapat melakukan iterasi atas kumpulan data terdistribusi ini dengan cara Pythonic atau membuat iterator menggunakan iter . Objek yang dikembalikan bukan turunan tf.data.Dataset dan tidak mendukung API lain yang mengubah atau memeriksa dataset dengan cara apa pun. Ini adalah API yang disarankan jika Anda tidak memiliki cara spesifik di mana Anda ingin membuang input Anda pada replika yang berbeda.
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)>)
Properti
Batching
tf.distribute mengambil kembali contoh input tf.data.Dataset dengan ukuran batch baru yang sama dengan ukuran batch global dibagi dengan jumlah replika yang disinkronkan. Jumlah replika dalam sinkronisasi sama dengan jumlah perangkat yang mengambil bagian dalam gradien allreduce selama pelatihan. Saat pengguna memanggil next pada iterator terdistribusi, ukuran kumpulan data per replika dikembalikan pada setiap replika. Kardinalitas set data ulang akan selalu berupa kelipatan dari jumlah replika. Berikut ini beberapa contohnya:
tf.data.Dataset.range(6).batch(4, drop_remainder=False)Tanpa distribusi:
Gelombang 1: [0, 1, 2, 3]
Gelombang 2: [4, 5]
Dengan distribusi lebih dari 2 replika:
Tumpukan 1: Replika 1: [0, 1] Replika 2: [2, 3]
Tumpukan 2: Replika 2: [4] Replika 2: [5]
Batch terakhir ([4, 5]) dibagi antara 2 replika.
tf.data.Dataset.range(4).batch(4)Tanpa distribusi:
Tumpukan 1: [[0], [1], [2], [3]]
Dengan distribusi lebih dari 5 replika:
Gelombang 1: Replika 1: [0] Replika 2: [1] Replika 3: [2] Replika 4: [3] Replika 5: []
tf.data.Dataset.range(8).batch(4)Tanpa distribusi:
Gelombang 1: [0, 1, 2, 3]
Gelombang 2: [4, 5, 6, 7]
Dengan distribusi lebih dari 3 replika:
Tumpukan 1: Replika 1: [0, 1] Replika 2: [2, 3] Replika 3: []
Tumpukan 2: Replika 1: [4, 5] Replika 2: [6, 7] Replika 3: []
Pengambilan ulang dataset memiliki kompleksitas ruang yang meningkat secara linear dengan jumlah replika. Ini berarti bahwa untuk kasus penggunaan pelatihan multi-pekerja, pipa input dapat mengalami kesalahan OOM.
Sharding
tf.distribute juga tf.distribute dataset input dalam pelatihan multi-pekerja. Setiap dataset dibuat pada perangkat CPU pekerja. Melakukan autosharding pada set data pada satu set pekerja berarti bahwa setiap pekerja diberi subset dari seluruh set data (jika tf.data.experimental.AutoShardPolicy benar disetel). Ini untuk memastikan bahwa di setiap langkah, ukuran batch global dari elemen set data yang tidak tumpang tindih akan diproses oleh setiap pekerja. Autosharding memiliki beberapa opsi berbeda yang dapat ditentukan menggunakan 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)
Ada tiga opsi berbeda yang dapat Anda tetapkan untuk tf.data.experimental.AutoShardPolicy :
- OTOMATIS: Ini adalah opsi default yang berarti upaya akan dilakukan untuk memecah belah oleh FILE. Upaya untuk beling oleh FILE gagal jika dataset berbasis file tidak terdeteksi.
tf.distributekemudian akan kembali ke sharding oleh DATA. Perhatikan bahwa jika set data masukan adalah berbasis file tetapi jumlah file kurang dari jumlah pekerja, kesalahan akan muncul. FILE: Ini adalah opsi jika Anda ingin membagi file input ke semua pekerja. Jika jumlah file kurang dari jumlah pekerja akan ada kesalahan yang muncul. Anda harus menggunakan opsi ini jika jumlah file input jauh lebih besar daripada jumlah pekerja dan data dalam file didistribusikan secara merata. Kelemahan dari opsi ini adalah memiliki pekerja yang menganggur jika data dalam file tidak didistribusikan secara merata. Misalnya, mari kita mendistribusikan 2 file ke 2 pekerja dengan masing-masing 1 replika. File 1 berisi [0, 1, 2, 3, 4, 5] dan File 2 berisi [6, 7, 8, 9, 10, 11]. Biarkan jumlah total replika dalam sinkronisasi menjadi 2 dan ukuran kumpulan global menjadi 4.
- Pekerja 0:
Tumpukan 1 = Replika 1: [0, 1]
Tumpukan 2 = Replika 1: [2, 3]
Gelombang 3 = Replika 1: [4]
Gelombang 4 = Replika 1: [5]
- Pekerja 1:
Gelombang 1 = Replika 2: [6, 7]
Tumpukan 2 = Replika 2: [8, 9]
Gelombang 3 = Replika 2: [10]
Gelombang 4 = Replika 2: [11]
DATA: Ini akan menghapus otomatis elemen di semua pekerja. Setiap pekerja akan membaca seluruh kumpulan data dan hanya memproses pecahan yang ditetapkan untuk itu. Semua pecahan lain akan dibuang. Ini umumnya digunakan jika jumlah file input kurang dari jumlah pekerja dan Anda ingin sharding data yang lebih baik di semua pekerja. Kelemahannya adalah bahwa seluruh dataset akan dibaca pada setiap pekerja. Misalnya, mari kita mendistribusikan 1 file ke 2 pekerja. File 1 berisi [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]. Biarkan jumlah total replika yang disinkronkan menjadi 2.
- Pekerja 0:
Gelombang 1 = Replika 1: [0, 1]
Gelombang 2 = Replika 1: [4, 5]
Angkatan 3 = Replika 1: [8, 9]
- Pekerja 1:
Gelombang 1 = Replika 2: [2, 3]
Tumpukan 2 = Replika 2: [6, 7]
Gelombang 3 = Replika 2: [10, 11]
OFF: Jika Anda mematikan autosharding, setiap pekerja akan memproses semua data. Misalnya, mari kita mendistribusikan 1 file ke 2 pekerja. File 1 berisi [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]. Biarkan jumlah total replika yang disinkronkan menjadi 2. Kemudian setiap pekerja akan melihat distribusi berikut:
- Pekerja 0:
Gelombang 1 = Replika 1: [0, 1]
Tumpukan 2 = Replika 1: [2, 3]
Gelombang 3 = Replika 1: [4, 5]
Gelombang 4 = Replika 1: [6, 7]
Gelombang 5 = Replika 1: [8, 9]
Gelombang 6 = Replika 1: [10, 11]
- Pekerja 1:
Tumpukan 1 = Replika 2: [0, 1]
Tumpukan 2 = Replika 2: [2, 3]
Tumpukan 3 = Replika 2: [4, 5]
Gelombang 4 = Replika 2: [6, 7]
Tumpukan 5 = Replika 2: [8, 9]
Angkatan 6 = Replika 2: [10, 11]
Mengambil lebih dulu
Secara default, tf.distribute menambahkan transformasi prefetch di akhir instance tf.data.Dataset disediakan pengguna. Argumen untuk transformasi prefetch yaitu buffer_size sama dengan jumlah replika yang disinkronkan.
tf.distribute.Strategy.experimental_distribute_datasets_from_function
Pemakaian
API ini mengambil fungsi input dan mengembalikan instance tf.distribute.DistributedDataset . Fungsi input yang dilewati pengguna memiliki argumen tf.distribute.InputContext dan harus mengembalikan instance tf.data.Dataset . Dengan API ini, tf.distribute tidak membuat perubahan lebih lanjut apa pun pada instance tf.data.Dataset pengguna yang dikembalikan dari fungsi input. Ini adalah tanggung jawab pengguna untuk mengumpulkan dan membuang dataset. tf.distribute memanggil fungsi input pada perangkat CPU masing-masing pekerja. Selain memungkinkan pengguna untuk menentukan logika batching dan sharding mereka sendiri, API ini juga menunjukkan skalabilitas dan performa yang lebih baik dibandingkan dengan tf.distribute.Strategy.experimental_distribute_dataset saat digunakan untuk pelatihan multi pekerja.
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',)
Properti
Batching
tf.data.Dataset yang merupakan nilai kembalian dari fungsi input harus di-batch menggunakan ukuran batch per replika. Ukuran tumpukan per replika adalah ukuran tumpukan global dibagi dengan jumlah replika yang mengambil bagian dalam pelatihan sinkronisasi. Ini karena tf.distribute memanggil fungsi input pada perangkat CPU masing-masing pekerja. Kumpulan data yang dibuat pada pekerja tertentu harus siap digunakan oleh semua replika pekerja tersebut.
Sharding
Objek tf.distribute.InputContext yang secara implisit diteruskan sebagai argumen ke fungsi masukan pengguna dibuat oleh tf.distribute bawah tenda. Ini memiliki informasi tentang jumlah pekerja, id pekerja saat ini dll. Fungsi input ini dapat menangani sharding sesuai kebijakan yang ditetapkan oleh pengguna menggunakan properti ini yang merupakan bagian dari objek tf.distribute.InputContext .
Prefetching
tf.distribute tidak menambahkan transformasi prefetch di akhir tf.data.Dataset dikembalikan oleh fungsi input yang disediakan pengguna.
Iterator yang didistribusikan
Mirip dengan instance tf.data.Dataset terdistribusi, Anda harus membuat iterator pada instance tf.distribute.DistributedDataset untuk beralih di atasnya dan mengakses elemen-elemen dalam tf.distribute.DistributedDataset . Berikut ini adalah cara-cara untuk membuat tf.distribute.DistributedIterator dan menggunakannya untuk melatih model Anda:
Penggunaan
Gunakan Pythonic untuk konstruksi loop
Anda bisa menggunakan loop Pythonic yang mudah digunakan untuk beralih di tf.distribute.DistributedDataset . Elemen yang dikembalikan dari tf.distribute.DistributedIterator dapat berupa satu tf.Tensor atau tf.distribute.DistributedValues yang berisi nilai per replika. Menempatkan loop di dalam fungsi tf.function akan memberikan peningkatan kinerja. Namun break dan return saat ini tidak didukung jika loop ditempatkan di dalam fungsi tf.function . Kami juga tidak mendukung penempatan loop di dalam tf.function saat menggunakan strategi multi-pekerja seperti tf.distribute.experimental.MultiWorkerMirroredStrategy dan tf.distribute.TPUStrategy . Menempatkan loop di dalam tf.function berfungsi untuk pekerja tunggal tf.distribute.TPUStrategy tetapi tidak ketika menggunakan pod 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)
Gunakan iter untuk membuat iterator eksplisit
Untuk beralih ke elemen dalam contoh tf.distribute.DistributedDataset , Anda bisa membuat tf.distribute.DistributedIterator menggunakan iter API di atasnya. Dengan iterator eksplisit, Anda dapat melakukan iterate untuk sejumlah langkah tertentu. Dalam rangka untuk mendapatkan elemen berikutnya dari tf.distribute.DistributedIterator contoh dist_iterator , Anda dapat menghubungi next(dist_iterator) , dist_iterator.get_next() , atau dist_iterator.get_next_as_optional() . Dua yang sebelumnya pada dasarnya sama:
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)
Dengan next() atau tf.distribute.DistributedIterator.get_next() , jika tf.distribute.DistributedIterator telah mencapai akhirnya, kesalahan OutOfRange akan dilemparkan. Klien dapat menangkap kesalahan di sisi python dan terus melakukan pekerjaan lain seperti checkpointing dan evaluasi. Namun, ini tidak akan berfungsi jika Anda menggunakan loop pelatihan host (mis. Jalankan beberapa langkah per tf.function ), yang terlihat seperti:
@tf.function
def train_fn(iterator):
for _ in tf.range(steps_per_loop):
strategy.run(step_fn, args=(next(iterator),))
train_fn berisi beberapa langkah dengan membungkus badan langkah di dalam tf.range . Dalam kasus ini, iterasi yang berbeda dalam loop tanpa ketergantungan dapat dimulai secara paralel, sehingga kesalahan OutOfRange dapat dipicu di iterasi selanjutnya sebelum komputasi iterasi sebelumnya selesai. Setelah kesalahan OutOfRange dilempar, semua ops dalam fungsi akan segera dihentikan. Jika ini adalah beberapa kasus yang ingin Anda hindari, alternatif yang tidak melemparkan kesalahan tf.distribute.DistributedIterator.get_next_as_optional() adalah tf.distribute.DistributedIterator.get_next_as_optional() . get_next_as_optional mengembalikan tf.experimental.Optional yang berisi elemen berikutnya atau tidak ada nilai jika tf.distribute.DistributedIterator telah mencapai akhir.
# 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], [])
Menggunakan properti element_spec
Jika Anda melewati elemen dari dataset didistribusikan ke tf.function dan ingin tf.TypeSpec jaminan, Anda dapat menentukan input_signature argumen dari tf.function . Output dari dataset terdistribusi adalah tf.distribute.DistributedValues yang dapat mewakili input ke satu perangkat atau beberapa perangkat. Untuk mendapatkan tf.TypeSpec sesuai dengan nilai terdistribusi ini, Anda dapat menggunakan properti element_spec dari kumpulan data terdistribusi atau objek iterator terdistribusi.
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]])
Batch parsial
Batch parsial dijumpai saat instance tf.data.Dataset yang dibuat pengguna mungkin berisi ukuran batch yang tidak dapat dibagi secara merata dengan jumlah replika atau jika kardinalitas dari instance tf.data.Dataset data tidak dapat dibagi oleh ukuran batch. Ini berarti bahwa ketika kumpulan data didistribusikan ke beberapa replika, panggilan next pada beberapa iterator akan menghasilkan OutOfRangeError. Untuk menangani kasus penggunaan ini, tf.distribute mengembalikan batch dummy ukuran batch 0 pada replika yang tidak memiliki data lagi untuk diproses.
Untuk kasus pekerja tunggal, jika data tidak dikembalikan oleh panggilan next pada iterator, kumpulan dummy 0 ukuran batch dibuat dan digunakan bersama dengan data nyata dalam kumpulan data. Dalam kasus batch parsial, kumpulan data global terakhir akan berisi data nyata di samping kumpulan data dummy. Kondisi berhenti untuk memproses data sekarang memeriksa apakah ada replika yang memiliki data. Jika tidak ada data di salah satu replika, kesalahan OutOfRange dilemparkan.
Untuk kasus multi-pekerja, nilai boolean yang mewakili keberadaan data pada masing-masing pekerja dikumpulkan menggunakan komunikasi lintas replika dan ini digunakan untuk mengidentifikasi apakah semua pekerja telah selesai memproses dataset yang didistribusikan. Karena ini melibatkan komunikasi lintas pekerja, ada beberapa hukuman kinerja yang terlibat.
Peringatan
Saat menggunakan API
tf.distribute.Strategy.experimental_distribute_datasetdengan pengaturan banyak pekerja, pengguna meneruskantf.data.Datasetyang membaca dari file. Jikatf.data.experimental.AutoShardPolicydiatur keAUTOatauFILE, ukuran batch per langkah yang sebenarnya mungkin lebih kecil dari ukuran batch global yang ditentukan pengguna. Ini bisa terjadi jika elemen yang tersisa di file lebih kecil dari ukuran batch global. Pengguna dapat menghabiskan kumpulan data tanpa bergantung pada jumlah langkah untuk menjalankan atau menyeteltf.data.experimental.AutoShardPolicykeDATAuntuk mengatasinya.Transformasi
tf.distributedata stateful saat ini tidak didukung dengantf.distributedan operasi stateful apa pun yang mungkin dimiliki dataset saat ini diabaikan. Misalnya, jika kumpulan data Anda memilikimap_fnyang menggunakantf.random.uniformuntuk memutar gambar, maka Anda memiliki grafik kumpulan data yang bergantung pada status (yaitu benih acak) pada mesin lokal tempat proses python dijalankan.Eksperimental
tf.data.experimental.OptimizationOptionsyang dinonaktifkan secara default dapat dalam konteks tertentu - seperti ketika digunakan bersama dengantf.distribute- menyebabkan penurunan kinerja. Anda hanya boleh mengaktifkannya setelah Anda memvalidasi bahwa mereka menguntungkan kinerja beban kerja Anda dalam pengaturan distribusi.Urutan pemrosesan data oleh pekerja saat menggunakan
tf.distribute.experimental_distribute_datasetatautf.distribute.experimental_distribute_datasets_from_functiontidak dijamin. Ini biasanya diperlukan jika Anda menggunakantf.distributeto scale prediksi. Namun Anda dapat memasukkan indeks untuk setiap elemen dalam batch dan memesan output yang sesuai. Cuplikan berikut adalah contoh cara memesan 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}
Bagaimana cara mendistribusikan data saya jika saya tidak menggunakan instance canonical tf.data.Dataset?
Kadang-kadang pengguna tidak dapat menggunakan tf.data.Dataset untuk mewakili input mereka dan selanjutnya API yang disebutkan di atas untuk mendistribusikan dataset ke beberapa perangkat. Dalam kasus seperti itu, Anda dapat menggunakan tensor atau input mentah dari generator.
Gunakan fungsi_distribusi_perimental untuk input tensor arbitrer
strategy.run menerima tf.distribute.DistributedValues yang merupakan output dari next(iterator) . Untuk meneruskan nilai tensor, gunakan experimental_distribute_values_from_function untuk membuat tf.distribute.DistributedValues dari tensor mentah.
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)
Gunakan tf.data.Dataset.from_generator jika input Anda dari generator
Jika Anda memiliki fungsi generator yang ingin Anda gunakan, Anda bisa membuat turunan tf.data.Dataset menggunakan from_generator API.
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',)