02.21 如何使用TensorFlow中的Dataset API

翻譯 | AI科技大本營

參與 | zzq

審校 | reason_W

本文已更新至TensorFlow1.5版本

我們知道，在TensorFlow中可以使用feed-dict的方式輸入數據信息，但是這種方法的速度是最慢的，在實際應用中應該儘量避免這種方法。而使用輸入管道就可以保證GPU在工作時無需等待新的數據輸入，這才是正確的方法。

幸運的是，TensorFlow提供了一種內置的API——Dataset，使得我們可以很容易地就利用輸入管道的方式輸入數據。在這篇教程中，我們將介紹如何創建和使用輸入管道以及如何高效地向模型輸入數據。

這篇文章將解釋DatasetAPI的基本工作機制，並給出了幾種最常用的例子。

你可以通過下面的網站地址下載文章中的代碼：

https://github.com/FrancescoSaverioZuppichini/Tensorflow-Dataset-Tutorial/blob/master/dataset_tutorial.ipynb

▌概述

使用Dataset的三個步驟：

1. 載入數據：為數據創建一個Dataset實例

2. 創建一個迭代器：使用創建的數據集來構造一個Iterator實例以遍歷數據集

3. 使用數據：使用創建的迭代器，我們可以從數據集中獲取數據元素，從而輸入到模型中去。

▌載入數據

首先，我們需要將一些數據放到數據集中。

從numpy載入

這是最常見的情況，假設我們有一個numpy數組，我們想將它傳遞給TensorFlow

# create a random vector of shape (100,2)
x = np.random.sample((100,2))
# make a dataset from a numpy array
dataset = tf.data.Dataset.from_tensor_slices(x)

我們也可以傳遞多個numpy數組，最典型的例子是當數據被劃分為特徵和標籤的時候：

features, labels = (np.random.sample((100,2)), np.random.sample((100,1)))
dataset = tf.data.Dataset.from_tensor_slices((features,labels))

從tensors中載入

我們當然也可以用一些張量初始化數據集

# using a tensor
dataset = tf.data.Dataset.from_tensor_slices(tf.random_uniform([100, 2]))

從placeholder中載入

如果我們想動態地改變Dataset中的數據，使用這種方式是很有用的。

x = tf.placeholder(tf.float32, shape=[None,2])
dataset = tf.data.Dataset.from_tensor_slices(x)

從generator載入

我們也可以從generator中初始化一個Dataset。當一個數組中元素長度不相同時，使用這種方式處理是很有效的。（例如一個序列）

sequence = np.array([[1],[2,3],[3,4]])
def generator():
 for el in sequence:
 yield el
dataset = tf.data.Dataset().from_generator(generator,
output_types=tf.float32,
 output_shapes=[tf.float32])

在這種情況下，你還需要指定數據的類型和大小以創建正確的tensor

▌創建一個迭代器

我們已經知道了如何創建數據集，但是如何從中獲取數據呢？我們需要使用一個Iterator遍歷數據集並重新得到數據真實值。有四種形式的迭代器。

One shot Iterator

這是最簡單的迭代器，下面給出第一個例子：

x = np.random.sample((100,2))
# make a dataset from a numpy array
dataset = tf.data.Dataset.from_tensor_slices(x)
# create the iterator
iter = dataset.make_one_shot_iterator()

接著你需要調用get_next()來獲得包含數據的張量

...
# create the iterator
iter = dataset.make_one_shot_iterator()
el = iter.get_next()

我們可以運行 el 來查看它們的值。

with tf.Session() as sess:
 print(sess.run(el)) # output: [ 0.42116176 0.40666069]

可初始化的迭代器

如果我們想建立一個可以在運行時改變數據源的動態數據集，我們可以用placeholder 創建一個數據集。接著用常見的feed-dict機制初始化這個placeholder。這些工作可以通過使用一個可初始化的迭代器完成。使用上一節的第三個例子

# using a placeholder
x = tf.placeholder(tf.float32, shape=[None,2])
dataset = tf.data.Dataset.from_tensor_slices(x)
data = np.random.sample((100,2))
iter = dataset.make_initializable_iterator() # create the iterator
el = iter.get_next()
with tf.Session() as sess:
 # feed the placeholder with data
 sess.run(iter.initializer, feed_dict={ x: data })
 print(sess.run(el)) # output [ 0.52374458 0.71968478]

這次，我們調用make_initializable_iterator。接著我們在 sess 中運行 initializer 操作，以傳遞數據，這種情況下數據是隨機的 numpy 數組。

假設我們有了訓練集和測試集，如下代碼所示

train_data = (np.random.sample((100,2)), np.random.sample((100,1)))
test_data = (np.array([[1,2]]), np.array([[0]]))

接著，我們訓練該模型，並在測試數據集上對其進行測試，這可以通過訓練後對迭代器再次進行初始化來完成。

# 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]))

可重新初始化的迭代器

這個概念和之前的相似，我們想在數據間動態切換。但是我們是轉換數據集而不是把新數據送到相同的數據集。和之前一樣，我們需要一個訓練集和一個測試集

# 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)))

