阅读(3053) 赞(10)

将数据导入TensorFlow

2017-09-25 15:14:49 更新

注意：将数据导入到 TensorFlow 程序的首选方法是使用数据集 API。

另外还有三种方法可以将数据导入到 TensorFlow 程序中：

Feeding：Python的代码在运行每个步骤时提供数据。
从文件读取：输入管道从 TensorFlow 图的开始处读取文件中的数据。
预加载数据：TensorFlow 图中的常量或变量保存所有数据(对于小型数据集)。

Feeding

TensorFlow 的 feed 机制允许您在计算图中向任何张量注入数据。因此，python 计算可以直接将数据导入到图中。

通过 feed_dict 参数向启动计算的 run() 或 eval () 调用提供 feed 数据。

注意：“Feeding” 是将数据传送到 TensorFlow 程序的最有效的方式，只能用于小型实验和调试。

with tf.Session():
  input = tf.placeholder(tf.float32)
  classifier = ...
  print(classifier.eval(feed_dict={input: my_python_preprocessing_fn()}))

虽然可以使用 Feed 数据(包括变量和常量)替换任何 Tensor，但最佳做法是使用 tf.placeholder 节点。placeholder(占位符)只是作为 feed 的目标存在。它未初始化，不包含任何数据如果占位符在没有 Feed 的情况下执行，则会产生错误,因此您不会忘记将其遗忘。

在 tensorflow/examples/tutorials/mnist/fully_connected_feed.py 中可以找到在 MNIST 数据上使用占位符和 Feeding 训练的示例，并在 MNIST 教程中进行了说明。

从文件导入

从文件导入记录的典型管道有以下几个阶段：

文件名列表
可选文件名洗牌
可选时期限制
文件名队列
用于文件格式的读取器
读者用于读取记录的解码器
可选预处理
示例队列

注意：本节讨论使用基于队列的API实现输入管道，该 API 可以被 ${$datasets$Dataset API} 完整地替换。

文件名,shuffling 和 epoch 限制

对于文件名列表，请使用常量字符串张量(如["file0", "file1"]或[("file%d" % i) for i in range(2)])或函数：tf.train.match_filenames_once。

将文件名列表传递给 tf.train.string_input_producer 函数。string_input_producer 创建一个 FIFO 队列，用于保存文件名，直到读取器需要它们为止。

string_input_producer 有选择的 shuffling 和设置一个最大的 epoch 数。队列运行程序为每个 epoch 将文件名的整个列表添加到队列中一次,如果洗牌 = True，则在一个 epoch 中重新排列文件名。此过程提供了一个统一的文件取样，以便相对于彼此不会对示例进行低估或过度采样。

队列运行程序在与从队列中抽取文件名的读取器分开的线程中工作，因此，shuffling 和 enqueuing 进程不会阻止读取器。

文件格式

选择与您的输入文件格式相匹配的读取器,并将文件名队列传递给读取器的读取方法。read 方法输出一个标识文件和记录的密钥 (如果有一些奇怪的记录，则对调试有用) 和一个标量字符串值，使用一个 (或多个) 解码器和转换 ops 将此字符串解码为构成示例的张量。

1、CSV 文件

若要以逗号分隔值 (CSV) 格式读取文本文件，请使用 tf.TextLineReader 与 tf.decode_csv 操作。例如：

filename_queue = tf.train.string_input_producer(["file0.csv", "file1.csv"])

reader = tf.TextLineReader()
key, value = reader.read(filename_queue)

# Default values, in case of empty columns. Also specifies the type of the
# decoded result.
record_defaults = [[1], [1], [1], [1], [1]]
col1, col2, col3, col4, col5 = tf.decode_csv(
    value, record_defaults=record_defaults)
features = tf.stack([col1, col2, col3, col4])

with tf.Session() as sess:
  # Start populating the filename queue.
  coord = tf.train.Coordinator()
  threads = tf.train.start_queue_runners(coord=coord)

  for i in range(1200):
    # Retrieve a single instance:
    example, label = sess.run([features, col5])

  coord.request_stop()
  coord.join(threads)

每次读取的执行都从文件中读取一行。然后，decode_csv 操作将结果解析为张量列表。该 record_defaults 参数确定生成的张量的类型，并设置在输入字符串中缺少值时要使用的默认值。

在调用 run 或 eval 执行读取之前，必须调用 tf.train.start_queue_runners 来填充队列。否则，读取将在等待队列中的文件名时阻止。

2、固定长度记录

要读取每条记录是固定字节数的二进制文件，请使用 tf.FixedLengthRecordReader 与 tf. decode_raw 操作.decode_raw 操作从一个字符串转换为 uint8 张量。

