TensorFlow数据读取方式：Dataset API

Datasets：一种为TensorFlow 模型创建输入管道的新方式。把数组、元组、张量等转换成DatasetV1Adapter格式

Dataset api有方法加载和操作数据，并将其输入到您的模型中。Dataset api与Estimators api很匹配。
下图是tensorflow API的完整架构图：

Datasets API是由以下图中所示的类组成：

其中：
**Dataset:**基类，包含创建和转换数据集的方法。还允许您从内存中的数据或Python中初始化数据集
生产者。
TextLineDataset: 从文本文件中读取行 (txt,csv…)。
TFRecordDataset: 从TFRecord文件读取记录。
FixedLengthRecordDataset: 从二进制文件中读取固定大小的记录。
Iterator: 提供一次访问一个数据集元素的方法
总之，Datasets API实现了从内存或者硬盘文件中加载数据组成数据集，同时对数据集进行一系列变换操作，最终将数据集提供给其他API使用的一系列功能。下面，本文就将从这三个方面对Datasets API进行介绍。

加载数据形成数据集

dataset = tf.data.Dataset.from_tensor_slices(data #数据)#按第一维度进行切分，返回dataset形式数据#shapes：切分后的维度#tf.float64：最小元素的类型

从内存或迭代器中加载数据：

数据集的单个元素包含一个或多个tf.Tensor对象，叫做分量。可能是单个张量，
张量的元组，或张量的嵌套元组。除了元组，您还可以使用collections.namedtuple或dictionary
将字符串映射到张量以表示数据集的单个元素。

# 一个张量 :
dataset1 = tf.data.Dataset.from_tensor_slices(tf.random_uniform([4, 10]))
# 张量的元组:
dataset2 =    tf.data.Dataset.from_tensor_slices((tf.random_uniform([4]),tf.random_uniform([4,100])))# 张量的元组, mnist_data是一个objection
images = mnist_data.train.images.reshape([-1, 28, 28, 1])
labels = mnist_data.train.labels
dataset = tf.contrib.data.Dataset.from_tensor_slices((images, labels))# 张量的嵌套元组:
dataset3 = tf.data.Dataset.zip((dataset1, dataset2))
# 一个集合.可命名元组或者一个字典映射字符串成张量
dataset = tf.data.Dataset.from_tensor_slices({"a": tf.random_uniform([4]),"b": tf.random_uniform([4, 100], maxval=100, dtype=tf.int32)})

读取数组

import tensorflow as tf
import numpy as np
# 创建dataset
dataset = tf.data.Dataset.from_tensor_slices(np.array([1.0, 2.0, 3.0, 4.0, 5.0]))#<DatasetV1Adapter shapes: (), types: tf.float64>
# 实例化了一个Iterator
iterator = dataset.make_one_shot_iterator()#<tensorflow.python.data.ops.iterator_ops.Iterator object at 0x000002016B501CC0>
# 从iterator里取出一个元素
one_element = iterator.get_next()# <tf.Tensor 'IteratorGetNext:0' shape=() dtype=float64>
with tf.Session() as sess:for i in range(5):print(sess.run(one_element))# 1.0# 2.0# 3.0# 4.0# 5.0

读取矩阵

import tensorflow.contrib.eager as tfe
import tensorflow as tf
import numpy as np
tfe.enable_eager_execution()
dataset = tf.data.Dataset.from_tensor_slices(np.array([[[1, 2, 3,],[4, 5, 6]],[[7, 8, 9,],[10,11,12]]]))
print("dataset:",dataset)#<DatasetV1Adapter shapes: (2, 3), types: tf.int32>
for one_element in tfe.Iterator(dataset):print(one_element)#tf.Tensor([1 2 3 4 5], shape=(5,), dtype=int32)#tf.Tensor([6  7  8  9 10], shape=(5,), dtype=int32)#tf.Tensor([11 12 13 14 15], shape=(5,), dtype=int32)

读取字典

import tensorflow.contrib.eager as tfe
import tensorflow as tf
import numpy as np
tfe.enable_eager_execution()
dataset = tf.data.Dataset.from_tensor_slices({"a": np.array([[1, 2, 3, 4, 5],[6, 7, 8, 9, 10],[11, 12, 13, 14, 15]]),  # 维度3x5"b": np.array([1.0, 2.0, 3.0])  # 维度3x0,注意：第一维度3要与上面相同})
print("dataset:",dataset)#<DatasetV1Adapter shapes: {a: (5,), b: ()}, types: {a: tf.int32, b: tf.float64}>
for one_element in tfe.Iterator(dataset):print(one_element)#{'a': <tf.Tensor: id=9, shape=(5,), dtype=int32, numpy=array([1, 2, 3, 4, 5])>, 'b': <tf.Tensor: id=10, shape=(), dtype=float64, numpy=1.0>}# {'a': <tf.Tensor: id=13, shape=(5,), dtype=int32, numpy=array([ 6,  7,  8,  9, 10])>, 'b': <tf.Tensor: id=14, shape=(), dtype=float64, numpy=2.0>}# {'a': <tf.Tensor: id=17, shape=(5,), dtype=int32, numpy=array([11, 12, 13, 14, 15])>, 'b': <tf.Tensor: id=18, shape=(), dtype=float64, numpy=3.0>}

文本文件：

filepaths = ["/var/data/file1.txt", "/var/data/file2.txt"]
dataset = tf.data.TextLineDataset(filepaths)

读取图片image

import tensorflow as tf
import glob
import os
# 函数的功能时将filename对应的图片文件读进来，并缩放到统一的大小
def _parse_function(filename,):image_string = tf.read_file(filename)image_decoded = tf.image.decode_bmp(image_string,)# 根据图片类型，选择以下函数# image_decoded = tf.image.decode_png(image_string)# image_decoded = tf.image.decode_jepg(image_string)# image_decoded = tf.image.decode_git(image_string)image_resized = tf.image.resize_images(image_decoded,[64, 64]#图像维度变化)return image_resized
def walk_type(path, file_type):paths = glob.glob(os.path.join(path,#存放图片的文件夹路径file_type # 文件类型))# path下所有file_type类型的文件的路径列表return paths
paths = walk_type('dataset/*/','*.bmp')#图片路径列表
filenames = tf.constant(paths)
dataset = tf.data.Dataset.from_tensor_slices((filenames))
dataset = dataset.map(_parse_function)
dataset = dataset.shuffle(buffer_size=1000).batch(1).repeat(10)

tfrecords文件：

filepaths = ["/data/file1.tfrecord", "/data/file2.tfrecord"]
dataset = tf.data.TFRecordDataset(filepaths)

二进制文件：

filepaths = [os.path.join(data_dir, 'data_batch_%d.bin' % i) for i in range(1, 6)]
image_bytes = image.height * image.width * image.depth
record_bytes = label_bytes + image_bytes
dataset = tf.data.FixedLengthRecordDataset(filepaths,record_bytes)

Datasets API支持一系列的变换操作

Datasets API支持 repeat、map、shuffle、batch等变换。

（1）`repeat`是将整个数据集重复多次，相当于一个或多个epoch，接受的参数数字代表repeat次数，若为空则无限重复

# Repeat infinitely.
dataset = tf.data.TFRecordDataset(filenames).repeat()

（2）`map`接收一个函数，Dataset中的每个元素都会被当作这个函数的输入，并将函数返回值作为新的Dataset。通常用于数据变换或者解析与编码文件数据。

def parser(self, serialized_example):"""解析单个tf.Example变成图片和标签张量."""features = tf.parse_single_example(serialized_example,features={'image': tf.FixedLenFeature([], tf.string),'label': tf.FixedLenFeature([], tf.int64),})image = tf.decode_raw(features['image'], tf.uint8)image.set_shape([DEPTH * HEIGHT * WIDTH])# 从[depth * height * width]变换维度到[depth, height, width].image = tf.cast(tf.transpose(tf.reshape(image, [DEPTH, HEIGHT, WIDTH]), [1, 2, 0]),tf.float32)label = tf.cast(features['label'], tf.int32)# 自定义预处理image = self.preprocess(image)return image, label
dataset = dataset.map(self.parser, num_threads=batch_size, output_buffer_size=2 * batch_size)

def decode_csv(line):parsed_line = tf.decode_csv(line, [[0.], [0.], [0.], [0.], [0]])label = parsed_line[-1:] # 最后一个元素是labeldel parsed_line[-1] # 删除最后一个元素features = parsed_line # 所有的(除了最后一个元素)都是特性d = dict(zip(feature_names, features)), labelreturn d
dataset = (tf.data.TextLineDataset(file_path) # 读文本文件.skip(1) # 跳过标题行.map(decode_csv)) # 通过应用decode_csv fn转换每个元素

（3）`shuffle`的功能为打乱dataset中的元素，它有一个参数buffersize，表示打乱时使用的buffer的大小。单位是以图片（张量）为单位，而不是byte；

# 可能的洗牌记录
if subset == 'train' or shuffle:min_queue_examples = int(Cifar10DataSet.num_examples_per_epoch(subset) * 0.4)# 确保容量足够大，以提供良好的随机洗牌。dataset = dataset.shuffle(buffer_size=min_queue_examples + 3 * batch_size)

（4）`batch`就是将多个元素组合成batch，接受一个batch_size的参数。

# Batch 组合
dataset = dataset.batch(batch_size)

迭代器

构建了表示输入数据的数据集之后，下一步是创建一个迭代器来访问来自该数据集的元素。Dataset API目前支持以下迭代器，以提高复杂程度:

one-shot :

一次性迭代器是迭代器的最简单形式，它只支持遍历数据集一次，不需要显式初始化。一次性迭代器处理现有基于队列的输入管道支持的几乎所有情况，但是它们不支持参数化。

iterator = dataset.make_one_shot_iterator()
image_batch, label_batch = iterator.get_next()
#返回image_batch label_batch

dataset = tf.data.Dataset.range(100)
iterator = dataset.make_one_shot_iterator()
next_element = iterator.get_next()
for i in range(100):value = sess.run(next_element)assert i == value

initializable

可初始化迭代器要求您运行显式迭代器。在使用它之前的初始化操作。作为对这种不便的交换，它允许您使用一个或多个tf.placeholder()张量对数据集的定义进行参数化，当您初始化迭代器时可以提供这些张量。

max_value = tf.placeholder(tf.int64, shape=[])
dataset = tf.data.Dataset.range(max_value)
iterator = dataset.make_initializable_iterator()
next_element = iterator.get_next()
# 在具有10个元素的数据集上初始化一个迭代器。
sess.run(iterator.initializer, feed_dict={max_value: 10})
for i in range(10):value = sess.run(next_element)assert i == value
# 在具有100个元素的数据集上初始化相同的迭代器。
sess.run(iterator.initializer, feed_dict={max_value: 100})
for i in range(100):value = sess.run(next_element)assert i == value

# 将训练数据加载到两个NumPy数组中
features = data["features"]
labels = data["labels"]
# 假设“features”的每一行都与“labels”对应
assert features.shape[0] == labels.shape[0]
features_placeholder = tf.placeholder(features.dtype, features.shape)
labels_placeholder = tf.placeholder(labels.dtype, labels.shape)
dataset = tf.data.Dataset.from_tensor_slices((features_placeholder, labels_placeholder))
# ['dataset'上的其他转换…...]
dataset = ...
iterator = dataset.make_initializable_iterator()
sess.run(iterator.initializer, feed_dict={features_placeholder: features,labels_placeholder: labels})

reinitializable

可重新初始化的迭代器可以从多个不同的数据集对象初始化。例如，您可能有一个训练输入管道，它使用对输入图像的随机扰动来改进泛化，还有一个验证输入管道，它评估未修改数据的预测。这些管道通常使用具有相同结构的不同数据集对象(即，对于每个组件具有相同的类型和兼容的形状)。
最后看两个比较完整的例子：
要在tf.estimator.Estimator的input_fn中使用数据集，我们建议使用Dataset.make_one_shot_iterator()。例如:

#解析数据
def parser(self,serialized_example):"""解析单个tf.Example变成图片和标签张量."""features = tf.parse_single_example(serialized_example,features={'image': tf.FixedLenFeature([], tf.string),'label': tf.FixedLenFeature([], tf.int64),})image = tf.decode_raw(features['image'], tf.uint8)image.set_shape([DEPTH * HEIGHT * WIDTH])# 从[深度*高度*宽度]到[深度、高度、宽度].image = tf.cast(tf.transpose(tf.reshape(image, [DEPTH, HEIGHT, WIDTH]), [1, 2, 0]),tf.float32)label = tf.cast(features['label'], tf.int32)# 自定义预处理.image = self.preprocess(image)return image, label
#变换数据
def preprocess(self,image):"""在[高度，宽度，深度]布局中对单个图像进行预处理."""if subset == 'train' and shuffle:# Pad 4像素在每个尺寸的特征图，在小批量完成image = tf.image.resize_image_with_crop_or_pad(image, 40, 40)image = tf.random_crop(image, [HEIGHT, WIDTH, DEPTH])image = tf.image.random_flip_left_right(image)# 因为这些操作不是可交换的，所以可以考虑将它们的操作顺序随机化。#注意:由于per_image_standardization会使平均值为零并使stddev单元为零，所以这可能不会有任何影响，请参阅tensorflow #1458。distorted_image = tf.image.random_brightness(distorted_image,max_delta=63)distorted_image = tf.image.random_contrast(distorted_image,lower=0.2, upper=1.8)# 减去平均值，除以像素的方差float_image = tf.image.per_image_standardization(distorted_image)return image
def input_fn(self,data_dir,batch_size,subset):if subset in ['train', 'validation', 'eval']:filepaths = [os.path.join(data_dir, subset + '.tfrecords')]else:raise ValueError('Invalid data subset "%s"' % subset)dataset = tf.contrib.data.TFRecordDataset(filepaths).repeat()# 解析记录.dataset = dataset.map(self.parser, num_threads=batch_size, output_buffer_size=2 * batch_size)# 可能的洗牌记录.if subset == 'train':min_queue_examples = int(Cifar10DataSet.num_examples_per_epoch(subset) * 0.4)# 确保容量足够大，以提供良好的随机变换。dataset = dataset.shuffle(buffer_size=min_queue_examples + 3 * batch_size)# Batch it up.dataset = dataset.batch(batch_size)iterator = dataset.make_one_shot_iterator()image_batch, label_batch = iterator.get_next()return image_batch, label_batch

tf.train.MonitoredTrainingSession API简化了在分布式设置中运行TensorFlow的许多方面， MonitoredTrainingSession使用tf.errors.OutOfRangeError表示训练已经完成，因此要与tf.dataAPI一起使用，我们建议使用Dataset.make_one_shot_iterator()。例如:

filenames = ["/var/data/file1.tfrecord", "/var/data/file2.tfrecord"]
dataset = tf.data.TFRecordDataset(filenames)
dataset = dataset.map(...)
dataset = dataset.shuffle(buffer_size=10000)
dataset = dataset.batch(32)
dataset = dataset.repeat(num_epochs)
iterator = dataset.make_one_shot_iterator()
next_example, next_label = iterator.get_next()
loss = model_function(next_example, next_label)
training_op = tf.train.AdagradOptimizer(...).minimize(loss)
with tf.train.MonitoredTrainingSession(...) as sess:while not sess.should_stop():sess.run(training_op)

查看dataset:DatasetV1Adapter的方法

import tensorflow as tf
import numpy as np
# 把数组转化为dataset模式：DatasetV1Adapter
data = tf.data.Dataset.from_tensor_slices(np.array([1, 2, 3, 4, 5]))
print("data:",data) #<DatasetV1Adapter shapes: (), types: tf.int32>
# 建立迭代器，并进行迭代操作（把data从DatasetV1Adapter格式转化为tensor格式）
element = data.make_one_shot_iterator().get_next()
print('element:',element) #Tensor("IteratorGetNext:0", shape=(), dtype=int32)
with tf.Session() as sess:try:while True:print(sess.run(element))except tf.errors.OutOfRangeError:print("Out range !")