图像数据处理

图像的亮度、对比度等属性对图像的影响是非常大的，然而在很多图像识别问题中，这些因素都不应该影响最后的识别结果，所以在训练模型之前，需要对图像数据进行预处理，使训练得到的模型尽可能小地被无关因素影响。

7.1 TFRecord输入数据格式

7.1.1. TFRecord 格式介绍

7.1.2 TFRecord 样例程序

把mnist数据保存为tfrecord格式：

#!coding:utf8

import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import numpy as np

def _int64_feature(value):
    return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))

def _bytes_feature(value):
    return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))

mnist = input_data.read_data_sets('D:\files\tfmnist', one_hot=True)  # 确定label是由0、1组成的数组，还是单个整数。
images = mnist.train.images  # (55000, 784)
labels = mnist.train.labels

pixels = images.shape[1]  # 784
num_examples = mnist.train.num_examples

filename = 'D:\files\tfyangxl.tfrecords'
writer = tf.python_io.TFRecordWriter(filename)
for index in range(num_examples):
    image_raw = images[index].tostring()
    example = tf.train.Example(features=tf.train.Features(feature={
        'pixels': _int64_feature(pixels),
        'label': _int64_feature(np.argmax(labels[index])),
        'image_raw': _bytes_feature(image_raw)
    }))
    writer.write(example.SerializeToString())
writer.close()

读取tfrecord文件：

import tensorflow as tf
import matplotlib.pyplot as plt
import numpy as np

reader = tf.TFRecordReader()

# 创建一个队列来维护输入文件列表
file_queue = tf.train.string_input_producer(['/home/yangxl/files/mnist.tfrecords'])

# 从文件中读取一个样例; 一次性读取多个样例使用read_up_to函数
_, serialized_example = reader.read(file_queue)  # tensor
# 解析样例; 一次性解析多个样例使用parse_example函数
features = tf.parse_single_example(
    serialized_example,
    features={
        # """
        # tf提供了两种属性解析方法，一种是定长tf.FixedLenFeature，解析结果为一个Tensor;
        # 另一种是变长tf.VarLenFeature，解析结果为SparseTensor，用于处理稀疏数据。
        # 这里解析数据的格式需要和写入数据的格式一致。
        # """
        # 使用多行注释会报错：`KeyError: 'pixels'`, 我擦泪...

        # 解析时的键需要与保存时的键一致
        'pixels': tf.FixedLenFeature([], tf.int64),
        'label': tf.FixedLenFeature([], tf.int64),
        'image_raw': tf.FixedLenFeature([], tf.string)
    }
)

# decode_raw可以把字符串解析为图像对应的像素数组
# cast转换数据类型
image = tf.decode_raw(features['image_raw'], tf.uint8)
label = tf.cast(features['label'], tf.int32)
pixels = tf.cast(features['pixels'], tf.int32)

with tf.Session() as sess:
    coord = tf.train.Coordinator()
    threads = tf.train.start_queue_runners(sess=sess, coord=coord)

    for i in range(15):  # 每次执行sess.run()都会从队列中取出一个样例，这样就会导致之后处理时可能不是同一个样例，没注意这个问题，这两天就卡在这上面了
        image_value, label_value, pixels_value = sess.run([image, label, pixels])
        print(label_value)

        # 可视化, 可视化之前需要把一维数组转为二维数组
        image_value = np.reshape(image_value, [28, 28])
        plt.imshow(image_value)
        plt.show()

7.2 图像数据处理

一张RGB图像可以看成一个三维矩阵，矩阵中的每个数字表示图像上不同位置、不同颜色的亮度。图像在存储时，并不是直接记录这些矩阵中的数字，而是记录经过压缩编码之后的结果。所以要将一张图像还原成一个三维矩阵，需要解码的过程。TF提供了对jpeg、png格式图像的编码/解码函数。

图像编码、解码

 1 import tensorflow as tf
 2 import matplotlib.pyplot as plt
 3 
 4 with tf.gfile.GFile('/home/error/cat.jpg', 'rb') as f:
 5     image_raw_data = f.read()
 6 # 对jpeg格式的图像进行解码, 得到图像对应的三维矩阵,　得到一个tensor
 7 image_data = tf.image.decode_jpeg(image_raw_data)  # 得到一个tensor, (1797, 2673, 3)  dtype=uint8
 8 
 9 # 编码
10 encoded_image = tf.image.encode_jpeg(image_data)  # 得到一个tensor
11 
12 
13 with tf.Session() as sess:
14     plt.imshow(sess.run(image_data))
15     plt.show()
16 
17     with tf.gfile.GFile('/home/error/cat_bk.jpg', 'wb') as f:
18         f.write(sess.run(encoded_image))

图像大小调整

图像大小是不固定的，但神经网络输入节点的个数是固定的，所以在将图像的像素作为输入提供给神经网络之前，需要先将图像的大小统一。

 1 import tensorflow as tf
 2 import matplotlib.pyplot as plt
 3 
 4 with tf.gfile.FastGFile('/home/error/cat.jpg', 'rb') as f:
 5     image_raw_data = f.read()
 6 # 对jpeg格式的图像进行解码, 得到图像对应的三维矩阵,　得到一个tensor
 7 image_data = tf.image.decode_jpeg(image_raw_data)  # (1797, 2673, 3)
 8 
 9 # 在图像处理之前将图像由uint8转为实数类型
10 img_data = tf.image.convert_image_dtype(image_data, dtype=tf.float32)
11 
12 resized_image = tf.image.resize_images(img_data, [300, 300], method=0)  # method取值为0~3
13 
14 with tf.Session() as sess:
15     print(resized_image)
16 
17     plt.imshow(sess.run(resized_image))
18     plt.show()

裁剪和填充，居中

1 # 裁剪, 如果原始图像的尺寸大于目标图像, 会自动截取原始图像居中的部分
2 croped = tf.image.resize_image_with_crop_or_pad(img_data, 1000, 1000)
3 # 填充, 如果目标图像的尺寸大于原始图像, 会自动在原始图像的四周填充全0背景
4 paded = tf.image.resize_image_with_crop_or_pad(img_data, 3000, 3000)

