tf.distribute.experimental.MultiWorkerMirroredStrategy

Стратегия распределения для синхронного обучения на нескольких узлах.

Наследуется от: Strategy

tf.distribute.experimental.MultiWorkerMirroredStrategy(
    communication=tf.distribute.experimental.CollectiveCommunication.AUTO
)

Эта стратегия реализует синхронное распределенное обучение на нескольких узлах, каждый из которых может иметь несколько графических процессоров. Подобно tf.distribute.MirroredStrategy, она создаёт копии всех переменных модели на каждом устройстве на всех узлах.

Она использует реализацию multi-worker all-reduce CollectiveOps для синхронизации переменных. Коллективная операция — это отдельная операция в графе TensorFlow, которая может автоматически выбирать алгоритм all-reduce в среде выполнения TensorFlow в соответствии с аппаратным обеспечением, топологией сети и размерами тензоров.

По умолчанию она использует все локальные графические процессоры или ЦП для обучения на одном узле.

Когда переменная окружения 'TF_CONFIG' установлена, она анализирует cluster_spec, task_type и task_id из 'TF_CONFIG' и преобразует их в стратегию для нескольких узлов, которая дублирует модели на графических процессорах всех машин в кластере. В текущей реализации она использует все графические процессоры в кластере и предполагает, что все узлы имеют одинаковое количество графических процессоров.

Она поддерживает как режим eager, так и режим графа. Однако для режима eager ей необходимо настроить контекст eager в конструкторе, поэтому все операции в режиме eager должны выполняться после создания объекта стратегии.

Методы

experimental_distribute_dataset

Просмотреть исходный код

experimental_distribute_dataset(
    dataset
)

Распределяет экземпляр tf.data.Dataset, предоставленный через dataset.

Возвращаемый распределённый набор данных можно перебирать аналогично тому, как можно перебирать обычные наборы данных. ПРИМЕЧАНИЕ: В настоящее время пользователь не может добавлять дополнительные преобразования к распределённому набору данных.

Вот пример:

strategy = tf.distribute.MirroredStrategy()

# Create a dataset
dataset = dataset_ops.Dataset.TFRecordDataset([
  "/a/1.tfr", "/a/2.tfr", "/a/3.tfr", "/a/4.tfr"])

# Distribute that dataset
dist_dataset = strategy.experimental_distribute_dataset(dataset)
# Iterate over the distributed dataset
for x in dist_dataset:
  # process dataset elements
  strategy.experimental_run_v2(train_step, args=(x,))

Мы будем предполагать, что входной набор данных сгруппирован по глобальному размеру пакета. При этом предположении мы будем стараться разделить каждый пакет на все реплики (один или несколько узлов).

В многоузловой среде мы сначала попытаемся распределить набор данных, пытаясь определить, создаётся ли набор данных из ReaderDatasets (например, TFRecordDataset, TextLineDataset и т. д.), и если да, то попытаться разделить входные файлы. Обратите внимание, что по крайней мере один входной файл должен приходиться на один узел. Если у вас меньше одного входного файла на узел, мы рекомендуем отключить распределение вашего набора данных с помощью метода ниже.

Если эта попытка окажется неудачной (например, набор данных создаётся из Dataset.range), мы разделим набор данных равномерно в конце, добавив операцию .shard в конец потока обработки. Это приведёт к тому, что весь поток предобработки всех данных будет выполняться на каждом узле, и каждый узел будет выполнять избыточную работу. Мы выведем предупреждение, если этот метод фрагментации выбран. В этом случае рассмотрите использование experimental_distribute_datasets_from_function вместо этого.

Вы можете отключить фрагментацию наборов данных между узлами, используя опцию auto_shard в tf.data.experimental.DistributeOptions.

Внутри каждого узла мы также разделим данные между всеми устройствами узла (если присутствует более одного), и это произойдёт даже если многоузловая фрагментация отключена с помощью метода выше.

Если вышеупомянутая логика разделения пакетов и фрагментации наборов данных нежелательна, используйте experimental_distribute_datasets_from_function вместо этого, который не выполняет автоматического разделения или фрагментации.

experimental_distribute_datasets_from_function

Просмотреть исходный код

