如何在TensorFlow中高效使用數據集

相關代碼地址：https://github.com/FrancescoSaverioZuppichini/Tensorflow-Dataset-Tutorial/blob/master/dataset_tutorial.ipynb

經常使用神經網絡框架的人都會知道，feed-dict 是向 TensorFlow 傳遞信息最慢的方式，應該儘量避免使用。向模型提供數據的正確方式是使用輸入管道，這樣才能保證 GPU 在工作時永遠無需等待新的數據。

幸運的是，TensorFlow 擁有一個名爲 Dataset 的內建 API，它可以讓我們的工作更加簡單。在本教程中，我們將介紹搭建內建管道，讓數據高效傳遞給模型的方法。

本文將解釋 Dataset 的基本原理，包含大多數常用案例。

概述

使用 Dataset 需要遵循三個步驟：

• 載入數據：爲數據創建一個數據集實例。

• 創建一個迭代器：通過使用創建的數據集構建一個迭代器來對數據集進行迭代。

• 使用數據：通過使用創建的迭代器，我們可以找到可傳輸給模型的數據集元素。

載入數據

我們首先需要一些可以放入數據集的數據。

從 Numpy 導入

這是一種常見情況：我們擁有一個 numpy 數組，想把它傳遞給 TensorFlow。

# create a random vector of shape (100,2)x = np.random.sample((100,2))# make a dataset from a numpy arraydataset = 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))

從張量導入

當然，我們也可以從張量中初始化自己的數據集。

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

從佔位符導入

當我們希望動態地修改 Dataset 中的數據時，這就會很有用，稍後會有詳述。

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

從生成器導入

我們還可以從生成器中初始化 Dataset，這種方式在擁有不同長度的元素的數組時有意義（例如一個序列）。

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

在這種情況下，你還需要告訴 Dataset 數據的類型和形狀以創建正確的張量。

創建迭代器

我們已經學會創建數據集了，但如何從中獲取數據呢？我們必須使用迭代器（Iterator），它會幫助我們遍歷數據集中的內容並找到真值。有四種類型的迭代器。

One Shot 迭代器

這是最簡單的迭代器，使用第一個示例：

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

隨後你需要調用 get_next() 來獲取包含這些數據的張量

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

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

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

可初始化的迭代器

如果我們想要創建一個動態的數據集，在其中可以實時更改數據源，我們可以用佔位符創建一個數據集。隨後我們可以使用通常的 feed-dict 機制來初始化佔位符。這一過程可用「可初始化迭代器（initializable iterator）」來完成。使用上一節中的第三個例子：

# using a placeholderx = 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 iteratorel = 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 datasetEPOCHS = 10x, 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 numpytrain_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 testtrain_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 typeiter = tf.data.Iterator.from_structure(train_dataset.output_types,                                           train_dataset.output_shapes)

以及兩個初始化運算：

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

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

features, labels = iter.get_next()

現在，我們可以直接使用會話運行這兩個初始化運算。總結起來我們得到：

# Reinitializable iterator to switch between DatasetsEPOCHS = 10# making fake data using numpytrain_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 testtrain_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 typeiter = tf.data.Iterator.from_structure(train_dataset.output_types,                                           train_dataset.output_shapes)features, labels = iter.get_next()# create the initialisation operationstrain_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]))

可饋送的迭代器

老實說，我並不認爲這個有什麼用。基本上，它是用迭代器之間的轉換取代了數據集之間的轉換，從而得到如一個來自 make_one_shot_iterator() 的迭代器，以及一個來自 make_initializable_iterator() 的迭代器。

使用數據

在前述例子中，我們利用會話輸出 Dataset 中下一個元素的值。

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

爲了將數據傳遞給模型，我們只需要傳遞從 get_next() 生成的張量。在下面的代碼中，我們有一個包含了兩個 numpy 數組的 Dataset，這裏用了和第一節一樣的例子。注意，我們需要將.random.sample 封裝到另一個 numpy 數組，以增加一個維度，從而將數據進行分批。

# using two numpy arraysfeatures, 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 modelnet = tf.layers.dense(x, 8) # pass the first value from iter.get_next() as inputnet = 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 labeltrain_op = tf.train.AdamOptimizer().minimize(loss)

我們直接使用來自 iter.get_next() 的張量作爲第一層的輸入和損失函數的標籤。總結起來我們得到：

# make a simple modelEPOCHS = 10BATCH_SIZE = 16# using two numpy arraysfeatures, 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 modelnet = tf.layers.dense(x, 8, activation=tf.tanh) # pass the first value from iter.get_next() as inputnet = 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 labeltrain_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

一些有用的技巧

數據分批

通常數據分批是一件令人痛苦的事情，但通過 Dataset API，我們可以利用 batch(BATCH_SIZE) 方法自動地將數據集按設定的批量大小進行分批。默認批量大小爲 1。在下面的示例代碼中，我們使用的批量大小爲 4。

# BATCHINGBATCH_SIZE = 4x = np.random.sample((100,2))# make a dataset from a numpy arraydataset = 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() 進行數據集 shuffle，默認是在每一個 epoch 中將數據集 shuffle 一次。記住：數據集 shuffle 是避免過擬合的重要方法。

我們還可以設置參數 buffer_size，下一個元素將從該固定大小的緩存中均勻地選取。例如：

# BATCHINGBATCH_SIZE = 4x = np.array([[1],[2],[3],[4]])# make a dataset from a numpy arraydataset = 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]]

這樣，數據集 shuffle 就完成了。你還可以設置 seed 參數。

MAP

你可以使用 map 方法對數據集中的所有成員應用定製化函數。下列示例中，我們把每個元素乘 2：

# MAPx = np.array([[1],[2],[3],[4]])# make a dataset from a numpy arraydataset = 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 數據集教程：https://www.tensorflow.org/programmers_guide/datasets

• 數據集文檔：https://www.tensorflow.org/api_docs/python/tf/data/Dataset

結論

該數據集 API 使我們快速、穩健地創建優化輸入流程來訓練、評估和測試我們的模型。本文中，我們瞭解了很多可以常見操作。

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