Tensorflow Dataset使用实战

本文主要来自TensorFlow Dataset API tutorial – build high performance data pipelines.

Tensorflow Dataset框架主要包括两部分：

• Dataset
• Iterator

Dataset就是读入数据后的数据集对象，数据可以通过多种方法获得：tensors，Numpy arrays，numpy文件，TFRecord文件还有文本文件。一旦将这些数据读入了Dataset对象中，我们就可以使用多种不同的方法来使用这些数据，如：

• batch()，批量消耗Dataset数据集中的数据；
• map()，使用一些函数来对数据进行转换（如常见的预处理和解析tfrecord）；
• zip()，将多个不同的Dataset对象合并成为一个Dataset对象；
• filter()，使用某些函数来过滤掉一些坏数据；
• repeat()，设置数据的总数，如1就代表训练一个epoch数据就已经被消耗完了，继续读取数据的话会抛出tf.errors.OutOfRangeError；
• shuffle()，打乱数据。
  除了上面这些外还有一些其他的API，参见这里

另一个主要部分Iterator其实就是一个迭代器，每次迭代都会生成一定数据用于训练/验证/测试。接下来我们将分别对这两部分进行更为详细的介绍。

一些简单Dataset 例子

在这部分我们将通过一些简单的例子来看看如何创建Dataset。首先来看看如何通过Numpy数组创建Dataset。在这里我们用到了from _tensor_slices()方法。

import tensorflow as tf 
import os 
import numpy as np 

x = np.arange(0, 10)
# create dataset object from numpy array
dx = tf.data.Dataset.from_tensor_slices(x)

现在dx是一个Dataset对象，下一步就是创建一个Iterator来从Dataset中不断地取出数据。在下面，我们使用make_one_shot_iterator()来创建Iterator，这种方式创建的Iterator只能初始化使用一次，无法再次使用。在后面我们会介绍可重复初始化使用的方法。

# create a one_shot iterator
iterator = dx.make_one_shot_iterator()
# extract an element
next_element = iterator.get_next()

with tf.Session() as sess:
    for i in range(11):
        val = sess.run(next_element)
        print(val)

# 输出：
0
1
2
3
4
5
6
7
8
9
---------------------------------------------------------------------------
OutOfRangeError  .... ...

来看看我们刚刚做了什么。首先，我们是用make_one_shot_iterator()创建了一个Iterator，接着我们使用get_next()来从里面取出数据。在最后，我们创建了会话和一个for循环，每次迭代都从Dataset中取出一个数据。这个用法与生成器基本是一致的。在Dataset中的数据消耗完之后，系统抛出了OutOfRangeError。

如果想要重复的使用Dataset中的数据，我们可以使用make_initializable_iterator()方法：

iterator = dx.make_initializable_iterator()
next_element = iterator.get_next()
with tf.Session() as sess:
    sess.run(iterator.initializer)
    for i in range(15):
        val = sess.run(next_element)
        print(val)
        if i%9==0 and i>0:
            sess.run(iterator.initializer)

# 输出：
0
1
2
3
4
5
6
7
8
9
0
1
2
3
4

值得注意的是，使用make_initializable_iterator()来创建Iterator时，我们每次重复使用时都需要进行初始化，也即上面的sess.run(iterator.initializer)，而且这个操作需要在运行next_element之前。最后的两行表示，一旦我们从Dataset中取出了10个数据（刚刚好一个完整的dx）则重新初始化Iterator，重新给Iterator 填充数据，这样系统就不会报OutOfRangeError了。

此外，还有一些其他的方法来使用Dataset数据。batch()使得我们能一次从Dataset中取出多个数据。

dx = tf.data.Dataset.from_tensor_slices(x).batch(3)
iterator = dx.make_initializable_iterator()
next_element = iterator.get_next()

