tf.compat.v1.distribute.experimental.MultiWorkerMirroredStrategy

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

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

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

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

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

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

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

Вы также можете передать экземпляр distribute.cluster_resolver.ClusterResolver при создании стратегии. task_type, task_id и т. д. будут извлечены из экземпляра решателя, а не из переменной среды TF_CONFIG.

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

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.

Методы

experimental_distribute_dataset

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

experimental_distribute_dataset(
    dataset, options=None
)

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

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

Следующий пример:

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 `tf.distribute.DistributedDataset`
for x in dist_dataset:
  # process dataset elements
  strategy.run(replica_fn, args=(x,))

В приведённом фрагменте кода tf.distribute.DistributedDataset dist_dataset сгруппирован по GLOBAL_BATCH_SIZE, и мы перебираем его с помощью for x in dist_dataset. x tf.distribute.DistributedValues, содержащий данные для всех реплик, которые агрегируются в пакет GLOBAL_BATCH_SIZE. tf.distribute.Strategy.run позаботится о подаче правильных данных для каждой реплики в x в соответствующую replica_fn операцию, выполняемую на каждой реплике.

Что происходит «под капотом» в этом методе, когда мы говорим, что экземпляр tf.data.Dataset — dataset — распределяется? Это зависит от того, как вы устанавливаете tf.data.experimental.AutoShardPolicy через tf.data.experimental.DistributeOptions. По умолчанию он установлен на tf.data.experimental.AutoShardPolicy.AUTO. В многоузловой среде мы сначала попытаемся распределить dataset , обнаружив, является ли dataset созданным из наборов данных ридера (например, tf.data.TFRecordDataset, tf.data.TextLineDataset и т. д.), и если да, то попробуем разбить входные файлы. Обратите внимание, что должно быть как минимум один входной файл на каждый рабочий узел. Если у вас меньше одного входного файла на рабочий узел, мы рекомендуем отключить фрагментацию наборов данных между рабочими узлами, установив tf.data.experimental.DistributeOptions.auto_shard_policy в tf.data.experimental.AutoShardPolicy.OFF.

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

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

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

Вы также можете использовать свойство element_spec экземпляра tf.distribute.DistributedDataset, возвращённого этим API, для запроса tf.TypeSpec элементов, возвращаемых итератором. Это можно использовать для установки свойства input_signature функции tf.function.

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)

@tf.function(input_signature=[dist_dataset.element_spec])
def train_step(inputs):
  # train model with inputs
  return

# Iterate over the `tf.distribute.DistributedDataset`
for x in dist_dataset:
  # process dataset elements
  strategy.run(train_step, args=(x,))

Примечание: Порядок обработки данных рабочими узлами при использовании tf.distribute.Strategy.experimental_distribute_dataset или tf.distribute.Strategy.experimental_distribute_datasets_from_function не гарантируется. Это обычно требуется, если вы используете tf.distribute для масштабирования предсказания. Однако вы можете вставить индекс для каждого элемента в пакете и упорядочить результаты соответственно. Обратитесь к этому фрагменту фрагменту для примера того, как упорядочить результаты.

experimental_distribute_datasets_from_function

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

experimental_distribute_datasets_from_function(
    dataset_fn, options=None
)

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

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

Этот метод может быть использован для нескольких целей. Например, где 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.

global_batch_size = 8
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)

strategy = tf.distribute.MirroredStrategy()
ds = strategy.experimental_distribute_datasets_from_function(dataset_fn)

def train(ds):
  @tf.function(input_signature=[ds.element_spec])
  def step_fn(inputs):
    # train the model with inputs
    return inputs

... for batch in ds: ... replica_results = strategy.run(replica_fn, args=(batch,))

train(ds)

Примечание: Порядок обработки данных рабочими процессами при использовании tf.distribute.Strategy.experimental_distribute_dataset или tf.distribute.Strategy.experimental_distribute_datasets_from_function не гарантируется. Это обычно требуется, если вы используете tf.distribute для масштабирования предсказания. Тем не менее, вы можете вставить индекс для каждого элемента в пакете и упорядочить результаты соответственно. Обратитесь к этому фрагменту для примера упорядочивания результатов.

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. Пожалуйста, переключитесь на использование run вместо этого.

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

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

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

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 по репликам.

При заданном значении на копию, возвращаемом run, скажем, потере на пример, пакет будет разделен между всеми копиями. Эта функция позволяет агрегировать значения по копиям и, по желанию, также по элементам пакета. Например, если у вас есть глобальный размер пакета 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.

run

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

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

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

Выполняет операции, указанные в fn на каждой копии. Если args или kwargs имеют tf.distribute.DistributedValues, такие как те, что созданы tf.distribute.DistributedDataset из tf.distribute.Strategy.experimental_distribute_dataset или tf.distribute.Strategy.experimental_distribute_datasets_from_function, когда fn выполняется на конкретной копии, она будет выполнена с компонентом tf.distribute.DistributedValues, соответствующим этой копии.

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

Все аргументы в args или kwargs должны быть либо вложенными тензорами, либо tf.distribute.DistributedValues, содержащими тензоры или составные тензоры.

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

1. Ввод тензора константы.

strategy = tf.distribute.MirroredStrategy()
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
<tf.Tensor: shape=(), dtype=float32, numpy=6.0>

1. Ввод DistributedValues.

strategy = tf.distribute.MirroredStrategy()
@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=2>

scope

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

scope()

Менеджер контекста для установки стратегии по умолчанию и распределения переменных.

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

strategy = tf.distribute.MirroredStrategy()
# 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>
}
# 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 или model.fit, чтобы сделать это за вас. Любая переменная, созданная вне области, не будет распределена и может повлиять на производительность. Общие вещи, создающие переменные в TF: модели, оптимизаторы, метрики. Они всегда должны создаваться внутри области. Другим источником создания переменных может быть восстановление контрольной точки — когда переменные создаются лениво. Обратите внимание, что любая переменная, созданная внутри стратегии, захватывает информацию о стратегии. Поэтому чтение и запись в эти переменные вне strategy.scope также могут работать без проблем, без необходимости для пользователя входить в область действия.
• Некоторые API стратегии (такие как strategy.run и strategy.reduce) которые требуют находиться в области действия стратегии, автоматически входят в область, что означает, что при использовании этих API вам не нужно входить в область действия самостоятельно.
• Когда tf.keras.Model создаётся внутри strategy.scope, мы захватываем эту информацию. Когда высокоуровневые методы обучения, такие как model.compile, model.fit и т. д., вызываются на этой модели, мы автоматически входим в область действия, а также используем эту стратегию для распределения обучения и т. д. Подробнее см. в руководстве по распределённому Keras . Обратите внимание, что простой вызов model(..) не затрагивается — только API высокоуровневых фреймворков обучения. model.compile, model.fit, model.evaluate, model.predict и model.save могут быть вызваны внутри или вне области действия.
• Следующее может быть как внутри, так и вне области действия: ** Создание входных наборов данных ** Определение tf.functions, которые представляют ваш шаг обучения ** Сохраняющие API, такие как tf.saved_model.save. Загрузка создаёт переменные, поэтому это должно происходить внутри области действия, если вы хотите обучать модель в распределённом виде. ** Сохранение контрольных точек. Как упоминалось выше — checkpoint.restore иногда может потребоваться находиться в области действия, если это создаёт переменные.

update_config_proto

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

update_config_proto(
    config_proto
)

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

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

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