Spec-Zone.ru › TensorFlow

tf.compat.v1.distribute.experimental.MultiWorkerMirroredStrategy

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

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

tf.compat.v1.distribute.experimental.MultiWorkerMirroredStrategy(
    communication=tf.distribute.experimental.CollectiveCommunication.AUTO,
    cluster_resolver=None
)

Эта стратегия реализует синхронное распределённое обучение на нескольких рабочих узлах, каждый из которых может иметь несколько графических процессоров. Как и tf.distribute.MirroredStrategy, она реплицирует все переменные и вычисления на каждом локальном устройстве. Отличие в том, что она использует распределённую реализацию коллективных операций (например, all-reduce), позволяя нескольким рабочим узлам работать вместе.

Необходимо запустить программу на каждом рабочем узле и правильно настроить cluster_resolver. Например, если вы используете tf.distribute.cluster_resolver.TFConfigClusterResolver, на каждом рабочем узле должны быть установлены соответствующие task_type и task_id в переменной окружения TF_CONFIG. Пример TF_CONFIG для worker-0 в кластере из двух рабочих узлов:

TF_CONFIG = '{"cluster": {"worker": ["localhost:12345", "localhost:23456"]}, "task": {"type": "worker", "index": 0} }'

Программа запускается на каждом рабочем узле без изменений. Обратите внимание, что для коллективных операций необходимо участие каждого рабочего узла. Все API tf.distribute и не относящиеся к tf.distribute могут использовать коллективные операции внутри себя, например при создании контрольных точек и сохранении, поскольку чтение tf.Variable с tf.VariableSynchronization.ON_READ приводит к выполнению all-reduce для значения. Поэтому рекомендуется запускать на каждом рабочем узле одну и ту же программу. Выполнение действий в зависимости от task_type или task_id рабочего узла чревато ошибками.

cluster_resolver.num_accelerators() определяет количество графических процессоров, используемых стратегией. Если оно равно нулю, стратегия использует ЦП. На всех рабочих узлах должно использоваться одинаковое количество устройств, иначе поведение не определено.

Эта стратегия не предназначена для TPU. Вместо неё используйте tf.distribute.TPUStrategy.

После настройки TF_CONFIG использование этой стратегии аналогично использованию tf.distribute.MirroredStrategy и tf.distribute.TPUStrategy.

strategy = tf.distribute.MultiWorkerMirroredStrategy()

with strategy.scope():
  model = tf.keras.Sequential([
    tf.keras.layers.Dense(2, input_shape=(5,)),
  ])
  optimizer = tf.keras.optimizers.SGD(learning_rate=0.1)

def dataset_fn(ctx):
  x = np.random.random((2, 5)).astype(np.float32)
  y = np.random.randint(2, size=(2, 1))
  dataset = tf.data.Dataset.from_tensor_slices((x, y))
  return dataset.repeat().batch(1, drop_remainder=True)
dist_dataset = strategy.distribute_datasets_from_function(dataset_fn)

model.compile()
model.fit(dist_dataset)

Вы также можете написать собственный цикл обучения:

@tf.function
def train_step(iterator):

  def step_fn(inputs):
    features, labels = inputs
    with tf.GradientTape() as tape:
      logits = model(features, training=True)
      loss = tf.keras.losses.sparse_categorical_crossentropy(
          labels, logits)

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

  strategy.run(step_fn, args=(next(iterator),))

for _ in range(NUM_STEP):
  train_step(iterator)

Подробное руководство см. в разделе Обучение с несколькими рабочими узлами и Keras.

Сохранение

Сохранять данные и создавать контрольные точки необходимо на всех рабочих узлах, а не только на одном. Это связано с тем, что для переменных с synchronization=ON_READ при сохранении выполняется агрегирование. Во избежание состояний гонки рекомендуется сохранять данные по разным путям на каждом рабочем узле. Все рабочие узлы сохраняют одни и те же данные. Примеры см. в руководстве Обучение с несколькими рабочими узлами и Keras.