按比例裁剪，居中

1 # 按比例截取原始图像居中的部分, 比例为(0, 1]之间的实数
2 central_cropped = tf.image.central_crop(img_data, 0.5)

在指定区域进行裁剪和填充

1 # 裁剪给定区域的图像, 该函数对给出的尺寸有一定的要求, 否则报错
2 croped_bound = tf.image.crop_to_bounding_box(img_data, 500, 500, 800, 800)
3 
4 # 填充, 图像从(500, 500)开始, 左侧和上侧全0背景, 显示图像后, 继续是全0背景。该函数对给出的尺寸有一定的要求, 否则报错
5 paded_bound = tf.image.pad_to_bounding_box(img_data, 500, 500, 2500, 3500)  # offset_height + img_height < target_height

图像翻转

图像的翻转不应该影响识别的效果，因此在训练图像识别神经网络时，可以随机地翻转训练图像，这样训练得到的模型可以识别不同角度的实体。

 1 # 上下翻转
 2 flipped = tf.image.flip_up_down(img_data)
 3 # 左右翻转
 4 flipped = tf.image.flip_left_right(img_data)
 5 # 沿对角线翻转, 主对角线
 6 flipped = tf.image.transpose_image(img_data)
 7 
 8 # 以50%的概率上下翻转图像
 9 flipped = tf.image.random_flip_up_down(img_data)
10 # 以50%的概率左右翻转图像
11 flipped = tf.image.random_flip_left_right(img_data)

图像色彩的调整

调整图像的亮度、对比度、饱和度和色相都不会影响识别结果，因此可以随机地调整这些属性

调整亮度

 1 # 调整亮度, 负号是调暗, -1为黑屏, 正数是调亮, 1为白屏
 2 adjusted = tf.image.adjust_brightness(img_data, -0.5)
 3 
 4 # 在[-max_delta, max_delta]范围内随机调整图像亮度
 5 adjusted = tf.image.random_brightness(img_data, 1)
 6 
 7 
 8 # 截断
 9 # 色彩调整的API可能导致像素的实数值超出0.0~1.0的范围，因此在最终输出图像前需要将其截断在0.0~1.0范围内，
10 # 否则不仅图像不能正常可视化，以此为输入的神经网络的训练质量也可能会受到影响。
11 #　如果对图像有多项处理，那么截断应该在所有处理完成之后进行
12 adjusted = tf.clip_by_value(adjusted, 0.0, 1.0)

调整对比度

1 # 调整对比度，将对比度减少到0.5倍
2 adjusted = tf.image.adjust_contrast(img_data, 0.5)
3 # 调整对比度，将对比度增加5倍
4 adjusted = tf.image.adjust_contrast(img_data, 5)
5 # 将对比度在[0.5, 5]范围内随机调整
6 adjusted = tf.image.random_contrast(img_data, 0.5, 5)

调整色相

1 # 分别取值[0.1, 0.3, 0.6, 0.9], 色彩从绿变为蓝，又变为红
2 adjusted = tf.image.adjust_hue(img_data, 0.9)
3 # 取值在[0.0, 0.5]之前
4 adjusted = tf.image.random_hue(0, 0.8)

调整饱和度

1 # 调整饱和度
2 adjusted = tf.image.adjust_saturation(img_data, -5)  # 饱和度-5(+5就是加5)
3 # 在[-5, 5]范围内随机调整饱和度
4 tf.image.random_saturation(img_data, -5, 5)

注意：对于色相、饱和度，需要输入数据的channels为3，例如mnist数据就不行，亮度、对比度没有限制。

将图像标准化

即将图像的亮度均值变为0，方差变为1

1 adjusted = tf.image.per_image_standardization(img_data)

处理标注框

 1 import tensorflow as tf
 2 import matplotlib.pyplot as plt
 3 
 4 with tf.gfile.GFile('/home/error/cat.jpg', 'rb') as f:
 5     image_raw_data = f.read()
 6 # 对jpeg格式的图像进行解码, 得到图像对应的三维矩阵,　得到一个tensor
 7 decoded_image_data = tf.image.decode_jpeg(image_raw_data)  # (1797, 2673, 3)
 8 
 9 # 在图像处理之前将图像由uint8转为实数类型
10 converted_image_data = tf.image.convert_image_dtype(decoded_image_data, dtype=tf.float32)
11 
12 # 把图像缩小一些，让标注框更清楚
13 resized_image_data = tf.image.resize_images(converted_image_data, [180, 267])
14 
15 # 输入是一个batch的数据，也就是多张图片组成的四维矩阵，所以需要加1个维度
16 expanded_image_data = tf.expand_dims(resized_image_data, 0)  # (1, 180, 267, ?)
17 # 标注框，数值为比例，秩为3，设置了两个标注框，为啥秩为3呢??
18 boxes = tf.constant([[[0.05, 0.05, 0.9, 0.7], [0.35, 0.47, 0.5, 0.56]]])
19 drawn_image_data = tf.image.draw_bounding_boxes(expanded_image_data, boxes)
20 
21 adjusted = tf.clip_by_value(drawn_image_data[0], 0.0, 1.0)
22 
23 with tf.Session() as sess:
26 
27     plt.imshow(sess.run(adjusted))
28     plt.show()

随机截取图像

随机截取图像上有信息含量的部分也是一种提高模型健壮性的方式。这样可以使训练得到的模型不受识别物体大小的影响。

import tensorflow as tf
import matplotlib.pyplot as plt

with tf.gfile.FastGFile('/home/yangxl/files/cat.jpg', 'rb') as f:
    image_raw_data = f.read()
# 对jpeg格式的图像进行解码, 得到图像对应的三维矩阵,　得到一个tensor
decoded_image_data = tf.image.decode_jpeg(image_raw_data)  # (1797, 2673, 3)
# 在图像处理之前将图像由uint8转为实数类型
converted_image_data = tf.image.convert_image_dtype(decoded_image_data, dtype=tf.float32)
# 把图像缩小一些，让标注框更清楚
resized_image_data = tf.image.resize_images(converted_image_data, [180, 267], method=1)

