Szkolenie rozproszone z TensorFlow

Zobacz na TensorFlow.org Uruchom w Google Colab Wyświetl źródło na GitHub Pobierz notatnik

Przegląd

tf.distribute.Strategy jest API TensorFlow rozpowszechniać szkolenia w wielu procesorów graficznych, wielu maszynach lub TPU. Korzystając z tego interfejsu API, możesz rozpowszechniać istniejące modele i kod szkoleniowy z minimalnymi zmianami w kodzie.

tf.distribute.Strategy został zaprojektowany z tych kluczowych celów na uwadze:

  • Łatwy w użyciu i wspiera wiele segmentów użytkowników, w tym badaczy, inżynierów ML itp.
  • Zapewnij dobrą wydajność po wyjęciu z pudełka.
  • Łatwe przełączanie między strategiami.

tf.distribute.Strategy można stosować API wysokiego poziomu jak Keras i może być również wykorzystywane do rozprowadzania pętli własnych treningów (a na ogół żadnych obliczeń za pomocą TensorFlow).

W TensorFlow 2.x, można wykonać programów chętnie, albo w postaci wykresu za pomocą tf.function . tf.distribute.Strategy zamierza wspierać oba te sposoby realizacji, ale działa najlepiej z tf.function . Tryb chętny jest zalecane tylko dla celów debugowania i nie jest obsługiwany przez TPUStrategy . Chociaż szkolenie jest głównym tematem tego przewodnika, ten interfejs API może być również używany do rozpowszechniania oceny i prognozowania na różnych platformach.

Można użyć tf.distribute.Strategy z niewielu zmian w kodzie, ponieważ składniki bazowe TensorFlow zostały zmienione, by stać się świadoma strategia. Obejmuje to zmienne, warstwy, modele, optymalizatory, metryki, podsumowania i punkty kontrolne.

W tym przewodniku dowiesz się o różnych rodzajach strategii oraz o tym, jak możesz z nich korzystać w różnych sytuacjach. Aby dowiedzieć się, jak do debugowania problemów z wydajnością, zobacz wydajność Optymalizacja TensorFlow GPU instrukcji.

# Import TensorFlow
import tensorflow as tf
2021-07-14 01:23:18.251555: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcudart.so.11.0

Rodzaje strategii

tf.distribute.Strategy zamierza objąć liczbę przypadków użycia wzdłuż różnych osi. Niektóre z tych kombinacji są obecnie obsługiwane, a inne zostaną dodane w przyszłości. Niektóre z tych osi to:

  • Synchroniczna vs szkolenia asynchronicznego: Są dwa typowe sposoby dystrybucji szkolenia z równoległości danych. Podczas szkolenia synchronizacji wszyscy pracownicy szkolą się na różnych wycinkach danych wejściowych w synchronizacji i agregują gradienty na każdym kroku. W szkoleniu asynchronicznym wszyscy pracownicy niezależnie trenują dane wejściowe i asynchronicznie aktualizują zmienne. Zazwyczaj uczenie synchronizacji jest obsługiwane za pośrednictwem architektury all-reduce i async za pośrednictwem architektury serwera parametrów.
  • Platforma sprzętowa: Możesz skalować swój trening na wielu GPU na jednym komputerze lub wielu komputerach w sieci (z 0 lub więcej GPU każda) lub w chmurze TPU.

W celu wsparcia tych przypadków użycia dostępnych jest 6 strategii. W następnej sekcji wyjaśniono, które z nich są obsługiwane w jakich scenariuszach w TensorFlow. Oto krótki przegląd:

Szkolenia API Strategia lustrzana Strategia TPU MultiWorkerMirroredStrategy Centralna strategia magazynowania ParametrServerStrategy
Keras API Utrzymany Utrzymany Utrzymany Wsparcie eksperymentalne Obsługiwany planowany post 2.4
Niestandardowa pętla treningowa Utrzymany Utrzymany Utrzymany Wsparcie eksperymentalne Wsparcie eksperymentalne
Interfejs API estymatora Ograniczone wsparcie Niewspierany Ograniczone wsparcie Ograniczone wsparcie Ograniczone wsparcie

Strategia lustrzana

tf.distribute.MirroredStrategy obsługuje synchroniczne rozprowadzane szkolenia na wielu GPU na jednej maszynie. Tworzy jedną replikę na urządzenie GPU. Każda zmienna w modelu jest odzwierciedlona we wszystkich replikach. Wszystkie te zmienne tworzą jeden koncepcyjny zmienną o nazwie MirroredVariable . Te zmienne są zsynchronizowane ze sobą dzięki zastosowaniu identycznych aktualizacji.

Wydajne algorytmy all-reduce są używane do przekazywania aktualizacji zmiennych na urządzeniach. All-reduce agreguje tensory na wszystkich urządzeniach, sumując je i udostępnia na każdym urządzeniu. Jest to połączony algorytm, który jest bardzo wydajny i może znacznie zmniejszyć obciążenie synchronizacji. Dostępnych jest wiele algorytmów all-reduce i implementacji, w zależności od rodzaju komunikacji dostępnej między urządzeniami. Domyślnie korzysta z komunikacji zbiorowej Biblioteka NVIDIA ( NCCL ) jak wdrożenie wszystkie ograniczenia. Możesz wybrać jedną z kilku innych opcji lub napisać własną.