with tf.Session() as sess:
    sess.run(iterator.initializer)
    for i in range(15):
        val = sess.run(next_element)
        print(val)
        if (i+1)%(10//3)==0 and i>0:
            sess.run(iterator.initializer)

#输出：
[0 1 2]
[3 4 5]
[6 7 8]
[0 1 2]
[3 4 5]
[6 7 8]
[0 1 2]
[3 4 5]
[6 7 8]
[0 1 2]
[3 4 5]
[6 7 8]
[0 1 2]
[3 4 5]
[6 7 8]

接下来，我们使用zip来将不同的数据合并起来。当我们在使用成对的training/validation数据（如images 和labels）时这个操作会变得异常方便:

def simple_zip_example():
    x = np.arange(0, 10)
    y = np.arange(1, 11)
    # create dataset objects from arrays
    dx = tf.data.Dataset.from_tensor_slices(x)
    dy = tf.data.Dataset.from_tensor_slices(y)
    # zip the two datasets together
    dcomb = tf.data.Dataset.zip((dx, dy)).batch(3)
    iterator = dcomb.make_initializable_iterator()
    # extract an element
    next_element = iterator.get_next()
    with tf.Session() as sess:
        sess.run(iterator.initializer)
        for i in range(15):
            val = sess.run(next_element)
            print(val)
            if (i+1)%(10//3) == 0 and i>0:
                sess.run(iterator.initializer)
                
simple_zip_example() 

# 输出：
(array([0, 1, 2]), array([1, 2, 3]))
(array([3, 4, 5]), array([4, 5, 6]))
(array([6, 7, 8]), array([7, 8, 9]))
(array([0, 1, 2]), array([1, 2, 3]))
(array([3, 4, 5]), array([4, 5, 6]))
(array([6, 7, 8]), array([7, 8, 9]))
(array([0, 1, 2]), array([1, 2, 3]))
(array([3, 4, 5]), array([4, 5, 6]))
(array([6, 7, 8]), array([7, 8, 9]))
(array([0, 1, 2]), array([1, 2, 3]))
(array([3, 4, 5]), array([4, 5, 6]))
(array([6, 7, 8]), array([7, 8, 9]))
(array([0, 1, 2]), array([1, 2, 3]))
(array([3, 4, 5]), array([4, 5, 6]))
(array([6, 7, 8]), array([7, 8, 9]))

来看看我们干了什么。我们定义了一个函数，这个函数将两个arrays，zip成一对对的数据，最后我们将这些数据成对的取出来。

除了像上面使用重复多次使用sess.run(iterator.initilizer)来初始化重复使用数据外，还可以通过repeat()来达到相同的目的。

dcomb = tf.data.Dataset.zip((dx, dy)).repeat().batch(3)

需要注意的是，repeat()方法是可以带参数的，当他不带参数时表示Dataset数据集无限重复，我们也可以指定数据集重复次数, 如repeat(10)表示重复10次，也就是10个epochs。这时当10次重复的数据集都被消耗完时，也会抛出OutOfRangeError错误。接下来我们将会通过Mnist手写数据集来介绍tf.data.Dataset实战。

Mnist手写数据实战

在这一部分，我们将会创建一个Mnist手写数字分类器，这些MNIST图片来自于sklearn包，我们将会将他们装载进Dataset中，然后从Dataset中读取数据。

第一步，从sklearn中载入数据并分成训练集和验证集，其中训练集占80%，验证集占20%：

from sklearn.datasets import load_digits

# load the data
digits = load_digits(return_X_y=True)

# split into train and validation sets
train_images = digits[0][:int(len(digits[0])*0.8)]
train_labels = digits[1][:int(len(digits[0])*0.8)]
valid_images = digits[0][int(len(digits[0])*0.8):]
valid_images = digits[1][int(len(digits[0])*0.8):]

接着，我们将会使用Dataset来创建训练集和验证集：

# create the training datasets
dx_train = tf.data.Dataset.from_tensor_slices(train_images)
# apply a one-hot transformation to each label for use in the neural network
dy_train = tf.data.Dataset.from_tensor_slices(
    train_labels).map(lambda z: tf.one_hot(z, 10))
# zip the x and y training data together and shuffle, batch etc.
train_dataset = tf.data.Dataset.zip(
    (dx_train, dy_train)).shuffle(500).repeat().batch(30)

# create the validation set in the same way
dx_valid = tf.data.Dataset.from_tensor_slices(valid_images)
dy_valid = tf.data.Dataset.from_tensor_slices(
    valid_labels).map(lambda z: tf.one_hot(z, 10))
valid_dataset = tf.data.Dataset.zip(
    (dx_valid, dy_valid)).shuffle(500).repeat().batch(30)

在上面我们通过from_tensor_slices()读入数据，但是在label数据上我们进行了一些操作——将label类型转换为了one-hot类型。最后，将对应的x和y zip进一个Dataset中，打乱，重复并以batch_size为30的大小读取数据。

现在，训练和验证的Dataset数据集都已经准备好了，我们准备从训练集和验证集中读取数据。在这里我们使用另一种创建Iterator的方式——from_structure(),他将会创建一个迭代器，这个迭代器没有绑定数据集，他需要的参数是输出数据的类型和输出数据的shape。对不同的数据集可以使用iterator.make_initializer(dataset)来指定初始化特定数据集。

# create general iterator 
iterator = tf.data.Iterator.from_structure(
    train_dataset.output_types, train_dataset.output_shapes)
next_element = iterator.get_next()

# make datasets that we can initialize seperately, 
# but using the same structure via the common iterator  
training_init_op = iterator.make_initializer(train_dataset)
validation_init_op = iterator.make_initializer(valid_dataset)

通过from_structure()来构建的Iterator，可以非常方便的进行不同数据集之间的切换。

接下来，我们将会创建一个简单的全连接层神经网络模型，并且在其中加入了batch normlization和dropout 技术。

def nn_model(in_data):
    bn = tf.layers.batch_normalization(in_data)
    fc1 = tf.layers.dense(bn, 50)
    fc2 = tf.layers.dense(fc1, 50)
    fc2 = tf.layers.dropout(fc2)
    fc3 = tf.layers.dense(fc2, 10)
    return fc3

接着我们便可以定义我们的loss函数和优化器了：

logits = nn_model(next_element[0])

# add the optimizer and loss
loss = tf.reduce_sum(tf.nn.softmax_cross_entropy_with_logits_v2(
    labels=next_element[1], logits=logits))
optimizer = tf.train.AdamOptimizer().minimize(loss)

# get accuracy
prediction = tf.argmax(logits, 1)
equality = tf.equal(prediction, tf.argmax(next_element[1], 1))
accuracy = tf.reduce_mean(tf.cast(equality, tf.float32))
init_op = tf.global_variables_initializer()

最后，我们便可以开始创建会话进行训练了：

# run the training
epochs = 600

with tf.Session() as sess:
    sess.run(init_op)
    sess.run(training_init_op)
    for i in range(epochs):
        loss_, _, acc = sess.run([loss, optimizer, accuracy])
        if i % 50 == 0:
            print('Epoch: {}, loss:{:.3f}, training accuracy: {:.2f}%'.format(
                i, loss_, acc * 100))

    # now setup the validation run
    valid_iters = 100
    # re-initilize the iterator, but this time with validation data
    sess.run(validation_init_op)
    avg_acc = 0
    for i in range(valid_iters):
        acc = sess.run([accuracy])
        avg_acc += acc[0]
    print(
        'Average validation set accuracy over {} iterations is {:.2f}%'.format(
            valid_iters, (avg_acc / valid_iters) * 100))

# 输出：
Epoch: 0, loss:569.320, training accuracy: 10.00%
Epoch: 50, loss:20.650, training accuracy: 76.67%
Epoch: 100, loss:8.922, training accuracy: 86.67%
Epoch: 150, loss:3.067, training accuracy: 96.67%
Epoch: 200, loss:4.746, training accuracy: 90.00%
Epoch: 250, loss:1.313, training accuracy: 100.00%
Epoch: 300, loss:4.666, training accuracy: 93.33%
Epoch: 350, loss:2.234, training accuracy: 96.67%
Epoch: 400, loss:0.791, training accuracy: 100.00%
Epoch: 450, loss:10.615, training accuracy: 90.00%
Epoch: 500, loss:1.893, training accuracy: 96.67%
Epoch: 550, loss:0.878, training accuracy: 100.00%
Average validation set accuracy over 100 iterations is 90.47%

如上所示，在训练和验证阶段，分别初始化了不同的数据集。最后验证阶段我们并非只使用了一次验证集，相反的我们进行了多次验证最后取了一个平均值，最后的准确率为90.47%，还挺差的。