# 输入是一个batch的数据，也就是多张图片组成的四维矩阵，所以需要加1个维度
expanded_image_data = tf.expand_dims(resized_image_data, 0)  # (1, 180, 267, ?)
# 标注框，数值为比例，秩为3，设置了两个标注框
boxes = tf.constant([[[0.05, 0.05, 0.9, 0.7], [0.35, 0.47, 0.5, 0.56]]])

# 扩维之前的shape
begin, size, bbox_for_draw = tf.image.sample_distorted_bounding_box(
    tf.shape(resized_image_data), bounding_boxes=boxes,
    min_object_covered=0.4
)
image_with_box = tf.image.draw_bounding_boxes(expanded_image_data, bbox_for_draw)

adjusted = tf.clip_by_value(image_with_box[0], 0.0, 1.0)
distorted_image = tf.slice(adjusted, begin, size)

with tf.Session() as sess:
    plt.imshow(sess.run(distorted_image))  # 如果不使用slice, 像这样plt.imshow(sess.run(adjusted)), 可视化结果为不截取只随机标注
    plt.show()
# 其实就多了两行：
# sample_distorted_bounding_box和slice

7.2.2 图像预处理完整样例

因为调整亮度、对比度、饱和度和色相的顺序会影响最后得到的结果，所以可以定义多种不同的顺序。具体使用哪一种可以随机选定。这样可以进一步降低无关因素对模型的影响。

import tensorflow as tf
import numpy as np
import matplotlib.pyplot as plt

def distort_color(image, color_ordering=0):
    if color_ordering == 0:
        image = tf.image.random_brightness(image, max_delta=32./255.)
        image = tf.image.random_saturation(image, lower=0.5, upper=1.5)
        image = tf.image.random_hue(image, max_delta=0.2)
        image = tf.image.random_contrast(image, lower=0.5, upper=1.5)
    elif color_ordering == 1:
        image = tf.image.random_saturation(image, lower=0.5, upper=1.5)
        image = tf.image.random_brightness(image, max_delta=32. / 255.)
        image = tf.image.random_contrast(image, lower=0.5, upper=1.5)
        image = tf.image.random_hue(image, max_delta=0.2)
    return tf.clip_by_value(image, 0.0, 1.0)

# 给定一张解码后的图像、目标图像的尺寸以及图像上的标注框，此函数可以对给出的图像进行预处理。
# 这个函数的输入图像是图像识别问题中的原始训练数据，而输出是神经网络模型的输入层。
# 注意，只处理模型的训练数据，对于预测数据，一般不需要使用随机变换的步骤。
def preprocessed_for_train(image, height, width, bbox):
    # 如果没有提供标注框，则认为整个图像就是需要关注的部分。
    if bbox is None:
        bbox = tf.constant([0.0, 0.0, 1.0, 1.0], dtype=tf.float32, shape=[1, 1, 4])  # 秩为3

    # 转换图像张量的类型
    if image.dtype != tf.float32:
        image = tf.image.convert_image_dtype(image, tf.float32)

    # 随机截取图像，减少需要关注的物体大小对图像识别算法的影响。
    bbox_begin, bbox_size, bbox_for_draw = tf.image.sample_distorted_bounding_box(tf.shape(image), bounding_boxes=bbox)
    distorted_image = tf.slice(image, bbox_begin, bbox_size)

    # 调整大小
    distorted_image = tf.image.resize_images(distorted_image, [height, width], method=np.random.randint(4))

    # 随机翻转
    distorted_image = tf.image.random_flip_left_right(distorted_image)

    # 调整色彩
    distorted_image = distort_color(distorted_image, np.random.randint(2))

    return distorted_image


image_raw_data = tf.gfile.GFile('/home/yangxl/files/cat.jpg', 'rb').read()
with tf.Session() as sess:
    img_data = tf.image.decode_jpeg(image_raw_data)
    boxes = tf.constant([[[0.05, 0.05, 0.9, 0.7], [0.35, 0.47, 0.5, 0.56]]])

    # 运行6次获得6中不同的图像
    for i in range(6):
        result = preprocessed_for_train(img_data, 299, 299, boxes)
        plt.imshow(result.eval())
        plt.show()

7.3 多线程输入数据处理框架

虽然图像预处理方法可以减小无关因素对图像识别模型效果的影响，但是这些复杂的预处理过程也会减慢整个训练过程。为了避免图像预处理成为神经网络模型训练效率的瓶颈，tensorflow提供了一套多线程处理输入数据的框架。

tensorflow中，队列不仅是一种数据结构，更提供了多线程机制。队列也是tensorflow多线程输入数据处理框架的基础。

7.3.1介绍队列和多线程

7.3.2介绍前三步

7.3.3介绍最后一步

7.3.4完整示例

7.4介绍数据集

7.3.1 队列与多线程

队列和变量类似，都是计算图上有状态的节点。其他状态节点可以修改它们的状态。

队列操作：

import tensorflow as tf

# 先进先出队列
q = tf.FIFOQueue(2, 'int32')
# 队列初始化
init = q.enqueue_many([[0, 10], ])  # 这个至少要有两层括号，否则报错：Shape () must have rank at least 1

x = q.dequeue()
y = x + 1
q_inc = q.enqueue([y])  # 可以没有括号

with tf.Session() as sess:
    init.run()  #　队列初始化需要明确调用
    for i in range(5):
        # 10, 1   1, 11   11, 2   2, 12   12, 3
        sess.run(q_inc)

    print(sess.run(x))  # 12
    print(sess.run(x))  # 3

tf提供了FIFOQueue和RandomShuffleQueue两种队列。FIFOQueue是先进先出队列，RandomShuffleQueue会将队列中的元素打乱，每次出队列操作得到的是从当前队列所有元素中随机选择的一个。在训练神经网络时希望每次使用的训练数据尽量随机，RandomShuffleQueue就提供了这样的功能。

