tensorflow module data读取数据方式

TensorFlow全新的数据读取方式：Dataset API入门教程    

以前的读取数据的方法实在是太复杂了，要建立各种队列，所以想换成这个更为简便的方式

参照以上教程，同时结合自己的实际例子，学习如何简单高效读取数据（tensorflow api 1.4）

Module:  tf.data

@@Dataset
@@Iterator
@@TFRecordDataset
@@FixedLengthRecordDataset
@@TextLineDataset

以上均是tf.data的类，分别讲述，这五个不同的类的定义和使用方式

tf.data.Dataset

Represents a potentially large set of elements 

properties(性质)： output_shapes,  output_types 

Method(方法)：   

1. apply(fun):   Apply a transformation function to this dataset.  对数据集中的数据做额外的变换

dataset = (dataset.map(lambda x: x ** 2)
           .apply(group_by_window(key_func, reduce_func, window_size))
           .map(lambda x: x ** 3))

2. batch(batch_size):  Combines consecutive elements of this dataset into batches.

3. concatenate(dataset): Creates a Dataset by concatenating given datset with this datset, 合并另一个数据集到该数据集中

# NOTE: The following examples use `{ ... }` to represent the
# contents of a dataset.
a = { 1, 2, 3 }
b = { 4, 5, 6, 7 }

# Input dataset and dataset to be concatenated should have same
# nested structures and output types.
# c = { (8, 9), (10, 11), (12, 13) }
# d = { 14.0, 15.0, 16.0 }
# a.concatenate(c) and a.concatenate(d) would result in error.

a.concatenate(b) == { 1, 2, 3, 4, 5, 6, 7 }

4. filter(predicate) : Filters this datset according to predicate   过滤符合条件的数据

5. from_generator():  Creates a Dataset whose elements are generated by generator  自己生成数据

6. from_sparse_tensor_slices(sparse_tensor):   由sparse_tensor指明的tensor组成数据

7. from_tensor_slices(tensors): Creates a Dataset whose elements are slices of the given tensors

8. from_tensors(tensors): Creates a Dataset with a single element, comprising the given tensors

9. list_files(file_pattern): A dataset of all files matching a pattern

10. make_initializable_iterator(): Creates an Iterator for enumerating the elements of this dataset.

Note: The returned iterator will be in an uninitialized state, and you must run the iterator.initializer operation before using it:

dataset = ...
iterator = dataset.make_initializable_iterator()
# ...
sess.run(iterator.initializer)

11. make_one_shot_iterator(): Creates an Iterator for enumerating the elements of this dataset.

Note: The returned iterator will be initialized automatically. A "one-shot" iterator does not currently support re-initialization.

12.map(map_func,num_parallel_calls=None): Maps map_func across this dataset 

13. range(*args): Creates a Dataset of a step-separated range of values.

Dataset.range(5) == [0, 1, 2, 3, 4]
Dataset.range(2, 5) == [2, 3, 4]
Dataset.range(1, 5, 2) == [1, 3]
Dataset.range(1, 5, -2) == []
Dataset.range(5, 1) == []
Dataset.range(5, 1, -2) == [5, 3]

14. repeat(count=None): Repeats this dataset count times.

15. shard(num_shards,index): Creates a Dataset that includes only 1/num_shards of this dataset, is useful when running distributed training.

d = tf.data.TFRecordDataset(FLAGS.input_file)   #从一个文件中读取
d = d.shard(FLAGS.num_workers, FLAGS.worker_index)
d = d.repeat(FLAGS.num_epochs)
d = d.shuffle(FLAGS.shuffle_buffer_size)
d = d.map(parser_fn, num_parallel_calls=FLAGS.num_map_threads)

d = Dataset.list_files(FLAGS.pattern)     #多个文件
d = d.shard(FLAGS.num_workers, FLAGS.worker_index)
d = d.repeat(FLAGS.num_epochs)
d = d.shuffle(FLAGS.shuffle_buffer_size)
d = d.repeat()
d = d.interleave(tf.data.TFRecordDataset,
                 cycle_length=FLAGS.num_readers, block_length=1)