Oto najprostszy sposób tworzenia MirroredStrategy :

mirrored_strategy = tf.distribute.MirroredStrategy()
WARNING:tensorflow:Collective ops is not configured at program startup. Some performance features may not be enabled.
2021-07-14 01:23:19.530023: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcuda.so.1
2021-07-14 01:23:20.213426: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:937] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2021-07-14 01:23:20.214078: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1733] Found device 0 with properties: 
pciBusID: 0000:00:05.0 name: NVIDIA Tesla V100-SXM2-16GB computeCapability: 7.0
coreClock: 1.53GHz coreCount: 80 deviceMemorySize: 15.78GiB deviceMemoryBandwidth: 836.37GiB/s
2021-07-14 01:23:20.214108: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcudart.so.11.0
2021-07-14 01:23:20.217733: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcublas.so.11
2021-07-14 01:23:20.217820: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcublasLt.so.11
2021-07-14 01:23:20.218871: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcufft.so.10
2021-07-14 01:23:20.219205: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcurand.so.10
2021-07-14 01:23:20.220324: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcusolver.so.11
2021-07-14 01:23:20.221308: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcusparse.so.11
2021-07-14 01:23:20.221482: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcudnn.so.8
2021-07-14 01:23:20.221579: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:937] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2021-07-14 01:23:20.222279: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:937] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2021-07-14 01:23:20.222966: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1871] Adding visible gpu devices: 0
2021-07-14 01:23:20.224084: I tensorflow/core/platform/cpu_feature_guard.cc:142] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations:  AVX2 AVX512F FMA
To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.
2021-07-14 01:23:20.224666: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:937] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2021-07-14 01:23:20.225322: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1733] Found device 0 with properties: 
pciBusID: 0000:00:05.0 name: NVIDIA Tesla V100-SXM2-16GB computeCapability: 7.0
coreClock: 1.53GHz coreCount: 80 deviceMemorySize: 15.78GiB deviceMemoryBandwidth: 836.37GiB/s
2021-07-14 01:23:20.225396: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:937] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2021-07-14 01:23:20.226008: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:937] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2021-07-14 01:23:20.226671: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1871] Adding visible gpu devices: 0
2021-07-14 01:23:20.226715: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcudart.so.11.0
2021-07-14 01:23:20.843669: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1258] Device interconnect StreamExecutor with strength 1 edge matrix:
2021-07-14 01:23:20.843707: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1264]      0 
2021-07-14 01:23:20.843716: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1277] 0:   N 
2021-07-14 01:23:20.843925: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:937] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2021-07-14 01:23:20.844616: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:937] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2021-07-14 01:23:20.845280: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:937] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2021-07-14 01:23:20.845985: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1418] Created TensorFlow device (/job:localhost/replica:0/task:0/device:GPU:0 with 14646 MB memory) -> physical GPU (device: 0, name: NVIDIA Tesla V100-SXM2-16GB, pci bus id: 0000:00:05.0, compute capability: 7.0)
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)

To utworzy MirroredStrategy przykład, która będzie wykorzystywać wszystkie karty graficzne, które są widoczne na TensorFlow i NCCL, jak komunikat różnych urządzeniach.

Jeśli chcesz używać tylko niektórych procesorów graficznych na swoim komputerze, możesz to zrobić w następujący sposób:

mirrored_strategy = tf.distribute.MirroredStrategy(devices=["/gpu:0", "/gpu:1"])
WARNING:tensorflow:Some requested devices in `tf.distribute.Strategy` are not visible to TensorFlow: /job:localhost/replica:0/task:0/device:GPU:0,/job:localhost/replica:0/task:0/device:GPU:1
WARNING:tensorflow:Collective ops is not configured at program startup. Some performance features may not be enabled.
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0', '/job:localhost/replica:0/task:0/device:GPU:1')

Jeśli chcesz zastąpić krzyż komunikację urządzenia, możesz to zrobić za pomocą cross_device_ops argumentu dostarczając wystąpienie tf.distribute.CrossDeviceOps . Obecnie tf.distribute.HierarchicalCopyAllReduce i tf.distribute.ReductionToOneDevice są dwie opcje inne niż tf.distribute.NcclAllReduce , która jest wartością domyślną.

mirrored_strategy = tf.distribute.MirroredStrategy(
    cross_device_ops=tf.distribute.HierarchicalCopyAllReduce())
WARNING:tensorflow:Collective ops is not configured at program startup. Some performance features may not be enabled.
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)

Strategia TPU

tf.distribute.TPUStrategy pozwala uruchomić szkolenia TensorFlow na Tensor Processing Units (TPU). TPU to wyspecjalizowane układy ASIC firmy Google zaprojektowane w celu znacznego przyspieszenia obciążeń związanych z uczeniem maszynowym. Są one dostępne na Google Colab , z TPU Badań chmurze , a chmura TPU .

