python - 将 numpy 数组传递给 tensorflow 队列

Question

我有一个 NumPy 数组，想使用 Queue 在 TensorFlow 的代码中读取它.我希望队列返回整个数据洗牌，一些指定数量的纪元，然后抛出错误。如果我不需要硬编码示例的大小或示例的数量，那将是最好的。我想shuffle batch就是为了达到这个目的。我试过如下使用它:

data = tf.constant(train_np) # train_np is my numpy array of shape (num_examples, example_size)
batch = tf.train.shuffle_batch([data], batch_size=5, capacity=52200, min_after_dequeue=10, num_threads=1, seed=None, enqueue_many=True)

sess.run(tf.initialize_all_variables())
tf.train.start_queue_runners(sess=sess)
batch.eval()

该方法的问题是它连续读取所有数据，我无法指定它在一定数量的纪元后完成。我知道我可以使用 RandomShuffleQueue并将数据插入其中几次，但是:a) 我不想浪费 epoch* 内存数据 b) 它将允许队列在 epoch 之间洗牌。

有没有一种不用编写自己的 Queue 就可以在 Tensorflow 中读取 epoch 中打乱的数据的好方法？

Answer 1

您可以创建另一个队列，将您的数据排入队列 num_epoch 次，关闭它，然后将其连接到您的 batch。为了节省内存，您可以使这个队列变小，并并行地将项目排入队列。时代之间会有一些混合。为了完全防止混合，您可以使用 num_epochs=1 并调用它 num_epochs 次。

tf.reset_default_graph()
data = np.array([1, 2, 3, 4])
num_epochs = 5
queue1_input = tf.placeholder(tf.int32)
queue1 = tf.FIFOQueue(capacity=10, dtypes=[tf.int32], shapes=[()])

def create_session():
    config = tf.ConfigProto()
    config.operation_timeout_in_ms=20000
    return tf.InteractiveSession(config=config)

enqueue_op = queue1.enqueue_many(queue1_input)
close_op = queue1.close()
dequeue_op = queue1.dequeue()
batch = tf.train.shuffle_batch([dequeue_op], batch_size=4, capacity=5, min_after_dequeue=4)

sess = create_session()

def fill_queue():
    for i in range(num_epochs):
        sess.run(enqueue_op, feed_dict={queue1_input: data})
    sess.run(close_op)

fill_thread = threading.Thread(target=fill_queue, args=())
fill_thread.start()

# read the data from queue shuffled
tf.train.start_queue_runners()
try:
    while True:
        print batch.eval()
except tf.errors.OutOfRangeError:
    print "Done"

顺便说一句，当队列不够大，无法将整个 numpy 数据集加载到其中时，上面的 enqueue_many 模式将挂起。您可以通过按如下方式分块加载数据来灵活地拥有更小的队列。

tf.reset_default_graph()
data = np.array([1, 2, 3, 4])
queue1_capacity = 2
num_epochs = 2
queue1_input = tf.placeholder(tf.int32)
queue1 = tf.FIFOQueue(capacity=queue1_capacity, dtypes=[tf.int32], shapes=[()])

enqueue_op = queue1.enqueue_many(queue1_input)
close_op = queue1.close()
dequeue_op = queue1.dequeue()

def dequeue():
    try:
        while True:
            print sess.run(dequeue_op)
    except:
        return 

def enqueue():
    for i in range(num_epochs):
        start_pos = 0
        while start_pos < len(data):
            end_pos = start_pos+queue1_capacity
            data_chunk = data[start_pos: end_pos]
            sess.run(enqueue_op, feed_dict={queue1_input: data_chunk})
            start_pos += queue1_capacity
    sess.run(close_op)

sess = create_session()

enqueue_thread = threading.Thread(target=enqueue, args=())
enqueue_thread.start()

dequeue_thread = threading.Thread(target=dequeue, args=())
dequeue_thread.start()