d = d.map(parser_fn, num_parallel_calls=FLAGS.num_map_threads)

16. shuffle(buffer_size,seed=None,reshuffle_each_iteration=None): Randomly shuffles the elements of this dataset.

17. zip(datasets): Creates a Dataset by zipping together the given datasets.

# NOTE: The following examples use `{ ... }` to represent the
# contents of a dataset.
a = { 1, 2, 3 }
b = { 4, 5, 6 }
c = { (7, 8), (9, 10), (11, 12) }
d = { 13, 14 }

# The nested structure of the `datasets` argument determines the
# structure of elements in the resulting dataset.
Dataset.zip((a, b)) == { (1, 4), (2, 5), (3, 6) }
Dataset.zip((b, a)) == { (4, 1), (5, 2), (6, 3) }

# The `datasets` argument may contain an arbitrary number of
# datasets.
Dataset.zip((a, b, c)) == { (1, 4, (7, 8)),
                            (2, 5, (9, 10)),
                            (3, 6, (11, 12)) }

# The number of elements in the resulting dataset is the same as
# the size of the smallest dataset in `datasets`.
Dataset.zip((a, d)) == { (1, 13), (2, 14) }

tf.data.FixedLengthRecordDataset

A Dataset of fixed-length records from one or more binary files

Inherits From: Dataset

Method和Dataset都一样，就是__init__函数不一样

__init__(
    filenames,
    record_bytes,
    header_bytes=None,
    footer_bytes=None,
    buffer_size=None
)

tf.data.Iterator

Represents the state of iterating through a Dataset

Properties: initializer, output_shapes, output_types

Method:    Creates a new iterator from the given iterator resource

__init__(
    iterator_resource,
    initializer,
    output_types,
    output_shapes
)

Note: Most users will not call this initializer directly, and will instead use Dataset.make_initializable_iterator() orDataset.make_one_shot_iterator().

2. get_next(name=None): Returns a nested structure of tf.Tensor containing the next element

3. make_initializer(dataset,name=None): Returns a tf.Operation that initializes this iterator on dataset.

tf.data.TFRecordDataset

A Dataset comprising records from one or more TFRecord files

Inherits From: Dataset  method和Dataset一样

__init__(
    filenames,
    compression_type=None,
    buffer_size=None
)

 这个方法提取出的是tf.train.Example格式的数据

# Transforms a scalar string `example_proto` into a pair of a scalar string and
# a scalar integer, representing an image and its label, respectively.
def _parse_function(example_proto):
  features = {"image": tf.FixedLenFeature((), tf.string, default_value=""),
              "label": tf.FixedLenFeature((), tf.int32, default_value=0)}
  parsed_features = tf.parse_single_example(example_proto, features)
  return parsed_features["image"], parsed_features["label"]

# Creates a dataset that reads all of the examples from two files, and extracts
# the image and label features.
filenames = ["/var/data/file1.tfrecord", "/var/data/file2.tfrecord"]
dataset = tf.data.TFRecordDataset(filenames)
dataset = dataset.map(_parse_function)

tf.data.TextLineDataset

A Dataset comprising lines from one or more text files

 Inherits From: Dataset method 和Dataset一样

__init__(
    filenames,
    compression_type=None,
    buffer_size=None
)

Datset API导入

tf1.3    tf.contrib.data.Dataset

tf1.4   tf.data.Dataset

Dataset和Iterator

只需要关注两个最重要的基础类： Dataset he Iterator

Dataset可以看作是相同类型“元素”的有序列表，单个“元素”可以使向量，字符串，图片甚至是tuple或者dict

 如下为非eager模式的每个元素为数字的Dataset的例子：

import tensorflow as tf
import numpy as np

dataset = tf.data.Dataset.from_tensor_slices(np.array([1.0, 2.0, 3.0, 4.0, 5.0]))

iterator = dataset.make_one_shot_iterator()
one_element = iterator.get_next()
with tf.Session() as sess:
   for i in range(5):
       print(sess.run(one_element))

输出为1.0到5.0 