Pod względem architektury rozproszonej szkoleniowej TPUStrategy jest taki sam MirroredStrategy -To wdraża synchronicznych rozprowadzane szkolenie. TPU zapewniają własne wdrożenie efektywnych wszystkie ograniczenia i innych działań zbiorowych w wielu rdzeni TPU, które są wykorzystywane w TPUStrategy .

Oto w jaki sposób wystąpienia TPUStrategy :

cluster_resolver = tf.distribute.cluster_resolver.TPUClusterResolver(
    tpu=tpu_address)
tf.config.experimental_connect_to_cluster(cluster_resolver)
tf.tpu.experimental.initialize_tpu_system(cluster_resolver)
tpu_strategy = tf.distribute.TPUStrategy(cluster_resolver)

TPUClusterResolver instancja pomaga zlokalizować TPU. W Colab nie musisz podawać żadnych argumentów.

Jeśli chcesz użyć tego w przypadku Cloud TPU:

  • Musisz podać nazwę zasobu w TPU tpu argument.
  • Musisz zainicjować system TPU wyraźnie na początku programu. Jest to wymagane przed użyciem jednostek TPU do obliczeń. Inicjalizacja systemu tpu powoduje również wymazanie pamięci TPU, dlatego ważne jest, aby najpierw wykonać ten krok, aby uniknąć utraty stanu.

MultiWorkerMirroredStrategy

tf.distribute.MultiWorkerMirroredStrategy jest bardzo podobna do MirroredStrategy . Implementuje synchroniczne, rozproszone szkolenia dla wielu pracowników, z których każdy ma potencjalnie wiele procesorów graficznych. Podobny do tf.distribute.MirroredStrategy , tworzy kopie wszystkich zmiennych w modelu na każdym urządzeniu na wszystkich pracowników.

Oto najprostszy sposób tworzenia MultiWorkerMirroredStrategy :

strategy = tf.distribute.MultiWorkerMirroredStrategy()
WARNING:tensorflow:Collective ops is not configured at program startup. Some performance features may not be enabled.
INFO:tensorflow:Single-worker MultiWorkerMirroredStrategy with local_devices = ('/device:GPU:0',), communication = CommunicationImplementation.AUTO

MultiWorkerMirroredStrategy dwie implementacje dla komunikacji na różnych urządzeniach. CommunicationImplementation.RING jest RPC -na i obsługuje zarówno procesory i GPU. CommunicationImplementation.NCCL wykorzystuje NCCL i zapewnia state-of-art na wydajność GPU, ale nie obsługuje procesorów. CollectiveCommunication.AUTO odracza do wyboru Tensorflow. Możesz je określić w następujący sposób:

communication_options = tf.distribute.experimental.CommunicationOptions(
    implementation=tf.distribute.experimental.CommunicationImplementation.NCCL)
strategy = tf.distribute.MultiWorkerMirroredStrategy(
    communication_options=communication_options)
WARNING:tensorflow:Collective ops is not configured at program startup. Some performance features may not be enabled.
INFO:tensorflow:Single-worker MultiWorkerMirroredStrategy with local_devices = ('/device:GPU:0',), communication = CommunicationImplementation.NCCL

Jedną z kluczowych różnic w szkoleniu z wieloma pracownikami w porównaniu ze szkoleniem z wykorzystaniem wielu procesorów graficznych jest konfiguracja z wieloma pracownikami. TF_CONFIG zmienna jest standardowym sposobem w TensorFlow określić konfigurację klastra do każdego pracownika, który jest częścią klastra. Więcej informacji na temat konfigurowania TF_CONFIG .

ParametrServerStrategy

Uczenie serwera parametrów to powszechna metoda równoległego przetwarzania danych służąca do skalowania uczenia modelu w górę na wielu komputerach. Klaster uczący serwera parametrów składa się z procesów roboczych i serwerów parametrów. Zmienne są tworzone na serwerach parametrów i na każdym kroku są odczytywane i aktualizowane przez pracowników. Sprawdź szkoleniowy serwer Parametr poradnik dla szczegółów.

W TensorFlow 2, szkolenia serwer parametr wykorzystuje architekturę centralnego koordynatora oparte poprzez tf.distribute.experimental.coordinator.ClusterCoordinator klasie.

W tej realizacji, worker i parameter server zadania uruchamiane tf.distribute.Server jakoby słuchać zadań z koordynatorem. Koordynator tworzy zasoby, wysyła zadania szkoleniowe, zapisuje punkty kontrolne i zajmuje się niepowodzeniem zadań.

W programowaniu działa na koordynatora, można użyć ParameterServerStrategy obiekt zdefiniować etap treningowy i użyć ClusterCoordinator do etapów szkoleniowych wysyłki do zdalnych pracowników. Oto najprostszy sposób ich tworzenia:

strategy = tf.distribute.experimental.ParameterServerStrategy(
    tf.distribute.cluster_resolver.TFConfigClusterResolver(),
    variable_partitioner=variable_partitioner)
coordinator = tf.distribute.experimental.coordinator.ClusterCoordinator(
    strategy)

W TensorFlow 1 ParameterServerStrategy jest dostępny tylko z prognozy poprzez tf.compat.v1.distribute.experimental.ParameterServerStrategy symbolem.