tf提供了tf.train.Coordinator和tf.QueueRunner两个类来完成多线程协同的功能。

tf.train.Coordinator主要用于协同多个线程一起停止，并提供了should_stop、request_stop和join三个函数。在启动线程之前，需要先声明一个tf.train.Coordinator类，并将这个类传入每个创建的线程中。启动的线程需要一直查询tf.Coordinator类中提供的should_stop函数，当这个函数的返回值为True时，则当前线程退出。每个线程都可以通过调用request_stop函数来通知其他线程退出，即当某一个线程调用request_stop函数之后，should_stop函数的返回值被设置为True，这样其他线程就可以同时退出了。

import tensorflow as tf
import numpy as np
import threading
import time

def MyLoop(coord, worker_id):
    while not coord.should_stop():
        if np.random.rand() < 0.05:
            print('Stoping from id: %d
' % worker_id)
            coord.request_stop()
        else:
            print('working on id: %d
' % worker_id)
        time.sleep(1)


coord = tf.train.Coordinator()
threads = [threading.Thread(target=MyLoop, args=(coord, i)) for i in range(5)]

for t in threads:
    t.start()

# 等待所有线程退出
coord.join(threads)

tf.train.QueueRunner主要用于启动多个线程来操作同一个队列。启动的线程可以通过tf.Coordinator类来统一管理。

queue = tf.FIFOQueue(100, 'float')
enqueue_op = queue.enqueue([tf.random_normal([1])])

# 启动5个线程来操作队列，每个线程中运行的是enqueue_op
qr = tf.train.QueueRunner(queue, [enqueue_op] * 5)
# 将qr加入到计算图指定的集合中，如果没有指定集合则默认加到tf.GraphKeys.QUEUE_RUNNERS
tf.train.add_queue_runner(qr)

out_tensor = queue.dequeue()

with tf.Session() as sess:
    coord = tf.train.Coordinator()
    # 使用tf.train.QueueRunner时，需要明确调用tf.train.start_queue_runners来启动所有线程。
    # tf.train.start_queue_runners会默认启动tf.GraphKeys.QUEUE_RUNNERS集合中的所有QueueRunner。
    # 因为这个函数只支持启动指定集合中的QueueRunner，所以tf.train.add_queue_runner和tf.train.start_queue_runners会指定同一个集合。
    threads = tf.train.start_queue_runners(sess=sess, coord=coord)

    for i in range(3):
        print(sess.run(out_tensor))

    coord.request_stop()
    coord.join(threads)

7.3.2 输入文件队列

使用TF中的队列管理输入文件列表。

虽然一个TFRecord文件中可以保存多个训练样例，但是当训练数据量较大时，可以将数据分成多个TFRecord文件来提高处理效率。tensorflow提供了tf.train.match_filenames_once函数来获取符合一个正则表达式的所有文件，得到的文件列表可以通过tf.train.string_input_producer函数进行有效的管理。注意，在使用tf.train.match_filenames_once时需要初始化一些变量，tf.local_variables_initizliaer().run()。

tf.train.string_input_producer函数会使用初始化时提供的文件列表创建一个输入队列，输入队列中原始的元素为文件列表中的所有文件，创建好的输入队列可以作为文件读取函数的参数。

每次调用文件读取函数时，该函数会先判断当前是否已有打开的文件可读，如果没有或者打开的文件已经读完，这个函数就会从输入队列中出队一个文件并从这个文件中读取数据。

reader = tf.TFRecordReader()
# 创建输入队列
filename_queue = tf.train.string_input_producer(['/home/error/output.tfrecords'])
# 读取样例
_, serializd_example = reader.read(filename_queue)

通过设置shuffle参数，tf.train.string_input_producer函数支持随机打乱文件列表中文件出队的顺序。随机打乱文件顺序以及加入输入队列的过程会跑在一个单独的线程上，这样不会影响获取文件的速度。

tf.train.string_input_producer生成的输入队列可以同时被多个文件读取线程操作，而且输入队列会将队列中的文件均匀地分配给不同的线程，不出现有些文件被处理多次而有些文件还没被处理的情况。

当一个输入队列中的所有文件都被处理后，它会将初始化时提供的文件列表中的文件全部重新加入队列。可以通过设置num_epochs参数来限制加载初始文件列表的最大轮数。当所有文件都已经被使用了设定的轮数后，如果继续尝试读取新的文件，输入队列会报错：OutOfRange。在测试神经网络时，因为所有测试数据只需要使用一次，所有可以将num_epochs设置为1，这样在计算完一轮之后程序将自动停止。

在展示tf.train.match_filenames_once和tf.train.string_input_producer函数的使用方法之前，先生成两个TFRecords文件，

num_shards = 2
instances_per_shard = 2

def _int64_feature(value):
    return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))

# 生成两个文件，每个文件保存2个样例
for i in range(num_shards):
    filename = '/home/yangxl/files/tfrecords/data.tfrecords-%.5d-of-%.5d' % (i, num_shards)  # 书上是带括号的，('...')
    writer = tf.python_io.TFRecordWriter(filename)

    for j in range(instances_per_shard):
        example = tf.train.Example(features=tf.train.Features(feature={
            'i': _int64_feature(i),
            'j': _int64_feature(j)
        }))
        writer.write(example.SerializeToString())
    writer.close()

读取多个TFRecord文件，获取样例数据，

files_list = tf.train.match_filenames_once('/home/error/tfrecord/data.tfrecords-*')  # 参数为正则表达式
filename_queue = tf.train.string_input_producer(files_list, num_epochs=2, shuffle=True)

reader = tf.TFRecordReader()
_, serialized_example = reader.read(filename_queue)
features = tf.parse_single_example(
    serialized_example,
    features={
        'i': tf.FixedLenFeature([], tf.int64),
        'j': tf.FixedLenFeature([], tf.int64)
    }
)