experimental_distribute_datasets_from_function(
    dataset_fn
)

Распределяет экземпляры tf.data.Dataset, созданные вызовами dataset_fn.

dataset_fn будет вызван один раз для каждого узла в стратегии. Каждая реплика на этом узле будет извлекать одну партию входных данных из локального Dataset (то есть, если у узла две реплики, две партии будут извлекаться из Dataset на каждом шаге).

Этот метод может использоваться для нескольких целей. Например, там, где experimental_distribute_dataset не может разделить входные файлы, этот метод может использоваться для ручного разделения набора данных (избегая медленного поведения по умолчанию в experimental_distribute_dataset). В случаях, когда набор данных бесконечен, это разделение можно выполнить, создав реплики наборов данных, отличающиеся только случайным зерном. experimental_distribute_dataset иногда также может не преуспеть в разделении пакета между репликами на узле. В этом случае этот метод может быть использован там, где такого ограничения нет.

dataset_fn должен принимать экземпляр tf.distribute.InputContext, где можно получить информацию о пакетировании и репликации входных данных:

def dataset_fn(input_context):
  batch_size = input_context.get_per_replica_batch_size(global_batch_size)
  d = tf.data.Dataset.from_tensors([[1.]]).repeat().batch(batch_size)
  return d.shard(
      input_context.num_input_pipelines, input_context.input_pipeline_id)

inputs = strategy.experimental_distribute_datasets_from_function(dataset_fn)

for batch in inputs:
  replica_results = strategy.experimental_run_v2(replica_fn, args=(batch,))

experimental_local_results

Просмотреть исходный код

experimental_local_results(
    value
)

Возвращает список всех локальных значений по реплике, содержащихся в value.

Примечание: Это возвращает только значения на узле, инициированном этим клиентом. При использовании tf.distribute.Strategy, например, tf.distribute.experimental.MultiWorkerMirroredStrategy, каждый узел будет своим клиентом, и эта функция вернёт только значения, вычисленные на этом узле.

experimental_make_numpy_dataset

Просмотреть исходный код

experimental_make_numpy_dataset(
    numpy_input, session=None
)

Создаёт tf.data.Dataset для входных данных, предоставленных через массив NumPy.

Это позволяет избежать добавления numpy_input как большой константы в граф и копирует данные на машину или машины, которые будут обрабатывать ввод.

Обратите внимание, что вам, скорее всего, потребуется использовать tf.distribute.Strategy.experimental_distribute_dataset с возвращённым набором данных, чтобы дополнительно распределить его с помощью стратегии.

Пример:

numpy_input = np.ones([10], dtype=np.float32)
dataset = strategy.experimental_make_numpy_dataset(numpy_input)
dist_dataset = strategy.experimental_distribute_dataset(dataset)

experimental_run

Просмотреть исходный код

experimental_run(
    fn, input_iterator=None
)

Выполняет операции в fn на каждой реплике с входными данными из input_iterator.

УСТАРЕЛО: Этот метод недоступен в TF 2.x. Переключитесь на использование experimental_run_v2 вместо этого.

При включённом режиме eager выполнения, выполняет операции, указанные в fn на каждой реплике. В противном случае создаёт граф для выполнения операций на каждой реплике.

Каждая копия будет принимать один, отличающийся ввод из ввода, предоставленного одним get_next вызовом итератора ввода.

fn может вызвать tf.distribute.get_replica_context() для доступа к членам, таким как replica_id_in_sync_group.

experimental_run_v2

Посмотреть исходный код

experimental_run_v2(
    fn, args=(), kwargs=None
)

Выполнить fn на каждой копии с заданными аргументами.

Выполняет операции, заданные fn на каждой копии. Если args или kwargs имеют значения «на копию», например, те, что созданы распределённым Dataset, когда fn выполняется на конкретной копии, она будет выполнена с компонентом этих значений «на копию», который соответствует этой копии.

fn может вызвать tf.distribute.get_replica_context() для доступа к членам, таким как all_reduce.

Все аргументы в args или kwargs должны быть либо вложенными тензорами, либо объектами «на копию», содержащими тензоры или составные тензоры.

make_dataset_iterator