Известные проблемы

  • tf.distribute.cluster_resolver.TFConfigClusterResolver возвращает неверное количество ускорителей. Стратегия использует все доступные графические процессоры, если cluster_resolver — это tf.distribute.cluster_resolver.TFConfigClusterResolver или None.
  • В eager-режиме стратегию необходимо создать до вызова любого другого API TensorFlow.
Атрибуты
cluster_resolver Возвращает распознаватель кластера, связанный с этой стратегией.

В общем случае при использовании стратегии tf.distribute для нескольких рабочих узлов, например tf.distribute.experimental.MultiWorkerMirroredStrategy или tf.distribute.TPUStrategy(), со стратегией связан tf.distribute.cluster_resolver.ClusterResolver, экземпляр которого возвращается этим свойством.

Стратегии, которым необходим связанный tf.distribute.cluster_resolver.ClusterResolver, должны задать соответствующий атрибут или переопределить это свойство; в противном случае по умолчанию возвращается None. Такие стратегии также должны предоставлять информацию о том, что возвращает это свойство.

Стратегии для одного рабочего узла обычно не имеют tf.distribute.cluster_resolver.ClusterResolver; в таких случаях это свойство возвращает None.

tf.distribute.cluster_resolver.ClusterResolver может пригодиться, когда пользователю нужно получить доступ к таким сведениям, как спецификация кластера, тип задачи или идентификатор задачи. Например:

os.environ['TF_CONFIG'] = json.dumps({
  'cluster': {
      'worker': ["localhost:12345", "localhost:23456"],
      'ps': ["localhost:34567"]
  },
  'task': {'type': 'worker', 'index': 0}
})

# This implicitly uses TF_CONFIG for the cluster and current task info.
strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()

...

if strategy.cluster_resolver.task_type == 'worker':
  # Perform something that's only applicable on workers. Since we set this
  # as a worker above, this block will run on this particular instance.
elif strategy.cluster_resolver.task_type == 'ps':
  # Perform something that's only applicable on parameter servers. Since we
  # set this as a worker above, this block will not run on this particular
  # instance.

Дополнительную информацию см. в документации API для tf.distribute.cluster_resolver.ClusterResolver.

extended tf.distribute.StrategyExtended с дополнительными методами.
num_replicas_in_sync Возвращает количество реплик, по которым агрегируются градиенты.

Методы

distribute_datasets_from_function

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

distribute_datasets_from_function(
    dataset_fn, options=None
)

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

Аргумент dataset_fn, передаваемый пользователем, — это входная функция, которая принимает аргумент tf.distribute.InputContext и возвращает экземпляр tf.data.Dataset. Предполагается, что набор данных, возвращаемый dataset_fn, уже разбит на пакеты размером на одну реплику (то есть глобальный размер пакета, делённый на число синхронизированных реплик) и разделён на шарды. tf.distribute.Strategy.distribute_datasets_from_function не разбивает на пакеты и шарды экземпляр tf.data.Dataset, возвращённый входной функцией. dataset_fn будет вызвана на устройстве ЦП каждого рабочего узла и создаст набор данных, из которого каждая реплика на этом рабочем узле будет извлекать один пакет входных данных (то есть если у рабочего узла две реплики, на каждом шаге из Dataset будут извлечены два пакета).

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

Функция dataset_fn должна принимать экземпляр tf.distribute.InputContext, содержащий сведения о разбиении на пакеты и репликации входных данных.

С помощью свойства element_spec объекта tf.distribute.DistributedDataset, возвращаемого этим API, можно запросить tf.TypeSpec элементов, возвращаемых итератором. Это можно использовать, чтобы задать свойство input_signature объекта tf.function. Пример см. в разделе tf.distribute.DistributedDataset.element_spec.