with tf.Session() as sess:
    # 虽然在本段程序中没有声明任何变量，但是使用tf.train.match_filenames_once函数时需要初始化一些变量
    tf.local_variables_initializer().run()

    print(sess.run(files_list))

    coord = tf.train.Coordinator()
　　　　# tf.train.string_input_producer创建文件队列也是调用了FIFOQueue、enqueue_many、QueueRunner、add_queue_runner这几个操作，所以需要明确调用启动线程的语句。
    threads = tf.train.start_queue_runners(sess=sess, coord=coord)

    for i in range(6):
        print(sess.run([features['i'], features['j']]))

    coord.request_stop()
    coord.join(threads)

7.3.3 组合训练数据(batching)

从文件列表中读取单个样例，将单个样例进行预处理，将经过预处理的单个样例组织成batch，提供给神经网络输入层。tensorflow提供了tf.train.batch和tf.train.shuffle_batch函数来将单个的样例组织成batch形式输出。这两个函数都会生成一个队列，队列的入队操作是生成单个样例的方法，而每次出队得到的是一个batch的样例，二者唯一的区别在于是否将数据顺序打乱。

tf.train.batch和tf.train.shuffle_batch的使用方法，

files_list = tf.train.match_filenames_once('/home/error/tfrecord/data.tfrecords-*')  # 参数为正则表达式
filename_queue = tf.train.string_input_producer(files_list, shuffle=False)

reader = tf.TFRecordReader()
_, serialized_example = reader.read(filename_queue)
features = tf.parse_single_example(
    serialized_example,
    features={
        'i': tf.FixedLenFeature([], tf.int64),
        'j': tf.FixedLenFeature([], tf.int64)
    }
)
example, label = features['i'], features['j']

batch_size = 5
# 队列中最多可以存储的样例个数。一般来说，队列的大小与每个batch的大小相关。
capacity = 1000 + 3 * batch_size


# 使用batch来组合样例。
   # capacity给出了队列的最大容量，当队列长度等于容量时，tensorflow暂停入队操作，而只是等待元素出队；当队列长度小于容量时，tensorflow自动重新启动入队操作。
example_batch, label_batch = tf.train.batch(
    [example, label], batch_size=batch_size, capacity=capacity
)

with tf.Session() as sess:
    # 虽然在本段程序中没有声明任何变量，但是使用tf.train.match_filenames_once函数时需要初始化一些变量
    tf.local_variables_initializer().run()
    print(sess.run(files_list))

    coord = tf.train.Coordinator()
    threads = tf.train.start_queue_runners(sess=sess, coord=coord)

    for i in range(3):
        cur_example_batch, cur_label_batch = sess.run([example_batch, label_batch])
        print(cur_example_batch, cur_label_batch)

    coord.request_stop()
    coord.join(threads)

tf.train.batch和tf.train.shuffle_batch的区别在于，shuffle_batch多一个参数min_after_dequeue，限制了出队时队列中元素的最少个数。当队列中元素太少时，随机打乱样例顺序的作用就不大了。当队列中元素不够时，出队操作将等待更多的元素入队才会完成。

# min_after_dequeue参数限制了出队时最少元素的个数来保证随机打乱顺序的作用。当出队函数被调用但是队列中元素不够时，出队操作将等待更多的元素入队才会完成。
example_batch, label_batch = tf.train.shuffle_batch(
    [example, label], batch_size=batch_size, capacity=capacity, min_after_dequeue=30
)

这两个函数除了可以将单个训练数据整理成输入batch，还提供了并行化处理输入数据的方法。通过设置num_threads参数，可以指定多个线程同时执行入队操作。入队操作就是数据读取以及预处理过程。当num_threads大于1时，多个线程会同时读取一个文件中的不同样例并进行预处理。

如果需要多个线程处理不同文件中的样例，可以使用tf.train.batch_join和tf.train.shuffle_batch_join函数。此函数会从输入文件队列中获取不同文件分配给不同的线程。一般来说，输入文件队列时通过tf.train.string_input_producer函数生成的，这个函数会平均分配文件以保证不同文件中的数据会尽量平均地使用。

tf.train.shuffle_batch和tf.train.shuffle_batch_join都可以完成多线程并行的方式来进行数据处理，但它们各有优劣。对于shuffle_batch，不同线程会读取同一个文件，如果一个文件中的样例比较相似(比如都属于同一个类别)，那么神经网络的训练效果有可能受到影响。所以使用shuffle_batch时，需要尽量将同一个TFRecord文件中的样例随机打乱。而使用shuffle_batch_join时，不同线程会读取不同文件，如果读取数据的线程数比文件数还多，那么多个线程可能会读取同一个文件中相近部分的数据。而且多个线程读取多个文件可能导致过多的硬盘寻址，从而降低读取效率。

3个shuffle：string_input_producer中的shuffle打乱队列中的文件；shuffle_batch中的shuffle打乱队列中的元素；shuffle_batch_join中的shuffle。

7.3.4 输入数据处理框架

事先准备，

把mnist数据集转为10个TFRecord文件，

import tensorflow as tf
import numpy as np
from tensorflow.examples.tutorials.mnist import input_data
import math

mnist = input_data.read_data_sets('/home/error/MNIST_DATA/', dtype=tf.uint8, one_hot=True)

images = mnist.train.images
num_examples = mnist.train.num_examples
labels = mnist.train.labels
pixels = images.shape[1]
height = width = int(math.sqrt(pixels))

num_shards = 10
# 每个文件有多少数据
instances_per_shard = int(mnist.train.num_examples / num_shards)  # 5500

def _int64_feature(value):
    return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))

def _bytes_feature(value):
    return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))


for i in range(num_shards):
    filename = '/home/error/tfrecord_mnist/data.tfrecords-%.5d-of-%.5d' % (i, num_shards)
    writer = tf.python_io.TFRecordWriter(filename)

    for j in range(instances_per_shard * i, instances_per_shard * (i+1)):
        example = tf.train.Example(features=tf.train.Features(feature={
            'image': _bytes_feature(images[j].tostring()),  # image[j]为长度为784的一维数组
            'label': _int64_feature(np.argmax(labels[j])),
            'height': _int64_feature(height),
            'width': _int64_feature(width),
            'channels': _int64_feature(1)
        }))
        writer.write(example.SerializeToString())
    writer.close()

