Press "Enter" to skip to content

Tensorflow数据读取指南

本站内容均来自兴趣收集,如不慎侵害的您的相关权益,请留言告知,我们将尽快删除.谢谢.

tensorflow的灵活性带来的学习成本是很多人头疼的问题,在tf中,读取数据基本有四种方法:
1. tf.data (官方推荐):方便地构建复杂的输入管道
2. Feeding:通过input_fn来yield数据
3. QueueRunner:基于队列的输入管道
4. 预加载数据。用constant或variable在内存中存储所有的数据
其实从tf的历史发展轨迹可以看出,tf.data是在feedintg技术和QueueRunner的基础上发展而来。
Feeding技术

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

简单教程中的数据加载,都是一行伪代码:(x, y) = load_data() 就搞定了
但现实中,令训练数据的使用变得复杂的主要原因有很多:
一、数据集不一致,不同的数据源的数据格式不一样,为了统一,我们需要一个转接层,我们称之为统一化过程
二、数据集过大,不适合一次性导入,需要分批次导入,边训练边导入
三、需要数据增强,在线生成,甚至完全在线生成,这样就需要将训练时的feed和读取的统一化数据之间加入一个动态生成过程
四、当然批次化(batch),train和val的划分也都是重复性的工作
tf.data就是用来解决以上问题的。
两个重要类介绍:tf.data.Dataset和tf.data.Iterator,
– Dataset用来表示一组数据,是一个序列,序列的每个元素包含一个或多个Tensor对象。
– Iterator用户从数据集中抽取元素,Iterator.get_next()可以返回数据集的下一个元素,可以看作是输入管道与模型之间的接口。
Dataset
面对输入的数据集,我们要建立一个清晰的数据流概念,一个原始的数据集开始,后续的每一次转换都是流水线式的操作。因而我们可以看到这样的代码形式:data.TextLineDataset(filenames).map(decode_func).shuffle(buffer_size=10000).batch(batch_size).make_initializable_iterator()。
Dataset的每一个元素一定是相同的结构,每个元素包含一个或多个Tensor对象,每个Tensor都有一个tf.DType来表示tensor中的元素数据类型,还有伊特tf.TensorShape来表示tensor中每个元素的形状。
Dataset的map、flat_map、filter等等方法都是常用的操作链上过滤器方法。过滤器模式是常见的设计模式。
Iterator
如果说Dataset是从源中一步一步创建出一个统一化的数据集的话,Iterator就是负责怎么访问这个统一后的数据集然后喂给你的模型。有四种迭代器
– one-shot
– initializable
– reinitializable
– feedable
one-shot最简单,就是一个元素一个元素的吐,但不能初始化,即参数化,就是遍历一遍数据集就什么也干不了了。
initializable迭代器就可以初始化,可以接受一个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()
 # Initialize an iterator over a dataset with 10 elements.
 sess.run(iterator.initializer, feed_dict={max_value: 10})

initializable更典型的应用方式:

# initializable iterator to switch between dataset
 EPOCHS = 10
 x, y = tf.placeholder(tf.float32, shape=[None,2]), tf.placeholder(tf.float32, shape=[None,1])
 dataset = tf.data.Dataset.from_tensor_slices((x, y))
 train_data = (np.random.sample((100,2)), np.random.sample((100,1)))
 test_data = (np.array([[1,2]]), np.array([[0]]))
 iter = dataset.make_initializable_iterator()
 features, labels = iter.get_next()
 with tf.Session() as sess:
 #     initialise iterator with train data
     sess.run(iter.initializer, feed_dict={ x: train_data[0], y: train_data[1]})
     for _ in range(EPOCHS):
         sess.run([features, labels])
 #     switch to test data
     sess.run(iter.initializer, feed_dict={ x: test_data[0], y: test_data[1]})
     print(sess.run([features, labels]))

Dataset的创建依赖一个placeholder,所以只是表明了数据集的结构信息,具体的数据是什么还得在运行期的时候赋值。
但如果我们不想在运行期使用feed_dict怎么办呢,用reinitializable迭代器支持多个数据集对象的初始化,创建的迭代器不再是dataset.make_xxx_iterator这种形式,而是使用结构直接使用Iterator的静态方法初始化tf.data.Iterator.from_structure(train_dataset.output_types, train_dataset.output_shapes)