Важно: Размер пакета на одну реплику у tf.data.Dataset, возвращаемого dataset_fn, должен отличаться от experimental_distribute_dataset, для которого используется глобальный размер пакета. Его можно вычислить с помощью input_context.get_per_replica_batch_size.
Примечание: При использовании TPUStrategy порядок обработки данных рабочими узлами при использовании tf.distribute.Strategy.experimental_distribute_dataset или tf.distribute.Strategy.distribute_datasets_from_function не гарантируется. Обычно это требуется, если вы используете tf.distribute для масштабирования прогнозирования. Однако можно добавить индекс каждому элементу пакета и соответствующим образом упорядочить результаты. Пример упорядочивания результатов см. в этом фрагменте кода.
Примечание: Преобразования набора данных с состоянием в настоящее время не поддерживаются для tf.distribute.experimental_distribute_dataset или tf.distribute.distribute_datasets_from_function. Все операции с состоянием, которые могут присутствовать в наборе данных, сейчас игнорируются. Например, если в наборе данных есть map_fn, использующий tf.random.uniform для поворота изображения, граф набора данных будет зависеть от состояния (то есть от случайного начального значения) на локальном компьютере, где выполняется процесс Python.

Руководство по использованию и свойствам этого метода см. в руководстве по распределённому вводу данных). Если вас интересует обработка последнего неполного пакета, прочитайте этот раздел.

Аргументы
dataset_fn Функция, принимающая экземпляр tf.distribute.InputContext и возвращающая tf.data.Dataset.
options tf.distribute.InputOptions, используемый для настройки параметров распределения этого набора данных.
Возвращает
Объект tf.distribute.DistributedDataset.

experimental_distribute_dataset

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

experimental_distribute_dataset(
    dataset, options=None
)

Создаёт tf.distribute.DistributedDataset из tf.data.Dataset.

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

Пример:

global_batch_size = 2
# Passing the devices is optional.
strategy = tf.distribute.MirroredStrategy(devices=["GPU:0", "GPU:1"])
# Create a dataset
dataset = tf.data.Dataset.range(4).batch(global_batch_size)
# Distribute that dataset
dist_dataset = strategy.experimental_distribute_dataset(dataset)
@tf.function
def replica_fn(input):
  return input*2
result = []
# Iterate over the `tf.distribute.DistributedDataset`
for x in dist_dataset:
  # process dataset elements
  result.append(strategy.run(replica_fn, args=(x,)))
print(result)
[PerReplica:{
  0: <tf.Tensor: shape=(1,), dtype=int64, numpy=array([0])>,
  1: <tf.Tensor: shape=(1,), dtype=int64, numpy=array([2])>
}, PerReplica:{
  0: <tf.Tensor: shape=(1,), dtype=int64, numpy=array([4])>,
  1: <tf.Tensor: shape=(1,), dtype=int64, numpy=array([6])>
}]

В этом методе автоматически выполняются три основные операции: разбиение на пакеты, распределение по шардам и предварительная выборка.

В приведённом выше фрагменте кода dataset разбивается на пакеты с помощью global_batch_size, а вызов experimental_distribute_dataset повторно разбивает dataset на пакеты нового размера, равного глобальному размеру пакета, делённому на число синхронизированных реплик. Мы перебираем набор данных с помощью цикла for в стиле Python. x — это tf.distribute.DistributedValues, содержащий данные для всех реплик; каждая реплика получает данные нового размера пакета. tf.distribute.Strategy.run позаботится о передаче правильных данных для каждой реплики из x нужному replica_fn, выполняемому на каждой реплике.

Распределение по шардам включает автоматическое распределение между несколькими рабочими узлами и внутри каждого узла. Во-первых, при распределённом обучении на нескольких рабочих узлах (то есть при использовании tf.distribute.experimental.MultiWorkerMirroredStrategy или tf.distribute.TPUStrategy) автоматическое распределение набора данных по рабочим узлам означает, что каждому узлу назначается подмножество всего набора данных (если задана соответствующая tf.data.experimental.AutoShardPolicy). Это позволяет на каждом шаге обрабатывать на каждом рабочем узле глобальный пакет элементов набора данных без пересечений. Для автоматического распределения по шардам предусмотрено несколько параметров, которые можно задать с помощью tf.data.experimental.DistributeOptions. Затем внутри каждого рабочего узла метод распределяет данные между всеми устройствами узла (если их больше одного). Это происходит независимо от автоматического распределения по шардам между несколькими рабочими узлами.