iterator = dataset.make_one_shot_iterator()从dataset中实例化了一个Iterator，这个Iterator是一个“one shot iterator”，即只能从头到尾读取一次。

 one_element = iterator.get_next()表示从iterator里取出一个元素

dataset=tf.data.Dataset.from_tensor_slices(np.random.uniform(size=(5,2)))

dataset = tf.data.Dataset.from_tensor_slices(
   {
       "a": np.array([1.0, 2.0, 3.0, 4.0, 5.0]),                                       
       "b": np.random.uniform(size=(5, 2))
   })

这时函数会分别切分"a"中的数值以及"b"中的数值，最终dataset中的一个元素就是类似于{"a": 1.0, "b": [0.9, 0.1]}的形式。

对Dataset中的元素做变换： Transformation

一个Dataset通过Transformation变成一个新的Dataset。通常我们可以通过Transformation完成数据变换，打乱，组成batch，生成epoch等一系列操作。

(1) map

map接收一个函数，Dataset中的每个元素都会被当作这个函数的输入，并将函数返回值作为新的Dataset，如我们可以对dataset中每个元素的值加1：

dataset = tf.data.Dataset.from_tensor_slices(np.array([1.0, 2.0, 3.0, 4.0, 5.0]))
dataset = dataset.map(lambda x: x + 1) # 2.0, 3.0, 4.0, 5.0, 6.0

(2) batch

batch就是将多个元素组合成batch，如下面的程序将dataset中的每个元素组成了大小为32的batch：

dataset=dataset.batch(32)

(3) shuffle

shuffle的功能为打乱dataset中的元素，它有一个参数buffersize，表示打乱时使用的buffer的大小：

dataset=dataset.shuffle(buffer_size=10000)

(4) repeat

  repeat的功能就是将整个序列重复多次，主要用来处理机器学习中的epoch，假设原先的数据是一个epoch，使用repeat(5)就可以将之变成5个epoch：

dataset=dataset.repeat(5)

如果直接调用repeat()的话，生成的序列就会无限重复下去，没有结束，因此也不会抛出tf.errors.OutOfRangeError异常：

dataset=dataset.repeat()

例子：

#函数的功能时将filename对应的图片文件读进来，并缩放到统一的大小
def _parse_function(filename, label):
 image_string = tf.read_file(filename)
 image_decoded = tf.image.decode_image(image_string)
 image_resized = tf.image.resize_images(image_decoded, [28, 28])
 return image_resized, label

# 图片文件的列表
filenames = tf.constant(["/var/data/image1.jpg", "/var/data/image2.jpg", ...])
# label[i]就是图片filenames[i]的label
labels = tf.constant([0, 37, ...])

# 此时dataset中的一个元素是(filename, label)
dataset = tf.data.Dataset.from_tensor_slices((filenames, labels))

# 此时dataset中的一个元素是(image_resized, label)
dataset = dataset.map(_parse_function)

# 此时dataset中的一个元素是(image_resized_batch, label_batch)
dataset = dataset.shuffle(buffersize=1000).batch(32).repeat(10)

注意，先shuffle,再batch，再repeat

Dataset的其他创建方法

• tf.data.TextLineDataset()：这个函数的输入是一个文件的列表，输出是一个dataset。dataset中的每一个元素就对应了文件中的一行。可以使用这个函数来读入CSV文件。

• tf.data.FixedLengthRecordDataset()：这个函数的输入是一个文件的列表和一个record_bytes，之后dataset的每一个元素就是文件中固定字节数record_bytes的内容。通常用来读取以二进制形式保存的文件，如CIFAR10数据集就是这种形式。

• tf.data.TFRecordDataset()：顾名思义，这个函数是用来读TFRecord文件的，dataset中的每一个元素就是一个TFExample。

更多类型的Iterator

在非Eager模式下，最简单的创建Iterator的方法就是通过dataset.make_one_shot_iterator()来创建一个one shot iterator。除了这种one shot iterator外，还有三个更复杂的Iterator，即：

• initializable iterator

• reinitializable iterator