# Reinitializable iterator to switch between Datasets
 EPOCHS = 10
 # making fake data using numpy
 train_data = (np.random.sample((100,2)), np.random.sample((100,1)))
 test_data = (np.random.sample((10,2)), np.random.sample((10,1)))
 # create two datasets, one for training and one for test
 train_dataset = tf.data.Dataset.from_tensor_slices(train_data)
 test_dataset = tf.data.Dataset.from_tensor_slices(test_data)
 # create a iterator of the correct shape and type
 iter = tf.data.Iterator.from_structure(train_dataset.output_types,
 train_dataset.output_shapes)
 features, labels = iter.get_next()
 # create the initialisation operations
 train_init_op = iter.make_initializer(train_dataset)
 test_init_op = iter.make_initializer(test_dataset)
 with tf.Session() as sess:
     sess.run(train_init_op) # switch to train dataset
     for _ in range(EPOCHS):
         sess.run([features, labels])
     sess.run(test_init_op) # switch to val dataset
     print(sess.run([features, labels]))

feedable结合了initializable和reinitializable两者,可以使用placeholder和多个数据集一起初始化。如果说reinitializable方便在数据集间切换,那么feedable就是方便在迭代器间切换。

# Define training and validation datasets with the same structure.
 training_dataset = tf.data.Dataset.range(100).map(
     lambda x: x + tf.random_uniform([], -10, 10, tf.int64)).repeat()
 validation_dataset = tf.data.Dataset.range(50)
 #
 # A feedable iterator is defined by a handle placeholder and its structure. We
 # could use the `output_types` and `output_shapes` properties of either
 # `training_dataset` or `validation_dataset` here, because they have
 # identical structure.
 handle = tf.placeholder(tf.string, shape=[])
 iterator = tf.data.Iterator.from_string_handle(
     handle, training_dataset.output_types, training_dataset.output_shapes)
 #
 next_element = iterator.get_next()
 #
 # You can use feedable iterators with a variety of different kinds of iterator
 # (such as one-shot and initializable iterators).
 training_iterator = training_dataset.make_one_shot_iterator()
 validation_iterator = validation_dataset.make_initializable_iterator()
 #
 # The `Iterator.string_handle()` method returns a tensor that can be evaluated
 # and used to feed the `handle` placeholder.
 training_handle = sess.run(training_iterator.string_handle())
 validation_handle = sess.run(validation_iterator.string_handle())
 #
 # Loop forever, alternating between training and validation.
 while True:
   # Run 200 steps using the training dataset. Note that the training dataset is
   # infinite, and we resume from where we left off in the previous `while` loop
   # iteration.
   for _ in range(200):
     sess.run(next_element, feed_dict={handle: training_handle})
   #
   # Run one pass over the validation dataset.
   sess.run(validation_iterator.initializer)
   for _ in range(50):
     sess.run(next_element, feed_dict={handle: validation_handle})

但怎么说呢,tensorflow的设计师一如既往地发挥了他们接口设计纷繁复杂的能力,真的是极其混乱难用,一点也不清晰,要不然tensorflow里也不会引入keras包了。

数据的消费

EPOCHS = 10
 BATCH_SIZE = 16
 # using two numpy arrays
 features, labels = (np.array([np.random.sample((100,2))]),
                     np.array([np.random.sample((100,1))]))
 dataset = tf.data.Dataset.from_tensor_slices((features,labels)).repeat().batch(BATCH_SIZE)
 iter = dataset.make_one_shot_iterator()
 x, y = iter.get_next()
 # make a simple model
 net = tf.layers.dense(x, 8, activation=tf.tanh) # pass the first value from iter.get_next() as input
 net = tf.layers.dense(net, 8, activation=tf.tanh)
 prediction = tf.layers.dense(net, 1, activation=tf.tanh)
 loss = tf.losses.mean_squared_error(prediction, y) # pass the second value from iter.get_net() as label
 train_op = tf.train.AdamOptimizer().minimize(loss)
 with tf.Session() as sess:
     sess.run(tf.global_variables_initializer())
     for i in range(EPOCHS):
         _, loss_value = sess.run([train_op, loss])
         print("Iter: {}, Loss: {:.4f}".format(i, loss_value))