Примечание: для автоматического распределения по шардам между несколькими рабочими узлами по умолчанию используется режим tf.data.experimental.AutoShardPolicy.AUTO. Если набор данных создан на основе наборов данных для чтения (например, tf.data.TFRecordDataset, tf.data.TextLineDataset и т. д.), этот режим попытается распределить входной набор данных по файлам; в противном случае данные будут распределены по шардам так, что каждый рабочий узел прочитает весь набор данных, но обработает только назначенный ему шард. Если входных файлов меньше, чем рабочих узлов, рекомендуется отключить автоматическое распределение наборов данных по шардам между узлами, задав для tf.data.experimental.DistributeOptions.auto_shard_policy значение tf.data.experimental.AutoShardPolicy.OFF.

По умолчанию этот метод добавляет преобразование предварительной выборки в конец предоставленного пользователем экземпляра tf.data.Dataset. Аргументом преобразования предварительной выборки, то есть buffer_size, является число синхронизированных реплик.

Если описанная выше логика разделения пакетов и распределения набора данных по шардам нежелательна, используйте tf.distribute.Strategy.distribute_datasets_from_function, который не выполняет автоматическое разбиение на пакеты или шарды.

Примечание: При использовании TPUStrategy порядок обработки данных рабочими узлами при использовании tf.distribute.Strategy.experimental_distribute_dataset или tf.distribute.Strategy.distribute_datasets_from_function не гарантируется. Обычно это требуется, если вы используете tf.distribute для масштабирования прогнозирования. Однако можно добавить индекс каждому элементу пакета и соответствующим образом упорядочить результаты. Пример упорядочивания результатов см. в этом фрагменте кода.
Примечание: Преобразования набора данных с состоянием в настоящее время не поддерживаются для tf.distribute.experimental_distribute_dataset или tf.distribute.distribute_datasets_from_function. Все операции с состоянием, которые могут присутствовать в наборе данных, сейчас игнорируются. Например, если в наборе данных есть map_fn, использующий tf.random.uniform для поворота изображения, граф набора данных будет зависеть от состояния (то есть от случайного начального значения) на локальном компьютере, где выполняется процесс Python.

Руководство по использованию и свойствам этого метода см. в руководстве по распределённому вводу данных. Если вас интересует обработка последнего неполного пакета, прочитайте этот раздел.

Аргументы
dataset tf.data.Dataset, который будет распределён по шардам между всеми репликами в соответствии с приведёнными выше правилами.
options tf.distribute.InputOptions, используемый для настройки параметров распределения этого набора данных.
Возвращает
Объект tf.distribute.DistributedDataset.

experimental_local_results

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

experimental_local_results(
    value
)

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

