tf.distribute.experimental.ParameterServerStrategy

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

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

tf.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,...)

<!-- Tabular view -->
 <table class="responsive fixed orange">
<colgroup><col width="214px"><col></colgroup>
<tr><th colspan="2"><h2 class="add-link">Args</h2></th></tr>

<tr>
<td>
`cluster_resolver`
</td>
<td>
Optional
<a href="../../../tf/distribute/cluster_resolver/ClusterResolver"><code>tf.distribute.cluster_resolver.ClusterResolver</code></a> object. Defaults to a
<a href="../../../tf/distribute/cluster_resolver/TFConfigClusterResolver"><code>tf.distribute.cluster_resolver.TFConfigClusterResolver</code></a>.
</td>
</tr>
</table>





<!-- Tabular view -->
 <table class="responsive fixed orange">
<colgroup><col width="214px"><col></colgroup>
<tr><th colspan="2"><h2 class="add-link">Attributes</h2></th></tr>

<tr>
<td>
`extended`
</td>
<td>
<a href="../../../tf/distribute/StrategyExtended"><code>tf.distribute.StrategyExtended</code></a> with additional methods.
</td>
</tr><tr>
<td>
`num_replicas_in_sync`
</td>
<td>
Returns number of replicas over which gradients are aggregated.
</td>
</tr>
</table>



## Methods

<h3 id="experimental_distribute_dataset"><code>experimental_distribute_dataset</code></h3>

<a target="_blank" href="https://github.com/tensorflow/tensorflow/blob/v1.15.0/tensorflow/python/distribute/distribute_lib.py#L614-L678">View source</a>

<pre class="devsite-click-to-copy prettyprint lang-py tfo-signature-link">
<code>experimental_distribute_dataset(
    dataset
)
</code></pre>

Distributes a tf.data.Dataset instance provided via `dataset`.

The returned distributed dataset can be iterated over similar to how
regular datasets can.
NOTE: Currently, the user cannot add any more transformations to a
distributed dataset.

The following is an example:

```python
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 distributed dataset
for x in dist_dataset:
  # process dataset elements
  strategy.experimental_run_v2(train_step, args=(x,))

Предположим, что набор входных данных сгруппирован по глобальному размеру пакета. С этим предположением мы будем делать все возможное, чтобы разделить каждый пакет между всеми репликами (один или более рабочих узлов).

В многоузловой настройке мы сначала попробуем распределить набор входных данных, попытавшись определить, создается ли набор входных данных из ReaderDatasets (например, TFRecordDataset, TextLineDataset и т. д.), и, если это так, попробуем разбить входные файлы. Обратите внимание, что должно быть по крайней мере один входной файл на рабочий узел. Если у вас меньше одного входного файла на рабочий узел, мы рекомендуем отключить распределение вашего набора входных данных с помощью метода ниже.

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

Вы можете отключить фрагментацию набора входных данных между рабочими узлами с помощью параметра auto_shard в tf.data.experimental.DistributeOptions.

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

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

experimental_distribute_datasets_from_function

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

experimental_distribute_datasets_from_function(
    dataset_fn
)

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

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

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

dataset_fn должен принимать экземпляр tf.distribute.InputContext, где можно получить доступ к информации о пакетировании и репликации ввода:

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)

inputs = strategy.experimental_distribute_datasets_from_function(dataset_fn)

for batch in inputs:
  replica_results = strategy.experimental_run_v2(replica_fn, args=(batch,))

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

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

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

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

experimental_run_v2

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

experimental_run_v2(
    fn, args=(), kwargs=None
)

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

Выполняет операции, указанные в fn, на каждой реплике. Если args или kwargs имеют значения "на реплику", такие как те, которые создаются "распределённым Dataset", при выполнении fn на определённой реплике, оно будет выполнено с компонентом этих значений "на реплику", соответствующим этой реплике.

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

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

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

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

scope

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

scope()

Возвращает менеджер контекста, выбирающий эту стратегию в качестве текущей.

Внутри блока кода with strategy.scope():, эта нить будет использовать создатель переменных, установленный strategy, и войдёт в свой "межрепликовый контекст".

update_config_proto

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

update_config_proto(
    config_proto
)

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

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

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