tf.compat.v1.distribute.MirroredStrategy

Синхронное обучение на нескольких репликах на одном компьютере.

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

tf.compat.v1.distribute.MirroredStrategy(
    devices=None, cross_device_ops=None
)

Эта стратегия обычно используется для обучения на одном компьютере с несколькими графическими процессорами. Для TPUs используйте tf.distribute.TPUStrategy. Чтобы использовать MirroredStrategy с несколькими рабочими узлами, обратитесь к tf.distribute.experimental.MultiWorkerMirroredStrategy.

Например, переменная, созданная в рамках MirroredStrategy, является MirroredVariable. Если в аргументе конструктора стратегии не указаны устройства, она будет использовать все доступные графические процессоры. Если графические процессоры не найдены, она будет использовать доступные центральные процессоры. Обратите внимание, что TensorFlow обрабатывает все центральные процессоры на компьютере как одно устройство и использует потоки для параллелизма внутри.

strategy = tf.distribute.MirroredStrategy()
with strategy.scope():
  x = tf.Variable(1.)
x
MirroredVariable:{
    0: <tf.Variable 'Variable:0' shape=() dtype=float32, numpy=1.0>
  }

При использовании стратегий распределения все создание переменных должно выполняться в области действия стратегии. Это позволит дублировать переменные на всех репликах и поддерживать их синхронизацию с помощью алгоритма all-reduce.

Переменные, созданные внутри MirroredStrategy, который обернут в tf.function, всё ещё MirroredVariables.

x = []
@tf.function  # Wrap the function with tf.function.
def create_variable():
  if not x:
    x.append(tf.Variable(1.))
strategy = tf.distribute.MirroredStrategy()
with strategy.scope():
  create_variable()
  print (x[0])
MirroredVariable:{
    0: <tf.Variable 'Variable:0' shape=() dtype=float32, numpy=1.0>
  }

experimental_distribute_dataset можно использовать для распределения набора данных по репликам при написании собственного цикла обучения. Если вы используете .fit и .compile методы, доступные в tf.keras, то tf.keras будет обрабатывать распределение за вас.

Например:

my_strategy = tf.distribute.MirroredStrategy()
with my_strategy.scope():
  @tf.function
  def distribute_train_epoch(dataset):
    def replica_fn(input):
      # process input and return result
      return result

    total_result = 0
    for x in dataset:
      per_replica_result = my_strategy.run(replica_fn, args=(x,))
      total_result += my_strategy.reduce(tf.distribute.ReduceOp.SUM,
                                         per_replica_result, axis=None)
    return total_result

  dist_dataset = my_strategy.experimental_distribute_dataset(dataset)
  for _ in range(EPOCHS):
    train_result = distribute_train_epoch(dist_dataset)

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 для явного распределения вычислений. См. пример в учебнике по пользовательскому циклу обучения custom training loop tutorial.

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

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

• Любая операция, создающая переменные, которые должны быть распределёнными переменными, должна быть в 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 distributed keras tutorial. Обратите внимание, что простые вызовы 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.

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