TensorFlow 官网API学习（Reading data）

小咪咪 2022-05-30 07:54 141阅读 0赞

# TensorFlow 官网API学习（Reading data） #

标签（空格分隔）： TensorFlow官网API

## [API地址][API] ##

## TensorFlow程序中四种获取数据途径 ##

1.`tf.data`API:很容易构建复杂的输入流水线（**推荐的方法！**）  
2.Feeding：在运行每一步时，利用Python代码提供数据。  
3.`QueueRunner`:在TensorFlow图最开始的时候，基于队列的输入队列来读取文件中的数据  
4.预先载入数据:在TensorFlow图中使用常数或者变量来存入数据（**仅适用于小数据集**）

## tf.data API ##

具体的细节在`tf.data.Dataset`中。`tf.data`API可以进行不同输入或文件格式的数据提取和预处理，同时可以对数据集进行批化、打乱和映射。它是旧的两种输入方式feeding和`QueueRunner`的改进版本。

## Feeding ##

**“Feeding”是效率最低的将数据加载进TensorFlow程序的方式，仅被用于小实验和debug**  
TensorFlow的feed方法让你在计算图中给任意Tensor注入数据。提供注入数据的方式是通过`feed_dict`的参数，然后使用`run()`和`eval()`来调用初始计算。

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

你可以使用输入数据来代替包括变量和常量在内的任意张量，但最好的选择是使用`tf.placeholder`节点。

## QueueRunner ##

**这个部分讨论的实现基于队列的输入流水线可以被`tf.data`API实现清晰地替代**  
一个典型的从文件读取记录基于队列的流水线有下列的阶段：

>  *  文件名列表
>  *  (可选)文件名是否打乱
>  *  (可选)训练轮数限制
>  *  文件名队列
>  *  针对文件格式的一个读取器(reader)
>  *  读取器的解码器
>  *  (可选)预处理
>  *  样本队列

### 文件名列表，打乱和训练轮数限制 ###

对前三个部分，可以将一个文件名列表传入`tf.train.string_input_producer`函数。`string_input_producer`创建一个FIFO队列来保存文件名直到reader需要它们。`string_input_producer`有是否打乱和设置训练轮数的限制。对于每一轮，队列runner将整个文件名列表添加进队列一次，如果`shuffle=True`那么打乱文件名，这个过程提供一个针对文件的均匀采样，因此样本彼此之间不会under-或者over-采样。**这个队列runner独立于reader将文件队列中的文件名提取，因此打乱和入队过程不会阻塞reader。**

### 文件格式 ###

选择一个合适的reader来匹配你的输入文件格式然后将文件名队列传入reader的read方法。read方法输出一个key来标识文件和记录(如果你有某些奇怪的记录对Debug很用)和一个标量字符串值，使用一个(或更多)decoder和转换器来解码这个字符串到Tensor。

### CSV格式 ###

读取CSV格式的文本文件，使用`tf.TextLineReader`和`tf.decode_csv`操作。  
例子如下：

import tensorflow as tf
    
    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:
            examples, label = sess.run([features, col5])
    
        coord.request_stop()
        coord.join(threads)

每一个read的执行读取会从文件中读取一行。`decode_csv`操作会解析结果到一个Tensor列表中，`record_defaults`参数在输入字符串值缺失时，会确定一个缺省值。  
**你必须在调用`run`或者`eval`来执行read方法前调用`tf.train.start_queue_runners`来开始队列**

### 定长记录 ###

对于每一个固定字节的记录形成的二进制文件，使用`tf.FixedLengthRecordReader`和`tf.decode_raw`操作。`tf.decode_raw`操作会将字符串转换为一个uint8的tensor。  
对于CIFAR10数据集，它的每一个记录都是定长的字节。1个字节的label然后紧跟着3072(32\*32\*3)字节的图像数据。

### 标准TensorFlow格式 ###