接下來創建兩個Dataset

# 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)

現在我們要用到一個小技巧，即創建一個通用的Iterator

# create a iterator of the correct shape and type
iter = tf.data.Iterator.from_structure(train_dataset.output_types,
 train_dataset.output_shapes)

接著創建兩個初始化運算

# create the initialisation operations
train_init_op = iter.make_initializer(train_dataset)
test_init_op = iter.make_initializer(test_dataset)

和之前一樣，我們得到下一個元素

features, labels = iter.get_next()

現在，我們可以直接使用session運行兩個初始化運算。把上面這些綜合起來我們可以得到：

# 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迭代器

老實說，我並不認為這種迭代器有用。這種方式是在迭代器之間轉換而不是在數據集間轉換，比如在來自make_one_shot_iterator()的一個迭代器和來自make_initializable_iterator()的一個迭代器之間進行轉換。

▌使用數據

在之前的例子中，我們使用session來打印Dataset中next元素的值

...
next_el = iter.get_next()
...
print(sess.run(next_el)) # will output the current element

現在為了向模型傳遞數據，我們只需要傳遞get_next()產生的張量。

在下面的代碼中，我們有一個包含兩個numpy數組的Dataset，這裡用到了和第一節一樣的例子。注意到我們需要將.random.sample封裝到另外一個numpy數組中，因此會增加一個維度以用於數據batch。

# 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) # pass the first value from iter.get_next() as input
net = tf.layers.dense(net, 8)
prediction = tf.layers.dense(net, 1)
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)

我們直接使用來自iter.get_next()的張量作為神經網絡第一層的輸入和損失函數的標籤。將上面的綜合起來可以得到：

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))

輸出：

Iter: 0, Loss: 0.1328
Iter: 1, Loss: 0.1312
Iter: 2, Loss: 0.1296
Iter: 3, Loss: 0.1281
Iter: 4, Loss: 0.1267
Iter: 5, Loss: 0.1254
Iter: 6, Loss: 0.1242
Iter: 7, Loss: 0.1231
Iter: 8, Loss: 0.1220
Iter: 9, Loss: 0.1210

▌

batch

通常情況下，batch是一件麻煩的事情，但是通過Dataset API我們可以使用batch(BATCH_SIZE)方法自動地將數據按照指定的大小batch，默認值是1。在接下來的例子中，我們使用的batch大小為4。

# BATCHING
BATCH_SIZE = 4
x = np.random.sample((100,2))
# make a dataset from a numpy array
dataset = tf.data.Dataset.from_tensor_slices(x).batch(BATCH_SIZE)
iter = dataset.make_one_shot_iterator()
el = iter.get_next()
with tf.Session() as sess:
 print(sess.run(el))

輸出：

[[ 0.65686128 0.99373963]
[ 0.69690451 0.32446826]
[ 0.57148422 0.68688242]
[ 0.20335116 0.82473219]]

Repeat

使用.repeat()我們可以指定數據集迭代的次數。如果沒有設置參數，則迭代會一直循環。通常來說，一直循環並直接用標準循環控制epoch的次數能取得較好的效果。

Shuffle

我們可以使用shuffle()方法將Dataset隨機洗牌，默認是在數據集中對每一個epoch洗牌，這種處理可以避免過擬合。

我們也可以設置buffer_size參數，下一個元素將從這個固定大小的緩存中按照均勻分佈抽取。例子：

# BATCHING
BATCH_SIZE = 4
x = np.array([[1],[2],[3],[4]])
# make a dataset from a numpy array
dataset = tf.data.Dataset.from_tensor_slices(x)
dataset = dataset.shuffle(buffer_size=100)
dataset = dataset.batch(BATCH_SIZE)
iter = dataset.make_one_shot_iterator()
el = iter.get_next()
with tf.Session() as sess:
 print(sess.run(el))

首次運行輸出：

[[4]
[2]
[3]
[1]]

第二次運行輸出：

[[3]
[1]
[2]
[4]]

這樣數據就被洗牌了。你還可以設置seed參數

▌

你可以使用map()方法對數據集的每個成員應用自定義的函數。在下面的例子中，我們將每個元素乘以2。

# MAP
x = np.array([[1],[2],[3],[4]])
# make a dataset from a numpy array
dataset = tf.data.Dataset.from_tensor_slices(x)
dataset = dataset.map(lambda x: x*2)
iter = dataset.make_one_shot_iterator()
el = iter.get_next()
with tf.Session() as sess:
# this will run forever
 for _ in range(len(x)):
 print(sess.run(el))

輸出：

[2]
[4]
[6]
[8]

其他資源

TensorFlow dataset tutorial: https://www.tensorflow.org/programmers_guide/datasets

Dataset docs:https://www.tensorflow.org/api_docs/python/tf/data/Dataset

▌結論

Dataset API提供了一種快速而且魯棒的方法來創建優化的輸入管道來訓練、評估和測試我們的模型。在這篇文章中，我們瞭解了很多常見的利用Dataset API的操作。

原文：https://towardsdatascience.com/how-to-use-dataset-in-tensorflow-c758ef9e4428