预处理mnist数据遇到的问题：

1). 判断类型

2). channels

    # 随机翻转
    distorted_image = tf.image.random_flip_left_right(distorted_image)
    # 调整色彩
    distorted_image = distort_color(distorted_image, np.random.randint(2))

    return distorted_image

######################

import tensorflow as tf
import matplotlib.pyplot as plt
import numpy as np
from meng42 import preprocessed_for_train
from tensorflow.examples.tutorials.mnist import input_data


mnist = input_data.read_data_sets('/home/yangxl/files/mnist/', dtype=tf.uint8, one_hot=True)
image = mnist.train.images[4]
image = image.reshape([28, 28, 1])

# 预处理过程中，`if image.dtype != tf.float32:`报错：TypeError: data type not understood
# 原因是image.dtype的类型为numpy, 而tf.float32的类型为tensor, 比较之前必须先统一类型。
image = tf.constant(image)

# 定义神经网络的输入大小
image_size = 28
# 预处理
distort_image = preprocessed_for_train(image, image_size, image_size, None)
distort_image = tf.squeeze(distort_image, axis=2)

with tf.Session() as sess:
    tf.global_variables_initializer().run()

    distort_image_val = sess.run(distort_image)
    print(distort_image_val.shape)
    plt.imshow(distort_image_val)
    plt.show()

完整示例：

import tensorflow as tf
from meng42 import preprocessed_for_train
import mnist_inference
import os


files = tf.train.match_filenames_once(pattern='/home/yangxl/files/mnist_tfrecords/mnist.tfrecords-*')
filename_queue = tf.train.string_input_producer(files, shuffle=False, num_epochs=1)

reader = tf.TFRecordReader()
_, serialized_example = reader.read(filename_queue)

features = tf.parse_single_example(serialized_example, features={
    'image': tf.FixedLenFeature([], tf.string),
    'label': tf.FixedLenFeature([], tf.int64),
    'height': tf.FixedLenFeature([], tf.int64),
    'width': tf.FixedLenFeature([], tf.int64),
    'channels': tf.FixedLenFeature([], tf.int64)
})

image, label = features['image'], features['label']
height, width = features['height'], features['width']
channels = features['channels']

decoded_image = tf.decode_raw(image, tf.uint8)  # shape=(?,)
decoded_image = tf.reshape(decoded_image, [28, 28, 1])  # shape=(28, 28, 1)

# 定义神经网络的输入大小
image_size = 28
# 预处理
distort_image = preprocessed_for_train(decoded_image, image_size, image_size, None)  # shape=(28, 28, ?)
distort_image = tf.reshape(distort_image, [28, 28, 1])  # 预处理过程损坏了shape，会在`shuffle_batch`时报错。

min_after_dequeue = 1000
batch_size = 100
capacity = min_after_dequeue + 3 * batch_size
image_batch, label_batch = tf.train.shuffle_batch([distort_image, label], batch_size, capacity, min_after_dequeue)

# 训练
BATCH_SIZE = 100

LEARNING_RATE_BASE = 0.9
LEARNING_RATE_DECAY = 0.9
REGULARIZATION_RATE = 0.0001  # lambda
TRAINING_STEPS = 20000
MOVING_AVERAGE_DACAY = 0.99

MODEL_SAVE_PATH = '/home/yangxl/files/save_model2'
MODEL_NAME = 'yangxl.ckpt'


def train(image_batch, label_batch):
    # 因为从池化层到全连接层要进行reshape，所以不能为shape[0]不能为None。
    x = tf.placeholder(tf.float32, [BATCH_SIZE, mnist_inference.IMAGE_SIZE, mnist_inference.IMAGE_SIZE, mnist_inference.NUM_CHANNELS], 'x-input')
    y_ = tf.placeholder(tf.int64, [BATCH_SIZE], 'y-input')
    # 因为从tfrecords文件中读取的label.shape=(), 所以这里进行了相应调整(y_以及用到y_的节点，测试代码也要对应)。

    # 正则化
    regularizer = tf.contrib.layers.l2_regularizer(REGULARIZATION_RATE)
    y = mnist_inference.inference(x, True, regularizer)

    global_step = tf.Variable(0, trainable=False)

    # 滑动平均
    variables_averages = tf.train.ExponentialMovingAverage(MOVING_AVERAGE_DACAY, global_step)
    variables_averages_op = variables_averages.apply(tf.trainable_variables())

    cross_entropy = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=y, labels=y_)
    cross_entropy_mean = tf.reduce_mean(cross_entropy)
    loss = cross_entropy_mean + tf.add_n(tf.get_collection('losses'))

    # learning_rate = tf.train.exponential_decay(LEARNING_RATE_BASE, global_step, 10000 / BATCH_SIZE, LEARNING_RATE_DECAY, staircase=True)
    learning_rate = 0.01
    train_step = tf.train.GradientDescentOptimizer(learning_rate).minimize(loss, global_step)
    with tf.control_dependencies([train_step, variables_averages_op]):
        train_op = tf.no_op(name='train')

    with tf.Session() as sess:
        tf.local_variables_initializer().run()
        tf.global_variables_initializer().run()

        coord = tf.train.Coordinator()
        threads = tf.train.start_queue_runners(sess=sess, coord=coord)

        saver = tf.train.Saver()

        image_batch_val, label_batch_val = sess.run([image_batch, label_batch])
        for i in range(TRAINING_STEPS):
            _, loss_value, step = sess.run([train_op, loss, global_step], feed_dict={x: image_batch_val, y_: label_batch_val})

            if i % 1000 == 0:
                print('after %d training steps, loss on training batch is %g ' % (i, loss_value))
                saver.save(sess, os.path.join(MODEL_SAVE_PATH, MODEL_NAME), global_step=global_step)

        coord.request_stop()
        coord.join(threads)