另一种将你的任意数据转换到一个支持的格式。这种方法使得混合和匹配数据集和网络结构变得简单。TensorFlow推荐的格式是`TFRecords file`，它包括`tf.train.Example protocol buffers`(包含`Features`作为域)。你需要写一点程序来得到你的数据，将它填充进一个`Example`协议缓存，序列化这个协议缓存到一个字符串，然后使用`tf.python_io.TFRecordWriter`来将字符串写入一个TFRecords文件。下面是官网给出的将MNIST数据转换到这种格式的样例[TFRecord write][]  
推荐的方式读取一个TFRecord文件是使用`tf.data.TFRecordDataset`，[TFRecord Read][]，为了使用基于输入流水线队列完成相同的工作，使用下面的代码:

filename_queue = tf.train.string_input_producer([filename], num_epochs=num_epochs)
    reader = tf.TFRecordReader()
    _, serialized_example = reader.read(filename_queue)
    image, label = decode(serialized_example)

### 预处理 ###

在解析出数据后，包括正则化数据、选取随机的切片以及添加噪声或者扭曲等预处理操作将进行。

### Batching ###

在流水线最后我们使用另一个队列来将样本变成批来进行后续的训练、求值和预测。我们使用`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 定义了随机采样的缓存大小,更大的意味着更好的随机化但是启动会比较慢同时会占用更多的内存。
        # capacity必须大于min_after_dequeue，而数量大则决定了我们预取的最大值。推荐: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_sizecapacity=capacity, min_after_dequeue=min_after_dequeue)
        return example_batch, label_batch

如果你需要在文件之间的样本进行更多并行的打乱，通过使用`tf.train.shuffle_batch_join`来使用多个reader实例。例子如下：

def read_my_file_format(filename_queue):
        # Same as above
        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, 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

我们依旧使用的是一个单独的文件名队列被所有的reader所共享。这样确保在相同的epoch使用不同的文件直到所有那个epoch的文件被启动。  
另外一个可以替代方式是，使用一个reader但是在`tf.train.shuffle_batch`的`num_threads`参数设置大于1。这样在同一时间只会读取一个文件(但是比只有一个线程要快)，而不是一次读取N个文件。这样做的重要性在于:

>  *  如果你有多于输入文件数量的读取线程数，避免有两个线程读取来自同一份文件同一个样本的风险。
>  *  并行读取N个文件可能会造成多次磁盘查找。  
>     那么我们应该使用什么数量的线程，`tf.train.shuffle_batch*`函数会将样本的队列的拥挤情况用一个summary添加到graph中。**如果拥有足够多的读取线程，summary应该会保持在0以上。**[使用示例][Link 1]

### 创建线程来预取正在使用的QueueRunner对象 ###

很多上面列举的`tf.train`函数会添加`tf.train.QueueRunner`对象到我们的graph中。这要求我们在运行任意训练或者预测步骤前必须调用`tf.train.start_queue_runners`，否则就会永远挂起。结合`tf.train.Coordinator`可以在出现错误时清晰地关闭这些线程。如果你设置了一个epoch的数量限制，那么将会需要一个epoch的计数器来进行初始化。推荐的示例代码如下：

import tensorflow as tf
    # 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()

## 内部工作原理（**很重要！**） ##

首先我们会先创建一个图，里面有一些队列连接起来的流水线阶段。第一个阶段会产生文件名来读取然后文件名会入队到文件名队列中(第一个队列)。第二个阶段第二个阶段使用文件名(使用Reader)来产生样本，然后把样本入队到样本队列中。根据你的设置方式，你可能实际上会在第二阶段有一些独立的样本使得你可以并行地读取。在这些阶段的最后是一个入队操作，入队进入到下一阶段的出队的队列。下面我们就要开启线程来执行这些入队操作， 这样我们就可以在训练循环里不断将样本队列里的样本取出来进行训练。  
![resnet\_result1.png-31.5kB][resnet_result1.png-31.5kB]  
为了创建这些队列和入队的操作我们需要使用`tf.train.add_queue_runner`函数将`tf.train.QueueRunner`添加到图中。每个`QueueRunner`负责一个阶段，拥有将被线程运行的入队操作的列表。一旦图被建立起来，`tf.train.start_queue_runners`函数会让每一个在图中的QueueRunner开启它们的线程来运行入队操作。  
如果一切正常，当运行训练过程时队列将会被背景的线程很快地填充。如果我们设置了训练轮数限制，在某个企图出队样本的节点，你会得到一个错误`tf.errors.OutOfRangeError`。这可以理解为TensorFlow中的**end of file(EOF)**  
最后一个组成部分就是`tf.train.Coordinator`。这个负责来让所有的线程知道是否有信号关闭。大多数情况下异常发生时，才会出现这种现象。

## 预先载入数据 ##

这种方法仅对于小的数据集比较合适，它们会被整  
个加载进内存。这里有两种方式：

>  *  将数据存入常量(constant)
>  *  将数据存入变量，然后你对其进行初始化（或者赋值）然后就不会改变

1.使用常量会比较简单，但是需要更多的内存(因为常量被内联地存储在图数据结构中，可能会被复制多次)  
2.使用变量则需要在图被建立后进行初始化。设置`trainable=False`保持变量在图中`GraphKeys.TRAINABLE_VARIABLES`集合之外，因此我们不会在训练过程中更新它。**设置`collections=[]`保持变量在`GraphKeys.GLOBAL_VARIABLES`集合外，用于存储和恢复检查点。**  
无论哪种方式，`tf.train.slice_input_producer`可以被用来产生一次切片。这个将会在一个完整的epoch内打乱样本，所以之后在batching时打乱是不必要的。因此我们使用`tf.train.batch`函数而不是shuffle\_batch函数。为了使用多个预处理线程，设置`num_threads`参数大于1。  
下面是两个MNIST预先加载数据的例子，第一个是使用常量第二个是使用变量。  
[MNIST预处理使用常量][MNIST]  
[MNIST预处理使用变量][MNIST 1]

## 多输入流水线 ##

通常来说你想要训练一个数据集然后对另一个评价，一种方式去实现是在可能独立的进程中有两个独立的图和会话：  
1.训练过程读取输入训练输入数据，定期将所有训练变量写入检查点文件。  
2.评价过程恢复检查点文件到预测模型，读取验证输入数据。  
在CIFAR-10示例模型中，这就是评价器和手动完成的工作，这有很多好处：  
1.eval是在已经训练好的变量单个快照上执行的。  
2.当训练已经完成并且退出后，你仍然可以执行eval操作。  
你可以在同一进程同一图中进行训练和评价，并且共享它们的变量或者网络层。详见共享变量的指南。[共享变量][Link 2]  
为了支持单独一个图的方法，`tf.data`也支持[advanced iterator types][]，它允许使用者不用在重建图或者会话的基础上，改变输入流水线。  
**Note：无所谓实现，很多操作(像`tf.layers.batch_normalization`和`tf.layers.dropout`)需要知道它们是处于训练模式还是评价模式，如果你要改变数据源你需要小心的设置**

[API]: https://www.tensorflow.org/api_guides/python/reading_data
[TFRecord write]: https://github.com/tensorflow/tensorflow/blob/r1.6/tensorflow/examples/how_tos/reading_data/convert_to_records.py
[TFRecord Read]: https://github.com/tensorflow/tensorflow/blob/r1.6/tensorflow/examples/how_tos/reading_data/fully_connected_reader.py
[Link 1]: https://www.tensorflow.org/programmers_guide/summaries_and_tensorboard
[resnet_result1.png-31.5kB]: /images/20220530/9f2661e95aa34fd9adb28b37e85bb25f.png
[MNIST]: https://github.com/tensorflow/tensorflow/blob/r1.6/tensorflow/examples/how_tos/reading_data/fully_connected_preloaded.py
[MNIST 1]: https://github.com/tensorflow/tensorflow/blob/r1.6/tensorflow/examples/how_tos/reading_data/fully_connected_preloaded_var.py
[Link 2]: https://www.tensorflow.org/programmers_guide/variables
[advanced iterator types]: https://www.tensorflow.org/programmers_guide/datasets#creating_an_iterator