Как распараллелить входной конвейер Python в распределенном тензорном потоке

Question

У меня нетривиальный входной конвейер, который состоит из чтения правды и необработанных данных и выполнения предварительной обработки на них, написанных на Python.Требуется много времени, чтобы запустить конвейер ввода для одного образца, поэтому у меня есть несколько процессов (из пакета многопроцессорной обработки Python), выполняющихся параллельно, и очереди для быстрого выполнения операции и предварительной выборки данных.Затем выходные данные передаются в мою сеть с помощью feed_dict.Затраты на этот процесс в моем цикле обучения на 2 порядка меньше, чем фактическое время tf.Session.run ().Я пытаюсь перейти к API tf.data, добавив tf.py_func мои функции препроцессора read +, но он работает медленно, возможно, из-за GIL, даже при увеличении количества множественных вызовов.Я хочу расширить свое обучение на несколько машин и не уверен, как происходит выборка данных в таком случае, также есть проблема с производительностью и для одной машины:)

Итак, в основном мой вопрос: какзапустить функции python во входном конвейере API tf.data параллельно на нескольких ядрах процессора?

Answer 1

Пара пояснений: tf.py_func может работать параллельно с вашим sess.run() (потому что sess.run() освобождает GIL), но вы не можете запустить несколько tf.py_func в одном и том же процессе Python.

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

Если вы можете выразить свою предварительную обработку с помощью операций tf, вы можете запустить ее параллельно, используя Dataset.map, но в tf.data нет встроенной поддержки многопроцессорной обработки Python.Если вышеперечисленное не работает по какой-либо причине, вам, вероятно, придется подключить многопроцессорную систему самостоятельно.

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