Centralna strategia magazynowania

tf.distribute.experimental.CentralStorageStrategy robi synchronicznego szkolenia, jak również. Zmienne nie są dublowane, zamiast tego są umieszczane na procesorze, a operacje są replikowane na wszystkich lokalnych procesorach graficznych. Jeśli jest tylko jeden GPU, wszystkie zmienne i operacje zostaną umieszczone na tym GPU.

Utwórz instancję CentralStorageStrategy przez:

central_storage_strategy = tf.distribute.experimental.CentralStorageStrategy()
INFO:tensorflow:ParameterServerStrategy (CentralStorageStrategy if you are using a single machine) with compute_devices = ['/job:localhost/replica:0/task:0/device:GPU:0'], variable_device = '/job:localhost/replica:0/task:0/device:GPU:0'

Spowoduje to utworzenie CentralStorageStrategy instancji, który będzie korzystał ze wszystkich widocznych GPU i CPU. Aktualizacja zmiennych w replikach zostanie zagregowana przed zastosowaniem do zmiennych.

Inne strategie

Oprócz powyższych strategii, istnieją dwa inne strategie, które mogą być przydatne do tworzenia prototypów i debugowania przy użyciu tf.distribute API.

Domyślna strategia

Strategia Domyślna jest strategią dystrybucji, która jest obecna, gdy w zakresie nie ma wyraźnej strategii dystrybucji. Realizuje tf.distribute.Strategy interfejs ale to pass-through i nie daje rzeczywisty rozkład. Na przykład, strategy.run(fn) będzie po prostu zadzwonić fn . Kod napisany przy użyciu tej strategii powinien zachowywać się dokładnie tak, jak kod napisany bez żadnej strategii. Możesz myśleć o tym jako o strategii „bez operacji”.

Strategia domyślna jest singletonem – i nie można stworzyć jej więcej. Można go uzyskać za pomocą tf.distribute.get_strategy poza zakresem żadnej wyraźnej strategii, (ten sam API, które można wykorzystać, aby uzyskać aktualną strategię wewnątrz zakresu jasno sprecyzowanej strategii jest).

default_strategy = tf.distribute.get_strategy()

Ta strategia służy dwóm głównym celom:

  • Umożliwia bezwarunkowe pisanie kodu biblioteki uwzględniającego dystrybucję. Na przykład, w tf.optimizer s stosowanie może tf.distribute.get_strategy i stosowania, że strategia na rzecz zmniejszenia gradientów-to zawsze zwraca obiekt strategiczny, na której można wywołać zmniejszenie API.
# In optimizer or other library code
# Get currently active strategy
strategy = tf.distribute.get_strategy()
strategy.reduce("SUM", 1., axis=None)  # reduce some values
1.0
  • Podobnie jak kod biblioteczny, może być używany do pisania programów użytkowników końcowych do pracy ze strategią dystrybucji i bez niej, bez konieczności stosowania logiki warunkowej. Oto przykładowy fragment kodu ilustrujący to:
if tf.config.list_physical_devices('GPU'):
  strategy = tf.distribute.MirroredStrategy()
else:  # Use the Default Strategy
  strategy = tf.distribute.get_strategy() 

with strategy.scope():
  # Do something interesting
  print(tf.Variable(1.))
WARNING:tensorflow:Collective ops is not configured at program startup. Some performance features may not be enabled.
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)
MirroredVariable:{
  0: <tf.Variable 'Variable:0' shape=() dtype=float32, numpy=1.0>
}

Strategia OneDevice

tf.distribute.OneDeviceStrategy to strategia, aby umieścić wszystkie zmienne i obliczenia na jednym określonym urządzeniu.

strategy = tf.distribute.OneDeviceStrategy(device="/gpu:0")

Strategia ta różni się od Strategii Domyślnej na wiele sposobów. W strategii domyślnej logika rozmieszczania zmiennych pozostaje niezmieniona w porównaniu z uruchamianiem TensorFlow bez żadnej strategii dystrybucji. Jednak przy użyciu OneDeviceStrategy wszystkie zmienne utworzone w jego zakres są wyraźnie umieszczone na określonym urządzeniu. Ponadto wszelkie funkcje wywoływane poprzez OneDeviceStrategy.run zostanie również umieszczony na określonym urządzeniu.

Dane wejściowe dystrybuowane za pośrednictwem tej strategii będą wstępnie pobierane do określonego urządzenia. W Strategii Domyślnej nie ma dystrybucji danych wejściowych.

Podobnie jak w przypadku strategii domyślnej, strategia ta może być również używana do testowania kodu przed przejściem na inne strategie, które faktycznie dystrybuują do wielu urządzeń/maszyn. Pozwoli to wykonywać maszyny strategii dystrybucji nieco więcej niż domyślny strategii, ale nie w pełnym zakresie, jak przy użyciu MirroredStrategy lub TPUStrategy itd. Jeśli chcesz kod, który zachowuje się tak, jakby żadna strategia, a następnie użyć domyślnej strategii.

