tf.compat.v1.distribute.Strategy

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

tf.compat.v1.distribute.Strategy(
    extended
)

См. руководство для обзора и примеров.

Примечание: Не все tf.distribute.Strategy реализации в настоящее время поддерживают разбиение переменных TensorFlow (где одна переменная разделена между несколькими устройствами).

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.functions, представляющих шаг обучения ** API сохранения, такие как tf.saved_model.save. Загрузка создает переменные, поэтому она должна находиться в области, если вы хотите обучить модель распределенным способом. ** Сохранение контрольных точек. Как упоминалось выше - checkpoint.restore иногда может потребоваться внутри области, если она создаёт переменные.

update_config_proto

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

update_config_proto(
    config_proto
)

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

УСТАРЕВШИЙ: Этот метод недоступен в TF 2.x.

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