其中x, y = iter.get_next()表示x,y都是一个获取数据的操作,每次sess.run就是执行一次x(), y()这样就获得一个样本数据。
kera使用数据集就特别简单

dataset = tf.data.Dataset.from_tensor_slices((data, labels))
 dataset = dataset.batch(32).repeat()
 #
 val_dataset = tf.data.Dataset.from_tensor_slices((val_data, val_labels))
 val_dataset = val_dataset.batch(32).repeat()
 #
 model.fit(dataset, epochs=10, steps_per_epoch=30,
           validation_data=val_dataset,
           validation_steps=3)

过滤器函数
从原始数据变成统一的数据格式,最重要的就是map

def input_parser(img_path, label):
     # convert the label to one-hot encoding
     one_hot = tf.one_hot(label, NUM_CLASSES)
     #
     # read the img from file
     img_file = tf.read_file(img_path)
     img_decoded = tf.image.decode_image(img_file, channels=3)
     #
     return img_decoded, one_hot
 tr_data = tr_data.map(input_parser)

input_parser接受了数据集每个元素,将转换完的结果返回,数据集就脱胎换骨了。
图像的转换处理
对于图像的转换处理,我们常常用到 tf.image下的decode_bmp decode_jpeg resize_images transpose_image等等函数,具体的参考[2]。
一个image文件,变成一个Tensor一般需要这样几步:
1. 获取文件名列表
2. [Optional]shuffle文件名列表
3. [Optional]根据epoch扔掉一些文件
4. 建立文件名的queue
5. 使用对应文件的Reader,不同的文件格式用不同的reader,比如TFRecordReader解析TFRecord格式,普通文件的话,就用FileReader,但读取普通文件也比较复杂,比如读单个文件可以使用tf.gfile.FastGfile(path, “r”).read(),对于filename_queue,则使用WholeFileReader(),如下面的语句

image_reader = tf.WholeFileReader()
 data_queue = tf.train.string_input_producer([image_dir], shuffle=False)
 image_key, image_value = image_reader.read(data_queue)
 img = tf.image.decode_jpeg(image_value, channels=3)

实际上还有更方便的read_file方法:

image_value = tf.read_file(image_dir)
 img = tf.image.decode_jpeg(image_value, channels=3)

以csv的读取为例,看queue的使用

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)

当然,google建议将任何形式的原始数据都转换成TFRecords文件。关于TFRecords文件如何制作,请参考[9]。

dataset = tf.data.TFRecordDataset(filename)
 dataset = dataset.repeat(num_epochs)
 #
 # map takes a python function and applies it to every sample
 dataset = dataset.map(decode)

6. 使用Decoder从Reader中反解成Tensor
7. [Optional] preprocessing预处理

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

参考资料
[1] tf.data指引 https://www..org/guide/datasets
[2] tensorflow Images处理指引 https://www.tensorflow.org/api_guides/python/image
[3] 使用coordinator加载图像 https://gist.github.com/eerwitt/518b0c9564e500b4b50f
[4] 使用Tensorlofw读取大数据踩坑之路 https://zhuanlan.zhihu.com/p/28450111
[5] Example of TensorFlows new Input Pipeline https://kratzert.github.io/2017/06/15/example-of-tensorflows-new-input-pipeline.html
[6]load_jpeg_with_tensorflow https://gist.github.com/eerwitt/518b0c9564e500b4b50f
[7] tensorflow Inputs and Readers指引 https://www.tensorflow.org/api_guides/python/io_ops#Readers
[8]tensorflow example mnist 数据集的读取 https://github.com/tensorflow/tensorflow/blob/r1.10/tensorflow/examples/tutorials/mnist/fully_connected_feed.py
[9] http://www.machinelearninguru.com/deep_learning/tensorflow/basics/tfrecord/tfrecord.html

Be First to Comment

发表评论

电子邮件地址不会被公开。 必填项已用*标注