Примечание: Возвращаются только значения на рабочем узле, запущенном этим клиентом. При использовании tf.distribute.Strategy, например tf.distribute.experimental.MultiWorkerMirroredStrategy, каждый рабочий узел является отдельным клиентом, и эта функция возвращает только значения, вычисленные на этом узле.
Аргументы
value Значение, возвращённое вызовом experimental_run() или run(), or a variable created inscope`.
Возвращает
Кортеж значений, содержащихся в value, где i-й элемент соответствует i-й реплике. Если value представляет одно значение, возвращается (value,).

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)
Аргументы
numpy_input Вложенная структура входных массивов NumPy, которые будут преобразованы в набор данных. Обратите внимание, что списки массивов NumPy объединяются в один массив, как обычно происходит в tf.data.Dataset.
session (Только для выполнения графов TensorFlow v1.x) Сеанс, используемый для инициализации.
Возвращает
Объект tf.data.Dataset, представляющий numpy_input.

experimental_run

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

experimental_run(
    fn, input_iterator=None
)

Выполняет операции из fn на каждой реплике, используя входные данные из input_iterator. (Устарело)

Устарело: ЭТА ФУНКЦИЯ УСТАРЕЛА. Она будет удалена в будущей версии. Инструкции по обновлению: этот метод недоступен в TF 2.x. Вместо него используйте run.
Устарело: Этот метод недоступен в TF 2.x. Вместо него используйте run.

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

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

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

Важно: В зависимости от реализации используемой tf.distribute.Strategy и от того, включён ли eager-режим, fn может быть вызвана один или несколько раз (по одному разу для каждой реплики).
Аргументы
fn Функция для выполнения. Входные данные функции должны соответствовать выходным данным input_iterator.get_next(). Выходные данные должны представлять собой tf.nest из Tensor.
input_iterator (Необязательно) входной итератор, из которого берутся входные данные.
Возвращает
Объединённое возвращаемое значение fn для всех реплик. Структура возвращаемого значения совпадает со структурой значения, возвращаемого fn. Каждый элемент структуры может быть PerReplica (если значения не синхронизированы), Mirrored (если значения синхронизированы) или Tensor (если используется одна реплика).

make_dataset_iterator

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

make_dataset_iterator(
    dataset
)

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

Устарело: Этот метод недоступен в TF 2.x.

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

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

Аргументы
dataset tf.data.Dataset, который будет равномерно распределён между всеми репликами.
Возвращает
Объект tf.distribute.InputIterator, возвращающий входные данные для каждого шага вычислений. Для получения входных данных пользователь должен вызвать initialize для возвращённого итератора.

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.

Аргументы
input_fn Функция, принимающая объект tf.distribute.InputContext и возвращающая tf.data.Dataset.
replication_mode Значение перечисления tf.distribute.InputReplicationMode. В настоящее время поддерживается только PER_WORKER, то есть input_fn вызывается один раз на рабочий узел. Реплики извлекают данные из локального tf.data.Dataset на своём рабочем узле.
Возвращает
Объект итератора, который сначала необходимо преобразовать с помощью .initialize(). Затем его можно передать в strategy.experimental_run() или вызвать iterator.get_next(), чтобы получить следующее значение для передачи в strategy.extended.call_for_each_replica().

reduce

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

reduce(
    reduce_op, value, axis=None
)

Выполняет редукцию value между репликами и возвращает результат на текущем устройстве.

strategy = tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])
def step_fn():
  i = tf.distribute.get_replica_context().replica_id_in_sync_group
  return tf.identity(i)

per_replica_result = strategy.run(step_fn)
total = strategy.reduce("SUM", per_replica_result, axis=None)
total
<tf.Tensor: shape=(), dtype=int32, numpy=1>

Чтобы увидеть, как это работает с несколькими репликами, рассмотрим тот же пример с MirroredStrategy и двумя графическими процессорами:

strategy = tf.distribute.MirroredStrategy(devices=["GPU:0", "GPU:1"])
def step_fn():
  i = tf.distribute.get_replica_context().replica_id_in_sync_group
  return tf.identity(i)

per_replica_result = strategy.run(step_fn)
# Check devices on which per replica result is:
strategy.experimental_local_results(per_replica_result)[0].device
# /job:localhost/replica:0/task:0/device:GPU:0
strategy.experimental_local_results(per_replica_result)[1].device
# /job:localhost/replica:0/task:0/device:GPU:1

total = strategy.reduce("SUM", per_replica_result, axis=None)
# Check device on which reduced result is:
total.device
# /job:localhost/replica:0/task:0/device:CPU:0

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

Примечание: Результат копируется на «текущее» устройство, которым обычно является ЦП рабочего узла, на котором выполняется программа. Для TPUStrategy это первый узел TPU. При использовании MultiWorkerMirroredStrategy с несколькими клиентами это ЦП каждого рабочего узла.

Для редукции значений между репликами предусмотрено несколько API tf.distribute:

  • tf.distribute.ReplicaContext.all_reduce: Отличается от Strategy.reduce тем, что предназначен для контекста реплики и не копирует результаты на хостовое устройство. Обычно all_reduce следует использовать для редукций внутри шага обучения, например для градиентов.
  • tf.distribute.StrategyExtended.reduce_to и tf.distribute.StrategyExtended.batch_reduce_to: эти API представляют собой более продвинутые версии Strategy.reduce, поскольку позволяют настраивать назначение результата. Они также вызываются в контексте между репликами.

Каким должно быть значение axis?

Значение, возвращаемое run для каждой реплики, например потеря для отдельного примера, распределяется по пакету между всеми репликами. Эта функция позволяет агрегировать значения по репликам, а также, при необходимости, по элементам пакета, задав соответствующий параметр axis.

Например, если глобальный размер пакета равен 8, а реплик 2, значения для примеров [0, 1, 2, 3] будут находиться на реплике 0, а [4, 5, 6, 7] — на реплике 1. С помощью axis=None reduce будет агрегировать значения только по репликам и возвращать [0+4, 1+5, 2+6, 3+7]. Это удобно, когда каждая реплика вычисляет скалярное значение или другое значение, не имеющее измерения «пакет» (например, градиент или потерю).

strategy.reduce("sum", per_replica_result, axis=None)

Иногда требуется агрегировать значения по всему глобальному пакету и по всем репликам. Для этого можно указать измерение пакета в качестве axis, обычно axis=0. В этом случае будет возвращён скаляр 0+1+2+3+4+5+6+7.

strategy.reduce("sum", per_replica_result, axis=0)

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

Аргументы
reduce_op значение tf.distribute.ReduceOp, указывающее, как следует объединять значения. Можно использовать строковое представление перечисления, например "SUM", "MEAN".
value экземпляр tf.distribute.DistributedValues, например возвращённый Strategy.run, который требуется объединить в один тензор. Также может быть обычным тензором при использовании с OneDeviceStrategy или стратегией по умолчанию.
axis задаёт измерение, по которому выполняется редукция внутри тензора каждой реплики. Обычно следует задавать измерение пакета или None, чтобы выполнять редукцию только по репликам (например, если у тензора нет измерения пакета).
Возвращает
Объект Tensor.

run

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

run(
    fn, args=(), kwargs=None, options=None
)

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

Этот метод — основной способ распределения вычислений с помощью объекта tf.distribute. Он вызывает fn для каждой реплики. Если args или kwargs содержат tf.distribute.DistributedValues, например созданные с помощью tf.distribute.DistributedDataset из tf.distribute.Strategy.experimental_distribute_dataset или tf.distribute.Strategy.distribute_datasets_from_function, при выполнении fn для конкретной реплики ей будет передан компонент tf.distribute.DistributedValues, соответствующий этой реплике.

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

Все аргументы в args или kwargs могут представлять собой вложенную структуру тензоров, например список тензоров. В этом случае args и kwargs будут переданы в fn, вызываемый для каждой реплики. Кроме того, args или kwargs могут быть объектами tf.distribute.DistributedValues, содержащими тензоры или составные тензоры, то есть tf.compat.v1.TensorInfo.CompositeTensor. В этом случае при каждом вызове fn будет получен компонент tf.distribute.DistributedValues, соответствующий данной реплике. Обратите внимание, что произвольные значения Python, не относящиеся к перечисленным выше типам, не поддерживаются.

Важно: В зависимости от реализации tf.distribute.Strategy и от того, включено ли нетерпеливое выполнение, fn может вызываться один или несколько раз. Если fn аннотирована с помощью tf.function или tf.distribute.Strategy.run вызывается внутри tf.function (нетерпеливое выполнение по умолчанию отключено внутри tf.function), fn вызывается один раз для каждой реплики, чтобы построить граф TensorFlow, который затем будет повторно использоваться при выполнении с новыми входными данными. В противном случае, если нетерпеливое выполнение включено, fn будет вызываться для каждой реплики на каждом шаге, как обычный код Python.

Примеры использования:

  1. Входной тензор-константа.

    strategy = tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])
    tensor_input = tf.constant(3.0)
    @tf.function
    def replica_fn(input):
      return input*2.0
    result = strategy.run(replica_fn, args=(tensor_input,))
    result
        PerReplica:{
          0: <tf.Tensor: shape=(), dtype=float32, numpy=6.0>,
          1: <tf.Tensor: shape=(), dtype=float32, numpy=6.0>
        }
        
  2. Входные данные DistributedValues.

    strategy = tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])
    @tf.function
    def run():
      def value_fn(value_context):
        return value_context.num_replicas_in_sync
      distributed_values = (
        strategy.experimental_distribute_values_from_function(
          value_fn))
      def replica_fn2(input):
        return input*2
      return strategy.run(replica_fn2, args=(distributed_values,))
    result = run()
    result
        <tf.Tensor: shape=(), dtype=int32, numpy=4>
        
  3. Используйте tf.distribute.ReplicaContext для выполнения allreduce значений.

    strategy = tf.distribute.MirroredStrategy(["gpu:0", "gpu:1"])
    @tf.function
    def run():
       def value_fn(value_context):
         return tf.constant(value_context.replica_id_in_sync_group)
       distributed_values = (
           strategy.experimental_distribute_values_from_function(
               value_fn))
       def replica_fn(input):
         return tf.distribute.get_replica_context().all_reduce(
             "sum", input)
       return strategy.run(replica_fn, args=(distributed_values,))
    result = run()
    result
        PerReplica:{
          0: <tf.Tensor: shape=(), dtype=int32, numpy=1>,
          1: <tf.Tensor: shape=(), dtype=int32, numpy=1>
        }
        
Аргументы
fn Функция, выполняемая для каждой реплики.
args Необязательные позиционные аргументы для fn. Элементом может быть тензор, вложенная структура тензоров или объект tf.distribute.DistributedValues.
kwargs Необязательные именованные аргументы для fn. Элементом может быть тензор, вложенная структура тензоров или объект tf.distribute.DistributedValues.
options Необязательный экземпляр tf.distribute.RunOptions, задающий параметры выполнения fn.
Возвращает
Объединённое возвращаемое значение fn по всем репликам. Структура возвращаемого значения совпадает со структурой значения, возвращаемого fn. Каждый элемент структуры может быть объектом tf.distribute.DistributedValues, объектом Tensor или объектом Tensor (например, при выполнении на одной реплике).

scope

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

scope()

Менеджер контекста, который делает стратегию текущей и распределяет переменные.

Этот метод возвращает менеджер контекста и используется следующим образом:

strategy = tf.distribute.MirroredStrategy(["GPU:0", "GPU:1"])
# Variable created inside scope:
with strategy.scope():
  mirrored_variable = tf.Variable(1.)
mirrored_variable
MirroredVariable:{
  0: <tf.Variable 'Variable:0' shape=() dtype=float32, numpy=1.0>,
  1: <tf.Variable 'Variable/replica_1:0' shape=() dtype=float32, numpy=1.0>
}
# Variable created outside scope:
regular_variable = tf.Variable(1.)
regular_variable
<tf.Variable 'Variable:0' shape=() dtype=float32, numpy=1.0>

Что происходит при входе в Strategy.scope?

  • strategy устанавливается в глобальном контексте как «текущая» стратегия. Внутри этой области tf.distribute.get_strategy() будет возвращать эту стратегию. Вне этой области возвращается стратегия по умолчанию, не выполняющая никаких действий.
  • При входе в область также выполняется вход в «контекст между репликами». Пояснение контекстов между репликами и реплик см. в разделе tf.distribute.StrategyExtended.
  • Создание переменных внутри scope перехватывается стратегией. Каждая стратегия определяет, как она влияет на создание переменных. Синхронные стратегии, такие как MirroredStrategy, TPUStrategy и MultiWorkerMiroredStrategy, создают переменные, реплицированные на каждой реплике, тогда как ParameterServerStrategy создаёт переменные на серверах параметров. Для этого используется пользовательский tf.variable_creator_scope.
  • В некоторых стратегиях также может выполняться вход в область устройства по умолчанию: в MultiWorkerMiroredStrategy на каждом рабочем узле устанавливается область устройства по умолчанию "/CPU:0".
Примечание: Вход в область не приводит к автоматическому распределению вычислений, за исключением случаев использования высокоуровневых платформ обучения, таких как keras model.fit. Если вы не используете model.fit, для явного распределения вычислений необходимо использовать API strategy.run. Пример приведён в руководстве по циклу пользовательского обучения.

Что должно находиться внутри области, а что — за её пределами?

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

  • Любые функции, создающие переменные, которые должны быть распределёнными, необходимо вызывать в strategy.scope. Это можно сделать, напрямую вызвав функцию создания переменной в контексте области, либо воспользовавшись другим API, например strategy.run или keras.Model.fit, который автоматически выполнит вход в область. Переменные, созданные вне области, не будут распределёнными, что может повлиять на производительность. В TF переменные создают, например, модели, оптимизаторы и метрики. Такие объекты всегда следует инициализировать в области, а любые функции, которые могут создавать переменные отложенно (например, Model.call(), трассировка tf.function и т. д.), также следует вызывать внутри области. Ещё одним источником создания переменных может быть восстановление из контрольной точки, если переменные создаются отложенно. Обратите внимание, что любая переменная, созданная внутри стратегии, сохраняет сведения о стратегии. Поэтому чтение и запись таких переменных за пределами strategy.scope также могут выполняться без проблем, и пользователю не нужно входить в область.
  • Некоторые API стратегии (например, strategy.run и strategy.reduce), для которых требуется находиться в области стратегии, входят в неё автоматически. Это означает, что при использовании таких API вам не нужно явно входить в область самостоятельно.
  • При создании tf.keras.Model внутри strategy.scope объект Model сохраняет сведения об области. При последующем вызове высокоуровневых методов платформы обучения, таких как model.compile, model.fit и т. д., сохранённая область будет открыта автоматически, а связанная с ней стратегия будет использоваться для распределения обучения и других операций. Подробный пример см. в руководстве по распределённому обучению с использованием Keras. ВНИМАНИЕ: простой вызов model(..) не открывает сохранённую область автоматически — такое поведение поддерживается только высокоуровневыми API платформы обучения: model.compile, model.fit, model.evaluate, model.predict и model.save можно вызывать как внутри, так и вне области.
  • Следующие действия можно выполнять как внутри области, так и вне её:
    • Создание входных наборов данных
    • Определение tf.function, представляющих шаг обучения
    • API сохранения, например tf.saved_model.save. При загрузке создаются переменные, поэтому, если вы хотите обучать модель распределённым способом, загрузку следует выполнять внутри области.
    • Сохранение контрольных точек. Как упоминалось выше, checkpoint.restore иногда необходимо выполнять внутри области, если при этом создаются переменные.
Возвращает
Менеджер контекста.

update_config_proto

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

update_config_proto(
    config_proto
)

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

Устарело: Этот метод недоступен в TF 2.x.

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

Аргументы
config_proto объект tf.ConfigProto.
Возвращает
Обновлённую копию config_proto.

© 2022 The TensorFlow Authors. All rights reserved.
Licensed under the Creative Commons Attribution License 4.0.
Code samples licensed under the Apache 2.0 License.
https://www.tensorflow.org/api_docs/python/tf/compat/v1/distribute/experimental/MultiWorkerMirroredStrategy

Spec-Zone.ru

Настройки Оффлайн Что нового Помощь О нас
Spec-Zone .ru
спецификации, руководства, описания, API