例如，CIFAR-10 数据集使用一种文件格式，其中每个记录使用固定的字节数表示：1 字节的标签,后跟3072字节的图像数据。一旦你有一个 uint8 张量，标准操作可以分割出每一块和并根据需要重新格式化。对于 CIFAR-10，您可以在 tensorflow_models/tutorials/image/cifar10/cifar10_input.py 中了解如何进行阅读和解码，并在本教程中介绍。

3、标准 TensorFlow 格式

另一种方法是将您拥有的任何数据转换为受支持的格式。这种方法使混合和匹配数据集和网络体系结构变得更加容易。TensorFlow 的推荐格式是包含 tf.train.Example 协议缓冲区 (包含作为字段的功能) 的 TFRecords 文件。您编写了一个小程序来获取您的数据，将它放在一个示例协议缓冲区中，将协议缓冲区序列化为一个字符串，然后使用 tf. python_io. TFRecordWriter 将该字符串写入 TFRecords 文件。例如，tensorflow/examples/how_tos/reading_data/convert_to_records.py 将 MNIST 数据转换为此格式。

要读取 TFRecords 的文件，请使用 tf.TFRecordReader 与 tf. parse_single_example 解码器.parse_single_example 操作将示例协议缓冲区解码为张量。使用 convert_to_records 生成的数据的 MNIST 示例可以在 tensorflow/examples/how_tos/reading_data/fully_connected_reader.py 中找到。您可以与 fully_connected_feed 版本进行比较。

预处理

然后，您可以对所需的这些示例进行任何预处理。这将是任何不依赖于训练参数的处理.示例包括数据正常化、选择随机切片、增加噪声或失真等。有关示例，请参见 tensorflow_models/tutorials/image/cifar10/cifar10_input.py。

批处理

在管道的最后,我们使用另一个队列来为训练，评估或推断一起批处理示例。为此，我们使用一个随机化的示例顺序的队列：tf.train.shuffle_batch。

例：

def read_my_file_format(filename_queue):
  reader = tf.SomeReader()
  key, record_string = reader.read(filename_queue)
  example, label = tf.some_decoder(record_string)
  processed_example = some_processing(example)
  return processed_example, label

def input_pipeline(filenames, batch_size, num_epochs=None):
  filename_queue = tf.train.string_input_producer(
      filenames, num_epochs=num_epochs, shuffle=True)
  example, label = read_my_file_format(filename_queue)
  # min_after_dequeue defines how big a buffer we will randomly sample
  #   from -- bigger means better shuffling but slower start up and more
  #   memory used.
  # capacity must be larger than min_after_dequeue and the amount larger
  #   determines the maximum we will prefetch.  Recommendation:
  #   min_after_dequeue + (num_threads + a small safety margin) * batch_size
  min_after_dequeue = 10000
  capacity = min_after_dequeue + 3 * batch_size
  example_batch, label_batch = tf.train.shuffle_batch(
      [example, label], batch_size=batch_size, capacity=capacity,
      min_after_dequeue=min_after_dequeue)
  return example_batch, label_batch

如果您需要在文件之间进行更多的并行性或示例的 shuffling，请使用多个读取器实例 tf.train.shuffle_batch_join。例如：

def read_my_file_format(filename_queue):
  # Same as above

def input_pipeline(filenames, batch_size, read_threads, num_epochs=None):
  filename_queue = tf.train.string_input_producer(
      filenames, num_epochs=num_epochs, shuffle=True)
  example_list = [read_my_file_format(filename_queue)
                  for _ in range(read_threads)]
  min_after_dequeue = 10000
  capacity = min_after_dequeue + 3 * batch_size
  example_batch, label_batch = tf.train.shuffle_batch_join(
      example_list, batch_size=batch_size, capacity=capacity,
      min_after_dequeue=min_after_dequeue)
  return example_batch, label_batch

您仍然只能使用由所有读取器共享的单个文件名队列.这样,我们确保不同的读取器使用不同的文件从同一个 epoch，直到所有的 epoch 文件已经开始。(通常只需一个线程填充文件名队列即可。)

另一种方法是使用 num_threads 大于1的 tf.train.shuffle_batch 的单一读取器。这将使它同时从单个文件中读取 (但速度比使用1线程快),而不是同时读 N 个文件.这可能很重要：

如果您有更多的读取线程而不是输入文件，则可以避免有两个线程在彼此附近的同一文件中读取同一示例的风险。
或者如果并行读取 N 个文件会导致过多的磁盘寻找。

您需要多少个线程？tf.train.shuffle_batch* 函数向关系图中添加一个摘要，以指示示例队列的完整程度。如果有足够的读取线程，该摘要将保持在零以上。您可以使用TensorBoard 查看您的摘要作为培训进度。