Do tej pory widziałeś różne dostępne strategie i sposoby ich tworzenia. W następnych kilku sekcjach przedstawiono różne sposoby wykorzystania ich do dystrybucji treningu.

Korzystanie tf.distribute.Strategy z tf.keras.Model.fit

tf.distribute.Strategy jest zintegrowany tf.keras , który jest realizacja TensorFlow jest w specyfikacji API Keras . tf.keras jest high-level API do budowania i pociągów modeli. Poprzez włączenie do tf.keras backend, to bez szwu, aby dystrybuować swój trening napisany w ramach szkolenia Keras wykorzystaniem Model.fit .

Oto, co musisz zmienić w swoim kodzie:

  1. Utwórz instancję odpowiedniej tf.distribute.Strategy .
  2. Przesuń tworzenie Keras modelu, optymalizacji i metryki wewnątrz strategy.scope .

Strategie dystrybucji TensorFlow obsługują wszystkie typy modeli Keras — sekwencyjne, funkcjonalne i podklasy.

Oto fragment kodu, który to zrobi dla bardzo prostego modelu Keras z jedną gęstą warstwą:

mirrored_strategy = tf.distribute.MirroredStrategy()

with mirrored_strategy.scope():
  model = tf.keras.Sequential([tf.keras.layers.Dense(1, input_shape=(1,))])

model.compile(loss='mse', optimizer='sgd')
WARNING:tensorflow:Collective ops is not configured at program startup. Some performance features may not be enabled.
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).

W tym przykładzie użyto MirroredStrategy , więc można uruchomić to na maszynie z wieloma procesorami graficznymi. strategy.scope() wskazuje Keras która strategia wykorzystania do dystrybucji szkolenia. Tworzenie modeli/optymalizatorów/metryk w tym zakresie umożliwia tworzenie zmiennych rozproszonych zamiast zwykłych zmiennych. Po skonfigurowaniu możesz dopasować swój model tak, jak zwykle. MirroredStrategy dba replikacji szkolenia modelki na dostępnych procesorów graficznych, sumując gradienty i inne.

dataset = tf.data.Dataset.from_tensors(([1.], [1.])).repeat(100).batch(10)
model.fit(dataset, epochs=2)
model.evaluate(dataset)
Epoch 1/2
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
2021-07-14 01:23:21.641884: W tensorflow/core/grappler/optimizers/data/auto_shard.cc:695] AUTO sharding policy will apply DATA sharding policy as it failed to apply FILE sharding policy because of the following reason: Found an unshardable source dataset: name: "TensorDataset/_2"
op: "TensorDataset"
input: "Placeholder/_0"
input: "Placeholder/_1"
attr {
  key: "Toutput_types"
  value {
    list {
      type: DT_FLOAT
      type: DT_FLOAT
    }
  }
}
attr {
  key: "output_shapes"
  value {
    list {
      shape {
        dim {
          size: 1
        }
      }
      shape {
        dim {
          size: 1
        }
      }
    }
  }
}

2021-07-14 01:23:21.671942: I tensorflow/compiler/mlir/mlir_graph_optimization_pass.cc:176] None of the MLIR Optimization Passes are enabled (registered 2)
2021-07-14 01:23:21.672424: I tensorflow/core/platform/profile_utils/cpu_utils.cc:114] CPU Frequency: 2000175000 Hz
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
2021-07-14 01:23:24.013985: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcublas.so.11
10/10 [==============================] - 3s 2ms/step - loss: 0.0014
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
Epoch 2/2
10/10 [==============================] - 0s 2ms/step - loss: 6.3687e-04
2021-07-14 01:23:24.434928: I tensorflow/stream_executor/platform/default/dso_loader.cc:53] Successfully opened dynamic library libcublasLt.so.11
2021-07-14 01:23:24.528027: W tensorflow/core/grappler/optimizers/data/auto_shard.cc:695] AUTO sharding policy will apply DATA sharding policy as it failed to apply FILE sharding policy because of the following reason: Found an unshardable source dataset: name: "TensorDataset/_2"
op: "TensorDataset"
input: "Placeholder/_0"
input: "Placeholder/_1"
attr {
  key: "Toutput_types"
  value {
    list {
      type: DT_FLOAT
      type: DT_FLOAT
    }
  }
}
attr {
  key: "output_shapes"
  value {
    list {
      shape {
        dim {
          size: 1
        }
      }
      shape {
        dim {
          size: 1
        }
      }
    }
  }
}
10/10 [==============================] - 1s 2ms/step - loss: 3.9551e-04
0.00039551049121655524

Tutaj tf.data.Dataset zapewnia szkolenia i wejście eval. Możesz także użyć tablic NumPy:

import numpy as np