Посмотреть исходный код

make_dataset_iterator(
    dataset
)

Создаёт итератор для ввода, предоставленного через dataset.

УСТАРЕЛО: Этот метод недоступен в TF 2.x.

Данные из заданного набора данных будут распределены равномерно по всем копиям вычислительных узлов. Мы будем считать, что набор данных ввода сгруппирован по глобальному размеру пакета. С этим предположением мы будем прилагать все усилия для разделения каждого пакета по всем копиям (один или несколько рабочих узлов). Если эти усилия окажутся безуспешными, будет выброшено исключение, и пользователь должен вместо этого использовать make_input_fn_iterator, который предоставляет больше контроля пользователю и не пытается разделить пакет по копиям.

Пользователь также может использовать make_input_fn_iterator, если хочет настроить ввод, подаваемый для каждой копии/рабочего узла и т. д.

make_input_fn_iterator

Посмотреть исходный код

make_input_fn_iterator(
    input_fn, replication_mode=tf.distribute.InputReplicationMode.PER_WORKER
)

Возвращает итератор, разделённый по копиям, созданный из функции ввода.

УСТАРЕЛО: Этот метод недоступен в TF 2.x.

input_fn должна принять объект tf.distribute.InputContext, где можно получить информацию о пакетировании и фрагментации ввода:

def input_fn(input_context):
  batch_size = input_context.get_per_replica_batch_size(global_batch_size)
  d = tf.data.Dataset.from_tensors([[1.]]).repeat().batch(batch_size)
  return d.shard(input_context.num_input_pipelines,
                 input_context.input_pipeline_id)
with strategy.scope():
  iterator = strategy.make_input_fn_iterator(input_fn)
  replica_results = strategy.experimental_run(replica_fn, iterator)

Возвращаемый tf.data.Dataset из input_fn должен иметь размер пакета на копию, который можно вычислить с помощью input_context.get_per_replica_batch_size.

reduce

Посмотреть исходный код

reduce(
    reduce_op, value, axis=None
)

Сведение value по копиям.

Учитывая значение на копию, возвращённое experimental_run_v2, скажем, потерю на пример, пакет будет разделён между всеми копиями. Эта функция позволяет агрегировать по копиям и, необязательно, также по элементам пакета. Например, если у вас есть глобальный размер пакета 8 и 2 копии, значения для примеров [0, 1, 2, 3] будут на копии 0, а [4, 5, 6, 7] на копии 1. По умолчанию reduce просто агрегирует по копиям, возвращая [0+4, 1+5, 2+6, 3+7]. Это полезно, когда каждая копия вычисляет скаляр или какое-либо другое значение, у которого нет «размера пакета» (например, градиент). Чаще всего вы захотите агрегировать по глобальному пакету, что можно сделать, задав размер пакета как axis, обычно axis=0. В этом случае она вернёт скаляр 0+1+2+3+4+5+6+7.

Если есть последний частичный пакет, вам необходимо указать ось, чтобы размер получившейся формы был согласован между копиями. Так, если последний пакет имеет размер 6 и разделён на [0, 1, 2, 3] и [4, 5], у вас возникнет несоответствие размеров формы, если вы не укажете axis=0. Если вы укажете tf.distribute.ReduceOp.MEAN, используя axis=0 будет использоваться правильный знаменатель 6. В отличие от вычисления reduce_mean для получения скалярного значения на каждой копии и этой функции для усреднения этих средних, которые будут взвешивать некоторые значения 1/8, а другие 1/4.

scope

Посмотреть исходный код

scope()

Возвращает менеджер контекста, выбирающий эту стратегию в качестве текущей.

Внутри блока кода with strategy.scope(): этот поток будет использовать создатель переменной, установленный strategy, и войдёт в свой «межкопийный контекст».

update_config_proto

Посмотреть исходный код

update_config_proto(
    config_proto
)

Возвращает копию config_proto, изменённую для использования с этой стратегией.

УСТАРЕЛО: Этот метод недоступен в TF 2.x.

Обновлённая конфигурация содержит необходимые для запуска стратегии элементы, например, конфигурацию для запуска коллективных операций или фильтры устройств для повышения производительности распределённого обучения.