tf.data.experimental.service.DispatchServer

Сервер диспетчеризации службы tf.data в процессе.

tf.data.experimental.service.DispatchServer(
    config=None, start=True
)

Сервер tf.data.experimental.service.DispatchServer координирует кластер tf.data.experimental.service.WorkerServer серверов. При запуске рабочие серверы регистрируются у диспетчера.

dispatcher = tf.data.experimental.service.DispatchServer()
dispatcher_address = dispatcher.target.split("://")[1]
worker = tf.data.experimental.service.WorkerServer(
    tf.data.experimental.service.WorkerConfig(
    dispatcher_address=dispatcher_address))
dataset = tf.data.Dataset.range(10)
dataset = dataset.apply(tf.data.experimental.service.distribute(
    processing_mode="parallel_epochs", service=dispatcher.target))
print(list(dataset.as_numpy_iterator()))
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

При запуске отдельного процесса диспетчеризации tf.data используйте join(), чтобы заблокировать выполнение бесконечно после запуска сервера.

dispatcher = tf.data.experimental.service.DispatchServer(
    tf.data.experimental.service.DispatcherConfig(port=5050))
dispatcher.join()

Чтобы запустить DispatchServer в отказоустойчивом режиме, установите work_dir и fault_tolerant_mode следующим образом:

dispatcher = tf.data.experimental.service.DispatchServer(
    tf.data.experimental.service.DispatcherConfig(
        port=5050,
        work_dir="gs://my-bucket/dispatcher/work_dir",
        fault_tolerant_mode=True))

dispatcher = tf.data.experimental.service.DispatchServer()
dataset = tf.data.Dataset.range(10)
dataset = dataset.apply(tf.data.experimental.service.distribute(
    processing_mode="parallel_epochs", service=dispatcher.target))

Методы

join

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

join()

Ожидает завершения работы сервера.

Это полезно при запуске отдельного процесса диспетчеризации.

dispatcher = tf.data.experimental.service.DispatchServer(
    tf.data.experimental.service.DispatcherConfig(port=5050))
dispatcher.join()

start

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

start()

Запускает этот сервер.

dispatcher = tf.data.experimental.service.DispatchServer(start=False)
dispatcher.start()