inputs, targets = np.ones((100, 1)), np.ones((100, 1))
model.fit(inputs, targets, epochs=2, batch_size=10)
Epoch 1/2
2021-07-14 01:23:25.775102: W tensorflow/core/grappler/optimizers/data/auto_shard.cc:695] AUTO sharding policy will apply DATA sharding policy as it failed to apply FILE sharding policy because of the following reason: Did not find a shardable source, walked to a node which is not a dataset: name: "FlatMapDataset/_9"
op: "FlatMapDataset"
input: "PrefetchDataset/_8"
attr {
  key: "Targuments"
  value {
    list {
    }
  }
}
attr {
  key: "f"
  value {
    func {
      name: "__inference_Dataset_flat_map_slice_batch_indices_1006"
    }
  }
}
attr {
  key: "output_shapes"
  value {
    list {
      shape {
        dim {
          size: 10
        }
      }
    }
  }
}
attr {
  key: "output_types"
  value {
    list {
      type: DT_INT64
    }
  }
}
. Consider either turning off auto-sharding or switching the auto_shard_policy to DATA to shard this dataset. You can do this by creating a new `tf.data.Options()` object then setting `options.experimental_distribute.auto_shard_policy = AutoShardPolicy.DATA` before applying the options object to the dataset via `dataset.with_options(options)`.
10/10 [==============================] - 1s 2ms/step - loss: 2.8150e-04
Epoch 2/2
10/10 [==============================] - 0s 2ms/step - loss: 1.2442e-04
<tensorflow.python.keras.callbacks.History at 0x7fe091028f50>

W obu przypadkach-z Dataset lub NumPy-każda partia tego wejścia podzielono równo między wieloma replikach. Na przykład, jeśli używasz MirroredStrategy z 2 kart graficznych, każda partia rozmiaru 10 dostanie podzielone między 2 GPU, z których każdy odbiera 5 przykładów wejściowych w każdym kroku. Każda epoka będzie trenować szybciej, gdy dodasz więcej procesorów graficznych. Zazwyczaj chciałbyś zwiększyć rozmiar partii, dodając więcej akceleratorów, aby efektywnie wykorzystać dodatkową moc obliczeniową. W zależności od modelu konieczne będzie również ponowne dostrojenie szybkości uczenia się. Można użyć strategy.num_replicas_in_sync aby uzyskać liczbę replik.

# Compute global batch size using number of replicas.
BATCH_SIZE_PER_REPLICA = 5
global_batch_size = (BATCH_SIZE_PER_REPLICA *
                     mirrored_strategy.num_replicas_in_sync)
dataset = tf.data.Dataset.from_tensors(([1.], [1.])).repeat(100)
dataset = dataset.batch(global_batch_size)

LEARNING_RATES_BY_BATCH_SIZE = {5: 0.1, 10: 0.15}
learning_rate = LEARNING_RATES_BY_BATCH_SIZE[global_batch_size]

Co jest teraz obsługiwane?

Szkolenia API Strategia lustrzana Strategia TPU MultiWorkerMirroredStrategy ParametrServerStrategy CentralStorageStrategy
Keras API Utrzymany Utrzymany Wsparcie eksperymentalne Wsparcie eksperymentalne Wsparcie eksperymentalne

Przykłady i tutoriale

Oto lista samouczków i przykładów ilustrujących powyższą integrację end-to-end z Keras:

  1. Poradnik trenować MNIST z MirroredStrategy .
  2. Poradnik trenować MNIST korzystając MultiWorkerMirroredStrategy .
  3. Przewodnik na MNIST szkolenia z wykorzystaniem TPUStrategy .
  4. Poradnik dla szkolenia serwer parametrów w TensorFlow 2 z ParameterServerStrategy .
  5. TensorFlow model Garden repozytorium zawierającego zbiory state-of-the-art modeli realizowanych przy użyciu różnych strategii.

Korzystanie tf.distribute.Strategy z pętlami zwyczaj szkoleniowych

Jak widać, przy użyciu tf.distribute.Strategy z Keras model.fit wymaga zmiany tylko kilka linijek kodu. Z trochę więcej wysiłku, można również użyć tf.distribute.Strategy z pętlami zwyczaj szkoleniowych.

Jeśli potrzebujesz większej elastyczności i kontroli nad pętlami treningowymi niż jest to możliwe w Estimatorze lub Keras, możesz napisać własne pętle treningowe. Na przykład, używając GAN, możesz chcieć wykonać inną liczbę kroków generatora lub dyskryminatora w każdej rundzie. Podobnie, ramy wysokiego poziomu nie są zbyt odpowiednie do szkolenia w zakresie uczenia się przez wzmacnianie.

W tf.distribute.Strategy klasy zapewniają podstawowy zestaw metod do wsparcia przez pętle zwyczaj szkoleniowych. Korzystanie z nich może początkowo wymagać niewielkiej przebudowy kodu, ale gdy to zrobisz, powinieneś być w stanie przełączać się między procesorami graficznymi, jednostkami TPU i wieloma maszynami, po prostu zmieniając instancję strategii.

Tutaj zobaczysz krótki fragment ilustrujący ten przypadek użycia dla prostego przykładu szkoleniowego przy użyciu tego samego modelu Keras co poprzednio.

Najpierw utwórz model i optymalizator w zakresie strategii. Zapewnia to, że wszelkie zmienne utworzone za pomocą modelu i optymalizatora są zmiennymi lustrzanymi.

with mirrored_strategy.scope():
  model = tf.keras.Sequential([tf.keras.layers.Dense(1, input_shape=(1,))])
  optimizer = tf.keras.optimizers.SGD()

