tensorflow - 像队列一样将数据输入 tf.contrib.data.Dataset

Question

关于 tf.contrib.data.Dataset (来自 TensorFlow 1.2，参见 here 和 here )用法:
如何获取数据的方式并不适合我通常获取数据的方式。就我而言，我有一个线程，我在那里接收数据，我事先不知道它什么时候结束，但我知道它什么时候结束。然后我等到处理完所有缓冲区，然后我完成了一个 epoch。如何使用 Dataset 获得此逻辑?

请注意，我更喜欢 Dataset QueueBase 上的接口(interface)接口(interface)，因为它为我提供了迭代器接口(interface)，我可以重新初始化甚至重置为不同的 Dataset .与当前关闭后无法重新打开的队列相比，这更强大(参见 here 和 here )。

也许是一个类似的问题，或者同样的问题:我怎样才能绕过 Dataset排队？我有一些线程可以从某个地方读取一些数据，并且可以以某种方式对其进行馈送和排队。如何将数据输入 Dataset ?我可以无限次重复一些虚拟张量，然后使用 map只需返回我的queue.dequeue()但这真的只能让我回到队列的所有原始问题，即如何重新打开队列。

Answer 1

新 Dataset.from_generator() 方法允许您定义 Dataset由 Python 生成器提供。 (目前要使用此功能，您必须下载 TensorFlow 的夜间版本或自己从源代码构建。它将成为 TensorFlow 1.4 的一部分。)

实现您的示例的最简单方法是用生成器替换您的接收线程，伪代码如下:

def receiver():
  while True:
    next_element = ...  # Receive next element from external source.
                        # Note that this method may block.

    end_of_epoch = ...  # Decide whether or not to stop based on next_element.

    if not end_of_epoch:
      yield next_element  # Note: you may need to convert this to an array.
    else:
      return  # Returning will signal OutOfRangeError on downstream iterators.

dataset = tf.contrib.data.Dataset.from_generator(receiver, output_types=...)

# You can chain other `Dataset` methods after the generator. For example:
dataset = dataset.prefetch(...)  # This will start a background thread
                                 # to prefetch elements from `receiver()`.

dataset = dataset.repeat(...)  # Note that each repetition will call
                               # `receiver()` again, and start from
                               # a fresh state.

dataset = dataset.batch(...)

更复杂的拓扑是可能的。例如，您可以使用 Dataset.interleave() 并行创建许多接收器。