tf.compat.v1.distribute.OneDeviceStrategy

Стратегия распределения для выполнения на одном устройстве.

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

tf.compat.v1.distribute.OneDeviceStrategy(
    device
)

Использование этой стратегии разместит все переменные, созданные в её области видимости, на указанном устройстве. Входные данные, распределённые с помощью этой стратегии, будут предварительно загружены на указанное устройство. Кроме того, все функции, вызываемые через strategy.run, также будут размещены на указанном устройстве.

Типичное использование этой стратегии — тестирование кода с помощью API tf.distribute.Strategy перед переключением на другие стратегии, которые фактически распределяют вычисления по нескольким устройствам/машинам.

Например:

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

with strategy.scope():
  v = tf.Variable(1.0)
  print(v.device)  # /job:localhost/replica:0/task:0/device:GPU:0

def step_fn(x):
  return x * 2

result = 0
for i in range(10):
  result += strategy.run(step_fn, args=(i,))
print(result)  # 90

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.

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