用于分布式训练的 TensorFlow 输入管道答案

【问题标题】：Tensorflow input pipeline for distributed training用于分布式训练的 TensorFlow 输入管道
【发布时间】：2017-08-07 19:56:06
【问题描述】：

我试图弄清楚如何在分布式训练中为 tensorflow 设置输入管道。目前尚不清楚读者是否会从单个进程中读取数据并将数据发送给所有工作人员，还是每个服务器将启动它自己的输入管道？我们如何确保每个工人都有不同的输入？

【问题讨论】：

如果您遵循 Google 的任何标准示例，每个工人都有自己的阅读器
请阅读您使用的标签的描述。 “ML”是指编程语言。

标签： python machine-learning tensorflow

【解决方案1】：

我将举例说明我是如何做到的：

import tensorflow as tf
batch_size = 50
task_index = 2
num_workers = 10
input_pattern = "gs://backet/dir/part-00*"

获取bucket中与input_pattern对应的所有文件名

files_names = tf.train.match_filenames_once(
                input_pattern, name = "myFiles")

为工作人员task_index 选择名称。 tf.strided_slice 就像列表的切片：a[::,task_index]（为工作人员task_index 选择每个task_indexth 文件）

to_process = tf.strided_slice(files_names, [task_index],
                 [999999999], strides=[num_workers])
filename_queue = tf.train.string_input_producer(to_process,
                     shuffle=True, #shufle files
                     num_epochs=num_epochs)

reader = tf.TextLineReader()
_ , value = reader.read(filename_queue)
col1,col2 = tf.decode_csv(value,
        record_defaults=[[1],[1]], field_delim="\t")

train_inputs, train_labels = tf.train.shuffle_batch([col1,[col2]],
        batch_size=batch_size,
        capacity=50*batch_size,
        num_threads=10,
        min_after_dequeue = 10*batch_size,
        allow_smaller_final_batch = True)

loss = f(...,train_inputs, train_labels)
optimizer = ...

with tf.train.MonitoredTrainingSession(...) as mon_sess:
    coord = tf.train.Coordinator()
    with coord.stop_on_exception():
        _ = tf.train.start_queue_runners(sess = mon_sess, coord=coord)
        while not coord.should_stop() and not mon_sess.should_stop():
            optimizer.run()

我不确定我的方法是在分布式 TensorFlow 实现的情况下实现输入管道的最佳方式，因为每个工作人员都会读取存储桶中所有文件的名称

关于 TensorFlow 输入管道的好讲座：http://web.stanford.edu/class/cs20si/lectures/notes_09.pdf

【讨论】：