if __name__ == '__main__':
    train(image_batch, label_batch)

把flower文件转为TFRecord文件，

import tensorflow as tf
import os
import glob
from tensorflow.python.platform import gfile
import numpy as np

INPUT_DATA = '/home/error/flower_photos'  # 输入文件


VALIDATION_PERCENTAGE = 10
TEST_PERCENTAGE = 10

def _int64_feature(value):
    return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))

def _bytes_feature(value):
    return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))

def create_image_lists(sess):
    sub_dirs = [x[0] for x in os.walk(INPUT_DATA)]  # 当前目录和子目录
    # print(sub_dirs)
    is_root_dir = True

    current_labels = 0

    # 读取所有子目录
    for sub_dir in sub_dirs:
        if is_root_dir:  # 把第一个排除了
            is_root_dir = False
            continue

        # 获取一个子目录中所有的图片文件
        extensions = ['jpg', 'jpeg', 'JPG', 'JPEG']
        file_list = []
        dir_name = os.path.basename(sub_dir)  # '/'最后面的部分
        print(dir_name)
        for extension in extensions:
            file_glob = os.path.join(INPUT_DATA, dir_name, '*.' + extension)
            file_list.extend(glob.glob(file_glob))  # glob.glob返回一个匹配该模式的列表, glob和os配合使用来操作文件
        if not file_list:
            continue

        OUTPUT_DATA = '/home/error/inception_v3_data/inception_v3_data_' + dir_name + '.tfrecords'  # 输出文件
        writer = tf.python_io.TFRecordWriter(OUTPUT_DATA)

        # 处理图片数据
        for file_name in file_list:
            print(file_name)
            image_raw_data = gfile.FastGFile(file_name, 'rb').read()  # 二进制数据
            image = tf.image.decode_jpeg(image_raw_data)  # tensor, dtype=uint8  333×500×3   色道0~255
            # if image.dtype != tf.float32:
            #     image = tf.image.convert_image_dtype(image, dtype=tf.float32)  # 色道值0～1
            # image = tf.image.resize_images(image, [299, 299])
            image_value = sess.run(image)  # numpy.ndarray
            # print(image_value.shape)
            height, width, channles = image_value.shape
            label = current_labels
            example = tf.train.Example(features=tf.train.Features(feature={
                'image': _bytes_feature(image_value.tostring()),
                'label': _int64_feature(np.argmax(label)),
                'height': _int64_feature(height),
                'width': _int64_feature(width),
                'channels': _int64_feature(channles)
            }))
            writer.write(example.SerializeToString())
        writer.close()

        current_labels += 1


with tf.Session() as sess:
    create_image_lists(sess)

7.4 数据集

除队列外，tensorflow提供了一套更高层的数据处理框架。在新的框架中，每一个数据来源被抽象成一个“数据集”，开发者可以以数据集为基本对象，方便地进行batching、shuffle等操作。推荐使用数据集作为输入数据的首选框架。数据集是tensorflow的核心部件。

7.4.1 数据集的基本使用方法

在数据集框架中，每个数据集代表一个数据来源：数据可能来自一个tensor，一个TFRecord文件，一个文本文件，或者经过sharding的一系列文件等。

由于训练数据通常无法全部写入内存中，从数据集中读取数据时需要使用一个迭代器按顺序进行读取，这点与队列的dequeue()操作和Reader的read()操作类似。与队列相似，数据集也是计算图上的一个节点。

示例，从一个张量创建一个数据集，

# 从数组创建数据集。不同数据来源，需要调用不同的构造方法。
input_data = [1, 2, 3, 4, 5]
dataset = tf.data.Dataset.from_tensor_slices(input_data)

# 定义一个迭代器用于遍历数据集。因为上面定义的数据集没有使用placeholder作为输入参数，所以可以使用最简单的one_shot_iterator。
iterator = dataset.make_one_shot_iterator()

x = iterator.get_next()
y = x * x

with tf.Session() as sess:
    for i in range(len(input_data)):
        print(sess.run([x, y]))

在真实项目中，训练数据通常保存在硬盘文件中。比如在自然语言处理任务中，训练数据通常以每行一条数据的形式存在文本文件中。这时可以用TextLineDataset来构造。

# 从文件创建数据集
# windows中必须要加后缀。'D:\files\tf\firsts.txt'
# 只有一个文件时，可以只传一个字符串格式的文件名。
input_files = ['/home/error/checkpoint', '/home/error/ten']
dataset = tf.data.TextLineDataset(input_files)

iterator = dataset.make_one_shot_iterator()

x = iterator.get_next()

with tf.Session() as sess:
    for i in range(20):
        print(sess.run(x))

在图像相关任务中，训练数据通常以TFRecords形式存储，这时可以用TFRecordDataset来读取数据。与文本文件不同的是，每个tfrecord都有自己不同的feature格式，因此需要提供一个parser函数来解析所读取的tfrecord格式的数据。

# 从TFRecord文件创建数据集
input_files = ['/home/error/tt.tfrecords', '/home/error/tt2.tfrecords']
dataset = tf.data.TFRecordDataset(input_files)

# map()函数表示对数据集中的每一条数据调用相应的方法。
# TFRecordDataset读出的是二进制数据，需要通过map调用parser来对二进制数据进行解析。
dataset = dataset.map(parser)

iterator = dataset.make_one_shot_iterator()
features = iterator.get_next()

with tf.Session() as sess:
    for i in range(5):  # 不能超过样例个数，否则报错
        print(sess.run(features['name']))

把上面的实例改成含有占位符的形式：

def parser(record):
    features = tf.parse_single_example(
        record,
        features={
            'name': tf.FixedLenFeature([], tf.string),
            'image': tf.FixedLenFeature([], tf.string),
            'label': tf.FixedLenFeature([], tf.int64),
            'height': tf.FixedLenFeature([], tf.int64),
            'width': tf.FixedLenFeature([], tf.int64),
            'channels': tf.FixedLenFeature([], tf.int64)
        }
    )
    return features