Następnie należy utworzyć zestaw danych wejściowych i nazywają tf.distribute.Strategy.experimental_distribute_dataset rozpowszechniać zestawu danych w oparciu o strategię.

dataset = tf.data.Dataset.from_tensors(([1.], [1.])).repeat(100).batch(
    global_batch_size)
dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)
2021-07-14 01:23:27.005233: W tensorflow/core/grappler/optimizers/data/auto_shard.cc:695] AUTO sharding policy will apply DATA sharding policy as it failed to apply FILE sharding policy because of the following reason: Found an unshardable source dataset: name: "TensorDataset/_2"
op: "TensorDataset"
input: "Placeholder/_0"
input: "Placeholder/_1"
attr {
  key: "Toutput_types"
  value {
    list {
      type: DT_FLOAT
      type: DT_FLOAT
    }
  }
}
attr {
  key: "output_shapes"
  value {
    list {
      shape {
        dim {
          size: 1
        }
      }
      shape {
        dim {
          size: 1
        }
      }
    }
  }
}

Następnie zdefiniuj jeden krok szkolenia. Zastosowanie tf.GradientTape do obliczenia gradientów i Optimizer, aby stosować te gradienty zaktualizować zmienne Twojego modelu. Dystrybuować ten etap szkolenia, umieścić go w funkcji train_step i przekazać go do tf.distrbute.Strategy.run wraz z wejściami zbioru danych, który dostał od dist_dataset utworzonego wcześniej:

loss_object = tf.keras.losses.BinaryCrossentropy(
  from_logits=True,
  reduction=tf.keras.losses.Reduction.NONE)

def compute_loss(labels, predictions):
  per_example_loss = loss_object(labels, predictions)
  return tf.nn.compute_average_loss(per_example_loss, global_batch_size=global_batch_size)

def train_step(inputs):
  features, labels = inputs

  with tf.GradientTape() as tape:
    predictions = model(features, training=True)
    loss = compute_loss(labels, predictions)

  gradients = tape.gradient(loss, model.trainable_variables)
  optimizer.apply_gradients(zip(gradients, model.trainable_variables))
  return loss

@tf.function
def distributed_train_step(dist_inputs):
  per_replica_losses = mirrored_strategy.run(train_step, args=(dist_inputs,))
  return mirrored_strategy.reduce(tf.distribute.ReduceOp.SUM, per_replica_losses,
                         axis=None)

Kilka innych rzeczy do zapamiętania w powyższym kodzie:

  1. Użyłeś tf.nn.compute_average_loss obliczyć straty. tf.nn.compute_average_loss sumuje za przykład straty i podzielenie sumy przez global_batch_size. Jest to ważne, ponieważ później po gradienty są obliczane na każdej repliki, są one łączone w poprzek replik przez ich zsumowanie.
  2. Użyłeś także tf.distribute.Strategy.reduce API do agregacji wyników zwracanych przez tf.distribute.Strategy.run . tf.distribute.Strategy.run zwraca wyniki z każdej lokalnej repliki w strategii, i istnieje wiele sposobów, aby spożywają ten wynik. Można reduce je, aby uzyskać zagregowaną wartość. Można również zrobić tf.distribute.Strategy.experimental_local_results aby uzyskać listę wartości zawarte w rezultacie, po jednej lokalnej repliki.
  3. Podczas rozmowy telefonicznej apply_gradients w zakresie strategii dystrybucji, jej zachowanie jest modyfikowany. W szczególności przed zastosowaniem gradientów w każdym wystąpieniu równoległym podczas treningu synchronicznego wykonuje sumę wszystkich replik gradientów.

Wreszcie, po zdefiniowaniu etap treningowy można iteracyjne nad dist_dataset i uruchomić szkolenia w pętli:

for dist_inputs in dist_dataset:
  print(distributed_train_step(dist_inputs))
tf.Tensor(0.36346385, shape=(), dtype=float32)
tf.Tensor(0.3616105, shape=(), dtype=float32)
tf.Tensor(0.3597728, shape=(), dtype=float32)
tf.Tensor(0.35795057, shape=(), dtype=float32)
tf.Tensor(0.35614368, shape=(), dtype=float32)
tf.Tensor(0.35435185, shape=(), dtype=float32)
tf.Tensor(0.3525751, shape=(), dtype=float32)
tf.Tensor(0.35081312, shape=(), dtype=float32)
tf.Tensor(0.3490658, shape=(), dtype=float32)
tf.Tensor(0.34733298, shape=(), dtype=float32)
tf.Tensor(0.34561458, shape=(), dtype=float32)
tf.Tensor(0.3439103, shape=(), dtype=float32)
tf.Tensor(0.3422201, shape=(), dtype=float32)
tf.Tensor(0.34054378, shape=(), dtype=float32)
tf.Tensor(0.33888122, shape=(), dtype=float32)
tf.Tensor(0.33723223, shape=(), dtype=float32)
tf.Tensor(0.3355967, shape=(), dtype=float32)
tf.Tensor(0.3339745, shape=(), dtype=float32)
tf.Tensor(0.33236548, shape=(), dtype=float32)
tf.Tensor(0.33076945, shape=(), dtype=float32)

