tf.compat.v1.distribute.experimental.ParameterServerStrategy

Асинхронная стратегия tf.distribute для многоузлового сервера параметров.

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

tf.compat.v1.distribute.experimental.ParameterServerStrategy(
    cluster_resolver=None
)

Эта стратегия требует двух ролей: рабочих узлов и серверов параметров. Переменные и обновления этих переменных будут назначены серверам параметров, а другие операции — рабочим узлам.

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

По умолчанию она использует TFConfigClusterResolver для обнаружения конфигураций для многоузлового обучения. Для этого требуется переменная окружения 'TF_CONFIG', и 'TF_CONFIG' должен содержать спецификацию кластера.

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

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

1) В целом не рекомендуется открывать область устройства в области действия стратегии. Область устройства (т. е. вызов tf.device) будет объединена или перезапишет устройство для операций, но не изменит устройство для переменных.

2) Также не рекомендуется открывать область соположения (т. е. вызов tf.compat.v1.colocate_with) в области действия стратегии. Для соположения переменных используйте strategy.extended.colocate_vars_with вместо этого. Соположение операций может привести к конфликтам в назначении устройств.

Примечание: Эта стратегия работает только с API оценщика. Передайте экземпляр этой стратегии в аргумент experimental_distribute при создании RunConfig. Этот экземпляр RunConfig затем должен быть передан в экземпляр Estimator, на котором вызывается train_and_evaluate.

Пример:

strategy = tf.distribute.experimental.ParameterServerStrategy()
run_config = tf.estimator.RunConfig(
    experimental_distribute.train_distribute=strategy)
estimator = tf.estimator.Estimator(config=run_config)
tf.estimator.train_and_evaluate(estimator,...)

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. Обратите внимание, что вызов 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.

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