• feedable iterator

initializable iterator必须要在使用前通过sess.run()来初始化。使用initializable iterator，可以将placeholder代入Iterator中，这可以方便我们通过参数快速定义新的Iterator。一个简单的initializable iterator使用示例：

limit = tf.placeholder(dtype=tf.int32, shape=[])

dataset = tf.data.Dataset.from_tensor_slices(tf.range(start=0, limit=limit))

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

with tf.Session() as sess:
   sess.run(iterator.initializer, feed_dict={limit: 10})
   for i in range(10):
     value = sess.run(next_element)
     assert i == value

此时的limit相当于一个“参数”，它规定了Dataset中数的“上限”。

initializable iterator还有一个功能：读入较大的数组。

在使用tf.data.Dataset.from_tensor_slices(array)时，实际上发生的事情是将array作为一个tf.constants保存到了计算图中。当array很大时，会导致计算图变得很大，给传输、保存带来不便。这时，我们可以用一个placeholder取代这里的array，并使用initializable iterator，只在需要时将array传进去，这样就可以避免把大数组保存在图里，示例代码为（来自官方例程）：

# 从硬盘中读入两个Numpy数组
with np.load("/var/data/training_data.npy") as data:
 features = data["features"]
 labels = data["labels"]

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))
iterator = dataset.make_initializable_iterator()
sess.run(iterator.initializer, feed_dict={features_placeholder: features,
                                         labels_placeholder: labels})

在非Eager模式下，Dataset中读出的一个元素一般对应一个batch的Tensor，我们可以使用这个Tensor在计算图中构建模型。

使用例子：

filenames = ["/var/data/file1.tfrecord", "/var/data/file2.tfrecord"]
dataset = tf.data.TFRecordDataset(filenames)
dataset = dataset.map(...)
   dataset = datset.shuffle(buffer_size=10000)
dataset = dataset.batch(32)
   dataset = datset.repeat()
iterator = dataset.make_initializable_iterator()
next_element = iterator.get_next()

# Compute for 100 epochs.
for _ in range(100):
  sess.run(iterator.initializer)
  while True:
    try:
      sess.run(next_element)
    except tf.errors.OutOfRangeError:
      break

  # [Perform end-of-epoch calculations here.]

Using high-level APIs

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)

To use a Dataset in the input_fn of a tf.estimator.Estimator, we also recommend using Dataset.make_one_shot_iterator(). For example:

def dataset_input_fn():
  filenames = ["/var/data/file1.tfrecord", "/var/data/file2.tfrecord"]
  dataset = tf.data.TFRecordDataset(filenames)

  # Use `tf.parse_single_example()` to extract data from a `tf.Example`
  # protocol buffer, and perform any additional per-record preprocessing.
  def parser(record):
    keys_to_features = {
        "image_data": tf.FixedLenFeature((), tf.string, default_value=""),
        "date_time": tf.FixedLenFeature((), tf.int64, default_value=""),
        "label": tf.FixedLenFeature((), tf.int64,
                                    default_value=tf.zeros([], dtype=tf.int64)),
    }
    parsed = tf.parse_single_example(record, keys_to_features)

    # Perform additional preprocessing on the parsed data.
    image = tf.decode_jpeg(parsed["image_data"])
    image = tf.reshape(image, [299, 299, 1])
    label = tf.cast(parsed["label"], tf.int32)

    return {"image_data": image, "date_time": parsed["date_time"]}, label

  # Use `Dataset.map()` to build a pair of a feature dictionary and a label
  # tensor for each example.
  dataset = dataset.map(parser)
  dataset = dataset.shuffle(buffer_size=10000)
  dataset = dataset.batch(32)
  dataset = dataset.repeat(num_epochs)
  iterator = dataset.make_one_shot_iterator()

  # `features` is a dictionary in which each value is a batch of values for
  # that feature; `labels` is a batch of labels.
  features, labels = iterator.get_next()
  return features, labels

注意顺序呀：

map  ->   shuffle  ->   batch  ->  repeat

如果不shuffle  就 map  -> repeat -> batch