W powyższym przykładzie, należy powtórzyć nad dist_dataset aby zapewnić wejście do treningu. Jesteś również wyposażony w tf.distribute.Strategy.make_experimental_numpy_dataset wspierania wejść NumPy. Można użyć tego interfejsu API do tworzenia zbioru danych przed wywołaniem tf.distribute.Strategy.experimental_distribute_dataset .

Innym sposobem iteracji po danych jest jawne użycie iteratorów. Możesz to zrobić, jeśli chcesz uruchomić określoną liczbę kroków, a nie iterować po całym zestawie danych. Powyższy iteracji będzie teraz być modyfikowana najpierw utworzyć iterator a następnie jawnie wywołać next na to, aby uzyskać dane wejściowe.

iterator = iter(dist_dataset)
for _ in range(10):
  print(distributed_train_step(next(iterator)))
tf.Tensor(0.3291864, shape=(), dtype=float32)
tf.Tensor(0.32761604, shape=(), dtype=float32)
tf.Tensor(0.32605836, shape=(), dtype=float32)
tf.Tensor(0.3245131, shape=(), dtype=float32)
tf.Tensor(0.32298028, shape=(), dtype=float32)
tf.Tensor(0.32145968, shape=(), dtype=float32)
tf.Tensor(0.3199512, shape=(), dtype=float32)
tf.Tensor(0.31845465, shape=(), dtype=float32)
tf.Tensor(0.31697, shape=(), dtype=float32)
tf.Tensor(0.31549713, shape=(), dtype=float32)

Obejmuje to najprostszym przypadku korzystania tf.distribute.Strategy API rozpowszechniać własne pętle szkoleniowych.

Co jest teraz obsługiwane?

Szkolenia API Strategia lustrzana Strategia TPU MultiWorkerMirroredStrategy ParametrServerStrategy CentralStorageStrategy
Niestandardowa pętla treningowa Utrzymany Utrzymany Wsparcie eksperymentalne Wsparcie eksperymentalne Wsparcie eksperymentalne

Przykłady i tutoriale

Oto kilka przykładów wykorzystania strategii dystrybucji z niestandardowymi pętlami treningowymi:

  1. Poradnik trenować MNIST korzystając MirroredStrategy .
  2. Przewodnik na MNIST szkolenia z wykorzystaniem TPUStrategy .
  3. TensorFlow model Garden repozytorium zawierającego zbiory state-of-the-art modeli realizowanych przy użyciu różnych strategii.

Inne tematy

W tej sekcji omówiono niektóre tematy dotyczące wielu przypadków użycia.

Konfigurowanie zmiennej środowiskowej TF_CONFIG

Do treningu wielu pracowników, jak już wcześniej wspomniano, trzeba skonfigurować TF_CONFIG zmienną środowiskową dla każdego pliku binarnego działa w klastrze. TF_CONFIG zmienna jest ciągiem JSON, który określa, jakie zadania stanowią klastrze, ich adresy i roli każdego zadania w klastrze. tensorflow/ecosystem repo zawiera szablon Kubernetes, który konfiguruje TF_CONFIG do zadań szkoleniowych.

Istnieją dwa składniki TF_CONFIG : klaster i zadanie.

  • Klaster dostarcza informacji o klastrze szkoleniowym, który jest dyktatem składającym się z różnych rodzajów zawodów, takich jak pracownik. W szkoleniu dla wielu pracowników zazwyczaj jeden pracownik bierze na siebie nieco większą odpowiedzialność, jak zapisywanie punktu kontrolnego i pisanie pliku podsumowującego dla TensorBoard, oprócz tego, co robi zwykły pracownik. Taki pracownik jest określany jako „główny” pracownika, i to jest w zwyczaju, że pracownik z indeksem 0 został wyznaczony jako główny pracownika (w istocie jest to jak tf.distribute.Strategy jest realizowany).
  • Zadanie natomiast dostarcza informacji o bieżącym zadaniu. Pierwszy klaster składników jest taki sam dla wszystkich procesów roboczych, a drugie zadanie składnika jest inne dla każdego pracownika i określa typ i indeks tego pracownika.

Jednym z przykładów TF_CONFIG jest:

os.environ["TF_CONFIG"] = json.dumps({
    "cluster": {
        "worker": ["host1:port", "host2:port", "host3:port"],
        "ps": ["host4:port", "host5:port"]
    },
   "task": {"type": "worker", "index": 1}
})

Ten TF_CONFIG określa, że istnieją trzy pracownicy i dwa "ps" zadania w "cluster" wraz z gospodarzami i portów. W "task" część nie określi, że rola bieżącego zadania w "cluster" -worker 1 (drugi pracownik). Ważne role w klastrze są "chief" , "worker" , "ps" i "evaluator" . Nie powinno być "ps" praca z wyjątkiem przypadku korzystania tf.distribute.experimental.ParameterServerStrategy .

Co dalej?

tf.distribute.Strategy aktywnie jest w fazie rozwoju. Wypróbuj go i dostarczyć i swoją opinię korzystając problemy GitHub .