创建线程以使用 QueueRunner 对象进行预取

简短的版本：tf. train 上面列出的许多函数将 tf.train.QueueRunner 对象添加到图形中。这些要求您在运行任何培训或推理步骤之前调用 tf.train.start_queue_runners，否则它将永远挂起.这将启动运行输入管道的线程,填充示例队列，以使得出列获得示例的成功。这与 tf.train.Coordinator 很好的结合，可以在发生错误的时候，快捷地关闭这些线程时。如果您对 epoch 的数量设置了限制，那么将使用需要初始化的 epoch 计数器。建议的代码模式组合如下：

# Create the graph, etc.
init_op = tf.global_variables_initializer()

# Create a session for running operations in the Graph.
sess = tf.Session()

# Initialize the variables (like the epoch counter).
sess.run(init_op)

# Start input enqueue threads.
coord = tf.train.Coordinator()
threads = tf.train.start_queue_runners(sess=sess, coord=coord)

try:
    while not coord.should_stop():
        # Run training steps or whatever
        sess.run(train_op)

except tf.errors.OutOfRangeError:
    print('Done training -- epoch limit reached')
finally:
    # When done, ask the threads to stop.
    coord.request_stop()

# Wait for threads to finish.
coord.join(threads)
sess.close()

这里发生了什么？

首先，我们创建图形。它将有几个通过队列连接的管道阶段.第一阶段将生成文件名，以便在文件名队列中读取和排队它们。第二阶段使用文件名 (读取器)，生成示例，并将其排入示例队列。根据您设置内容的方式，您可能实际上拥有第二个阶段的一些独立副本，以便可以并行读取多个文件。在这些阶段的结束时，是一个排队操作，排入队列，下一个队列出队。我们要启动运行这些入队操作的线程，以便我们的训练循环可以从示例队列中出示示例。

在 tf. train 中创建这些队列和入队操作的帮助器使用 tf.train.add_queue_runner 函数向图形添加 tf.train.QueueRunner。每个 QueueRunner 负责一个阶段,并保存需要在线程中运行的入队操作的列表。一旦构造了图形，tf.train.start_queue_runners 函数就会要求图中的每个 QueueRunner 开始运行进行操作的线程。

如果一切顺利，您现在可以运行您的培训步骤，并且队列将由后台线程填充。如果你已经设置了一个 epoch 的限制，在某些时候试图出示的例子将得到一个 tf.errors.OutOfRangeError。这是 TensorFlow 相当于 "文件结束" (EOF)——这意味着已经达到了 epoch 限制,没有更多的示例可用。

最后是 tf.train.Coordinator。这是负责让所有线程知道有没有发出关闭信号。最常见的情况是，这是因为引发了一个异常，例如，其中一个线程在运行某项操作 (或一个普通的 Python 异常) 时发生了错误。

有关线程,队列，QueueRunners 和 Coordinators 的更多信息,请参见此处。

当限制 epochs 的工作时,如何清除关闭

想象一下，你有一个模型，它对训练的 epoch 数设定了限制。这意味着生成文件名的线程只会在生成 OutOfRange 错误之前运行多次。QueueRunner 将捕获该错误,关闭文件名队列，然后退出该线程.关闭队列有两点注意项：

任何将来在文件名队列中排队的尝试都将会产生错误。在这一点上,不应该有任何线程尝试这样做，但是当队列由于其他错误而关闭时,这是有用的。
任何当前或未来的出队都将立即成功(如果剩下足够的元素)或失败(OutOfRange 错误)。他们不会阻止等待更多的元素被排队，因为之前的一点不可能发生。