# 从TFRecord文件创建数据集
input_files = tf.placeholder(tf.string)
dataset = tf.data.TFRecordDataset(input_files)

# map()函数表示对数据集中的每一条数据调用相应的方法。
# TFRecordDataset读出的是二进制数据，需要通过map调用parser来对二进制数据进行解析。
dataset = dataset.map(parser)

iterator = dataset.make_initializable_iterator()
features = iterator.get_next()

with tf.Session() as sess:
    sess.run(iterator.initializer, feed_dict={input_files: ['/home/error/tt.tfrecords', '/home/error/tt2.tfrecords']})
　　 # 因为不同数据来源的数据量大小难以预知。使用while True可以把所有数据遍历一遍。
    while True:
        try:
            print(sess.run([features['name'], features['height']]))
        except tf.errors.OutOfRangeError:
            break

7.4.2 数据集的高层操作

dataset = dataset.map(parser)

对数据集中的每一条数据调用参数中指定的parser方法，经过处理后的数据重新组合成一个数据集。

distorted_image = preprocess_for_train(
    decoded_image, image_size, image_size, None
)
转为
dataset = dataset.map(
    lambda x: preprocess_for_train(x, image_size, image_size, None)
)

这样处理的优点是，返回一个新数据集，可以直接继续调用其他高层操作。

在队列框架中，预处理、shuffle、batch等操作有的在队列上进行，有的在图片张量上进行，整个处理流程在处理队列和张量的代码片段中来回切换。而在数据集操作中，所有操作都在数据集上进行。

dataset = dataset.shuffle(buffer_size)  # 随机打乱顺序
dataset = dataset.batch(batch_size)  # 将数据组合成batch

shuffle方法中的buffer_size等效于tf.train.shuffle_batch的min_after_dequeue，shuffle算法在内部使用一个缓冲区保存buffer_size条数据，每读入一个新数据时，从这个缓冲区随机选择一条数据进行输出。缓冲区越大，随机性能越好，但占用的内存也越多。

batch方法的batch_size代表要输出的每个batch由多少条数据组成。如果数据集包含多个张量，那么batch操作将对每个张量分开进行。例如，如果数据集中的每个数据是image、label两个张量，其中image的维度是[300, 300]，label的维度是[]，batch_size是128，那么经过batch操作后的数据集的每个输出将包含两个维度分别为[128, 300, 300]和[128]的张量。

dataset = dataset.repeat(N)  # 将数据集重复N份

将数据集重复N份，每一份数据被称为一个epoch。

需要指出的是，如果数据集在repeat之前进行了shuffle操作，输出的每个epoch中随机shuffle的结果并不会相同。因为repeat和map、shuffle、batch等操作一样，都只是计算图上的一个计算节点，repeat只代表重复相同的处理过程，并不会记录前一epoch的处理结果。

其他方法，

dataset.concatenate()  # 将两个数据集顺序连接起来
dataset.take(N)  # 从数据集中读取前N项数据
dataset.skip(N)  # 在数据集中跳过前N项数据
dataset.flat_map()  # 从多个数据集中轮流读取数据

与队列框架下的样例不同的是，在训练数据集之外，还另外读取了测试数据集，并对测试集进行了略微不同的预处理。在训练时，调用preprocessed_for_train对图像进行随机反转等预处理操作；而在测试时，测试集以原本的样子直接输入测试。

import tensorflow as tf
from meng42 import preprocessed_for_train

train_files = tf.train.match_filenames_once('/home/yangxl/files/mnist_tfrecords/mnist.tfrecords-*')
test_files = tf.train.match_filenames_once('/home/yangxl/files/mnist_tfrecords/mnist.tfrecords-0000[49]-of-00010')


def parser(record):
    features = tf.parse_single_example(
        record,
        features={
            'image': tf.FixedLenFeature([], tf.string),
            'label': tf.FixedLenFeature([], tf.int64),
            'height': tf.FixedLenFeature([], tf.int64),
            'width': tf.FixedLenFeature([], tf.int64),
            'channels': tf.FixedLenFeature([], tf.int64),
        }
    )

    decoded_image = tf.decode_raw(features['image'], tf.uint8)
    decoded_image = tf.reshape(decoded_image, [features['height'], features['width'], features['channels']])
    label = features['label']
    return decoded_image, label


image_size = 28
batch_size = 100
shuffle_buffer = 1000

dataset = tf.data.TFRecordDataset(train_files)
dataset = dataset.map(parser)
# lambda中的参数image、label, 返回的是一个元组(image, label)
dataset = dataset.map(lambda image, label: (preprocessed_for_train(image, image_size, image_size, None), label))
dataset = dataset.shuffle(shuffle_buffer).batch(batch_size)
# 重复NUM_EPOCHS个epoch。在7.3.4小节中TRAINING_ROUNDS指定了训练轮数，而这里指定了整个数据集重复的次数，这也间接确定了训练的轮数
NUM_EPOCHS = 10
dataset = dataset.repeat(NUM_EPOCHS)

# 虽然定义数据集时没有直接使用placeholder来提供文件地址，但是tf.train.match_filenames_once方法得到的结果与placeholder的机制类似，也需要初始化
iterator = dataset.make_initializable_iterator()
image_batch, label_batch = iterator.get_next()
print(image_batch.shape, label_batch.shape)


test_dataset = tf.data.TFRecordDataset(test_files)
# 对于测试集，不需要预处理、shuffle、repeat操作，只需用相同的parser进行解析、调整输入层大小、batch即可
test_dataset = test_dataset.map(parser)
test_dataset = test_dataset.map(lambda image, label: (tf.image.resize_images(image, [image_size, image_size]), label))
test_dataset = test_dataset.batch(batch_size)

test_iterator = test_dataset.make_initializable_iterator()
test_image_batch, test_label_batch = test_iterator.get_next()
print(test_image_batch.shape, test_label_batch.shape)

with tf.Session() as sess:
    tf.local_variables_initializer().run()
    print(test_files.eval())

ok!