关键是,当文件名队列关闭时，该队列中可能仍有许多文件名，因此，管道的下一阶段 (读取器和其他预处理) 可能会继续运行一段时间。不过，一旦文件名队列被耗尽，下一次尝试将一个文件名排队 (例如,从已完成文件的读取器读取) 将触发 OutOfRange 错误。但是，在这种情况下，您可能有多个与单个 QueueRunner 关联的线程。如果这不是 QueueRunner 中的最后一个线程，则 OutOfRange 错误只会导致一个线程退出。这允许其他线程，它们仍然完成他们的最后一个文件，继续前进，直到他们也完成。(假设您正在使用的是 tf.train.Coordinator，其他类型的错误将导致所有线程停止。一旦所有读取器线程都命中了 OutOfRange 错误，则只有下一个队列 (即示例队列) 才会关闭。

同样，示例队列将有一些元素排队，因此训练将继续进行，直到耗尽为止。如果示例队列是 tf.RandomShuffleQueue，因为你使用 shuffle_batch 或 shuffle_batch_join，它通常会避免比其 min_after_dequeue 缓冲的 attr 元素少。但是，一旦队列关闭，将取消限制，并且队列最终将为空。在这一点上，实际的训练线程，当他们尝试从示例队列中出队时，将开始获得 OutOfRange 错误并且退出。一旦所有的训练线程完成，tf.train.Coordinator.join 将返回，你可以彻底退出。

筛选记录或每个记录生成多个示例

与形状 [x、y、z] 的示例相反，您将生成一批具有形状 [batch、x、y、z] 的示例。如果要将此记录过滤(也许它在保留集合中)，则批量大小可以为0；如果您每个记录生成多个示例，则大于1。然后在调用一个批处理函数 (如 shuffle_batch 或 shuffle_batch_join) 时，简单地设置 enqueue_many = True。

稀疏输入数据

SparseTensors(稀疏传感器)不能很好地排队。如果您使用 SparseTensors，则必须在批处理后使用 tf parse_example 对字符串记录进行解码 (而不是在批处理之前使用 tf. parse_single_example).

预加载数据

这仅用于可以完全加载到内存中的小型数据集.有两种方法：

将数据存储在常量中
将数据存储在变量中，并将其初始化 (或分配),然后再不更改

使用常数有点简单，但是会使用更多的内存(因为常量在图形数据结构中是内联存储的,这可能会重复几次)。

training_data = ...
training_labels = ...
with tf.Session():
  input_data = tf.constant(training_data)
  input_labels = tf.constant(training_labels)
  ...

要使用变量，您还需要在构建图形之后对其进行初始化。

training_data = ...
training_labels = ...
with tf.Session() as sess:
  data_initializer = tf.placeholder(dtype=training_data.dtype,
                                    shape=training_data.shape)
  label_initializer = tf.placeholder(dtype=training_labels.dtype,
                                     shape=training_labels.shape)
  input_data = tf.Variable(data_initializer, trainable=False, collections=[])
  input_labels = tf.Variable(label_initializer, trainable=False, collections=[])
  ...
  sess.run(input_data.initializer,
           feed_dict={data_initializer: training_data})
  sess.run(input_labels.initializer,
           feed_dict={label_initializer: training_labels})

设置 trainable = False 将变量从图中的 GraphKeys TRAINABLE_VARIABLES 集合中保留出来，这样我们就不会在训练时尝试更新它.设置 collections = [] 将变量保留在用于保存和还原检查点的 GraphKeys. GLOBAL_VARIABLES 集合中。

无论哪种方式，tf.train.slice_input_producer 都可以用于一次生成 slice。这个 shuffles 的例子可以在整个 epoch 中使用，所以进一步 shuffling 时，batching 是不可取的。因此，我们不使用 shuffle_batch 函数，而是使用普通的 tf.train.batch 函数.要使用多个预处理线程，请将 num_threads 参数设置为大于1的数字。

用于预先加载使用常量数据的 MNIST 例子可以在 tensorflow/examples/how_tos/reading_data/fully_connected_preloaded.py 找到，而一个用于预先加载使用变量的数据的 MNIST 例子可以在 tensorflow/examples/how_tos/reading_data/fully_connected_preloaded_var.py 找到，您可以将这些与上面的 fully_connected_feed 和 fully_connected_reader 版本进行比较。

多个输入管道

通常,您将希望在一个数据集上进行训练，并对另一个数据集进行评估 (或 "eval")。实现这一目标的一个方法是实际上有两个不同的图形和会话,可能在不同的进程中：

训练过程读取训练输入数据，并定期将检查点文件写入所有经过训练的变量。
评估过程将检查点文件还原为读取验证输入数据的推理模型。

这是在 CIFAR-10 示例中完成的估计和手动操作.这有几个好处：

eval 是在训练变量的单个快照上执行的
即使在训练完成并退出后，您也可以执行评估(eval)

您可以在同一进程中的同一图形中进行训练和评估(eval),并共享他们训练的变量或层.请参阅共享变量教程。
为了支持 single-graph(单图) 方法,数据集还提供了高级迭代器类型，允许用户在不重建图或会话的情况下更改输入管道。

您可以在同一个过程中的同一个图表中列出列车和eval，并分享训练有素的变量或层.请参阅共享变量教程。

为了支持单图方法，Datasets还提供了高级迭代器类型，允许用户在不重建图形或会话的情况下更改输入管道。

注意：无论执行情况如何，许多操作(如$ {tf.layers.batch_normalization}和 tf.layers.dropout)都需要知道他们是否处于训练或评估模式,如果更改数据源,则必须小心设置。

← TensorFlow 包装python函数

TensorFlow app模块 →