『TensorFlow』數據讀取類_data.Dataset
一、資料
參考原文:
TensorFlow全新的數據讀取方式:Dataset API入門教程
API接口簡介:
TensorFlow的數據集
二、背景
註意,在TensorFlow 1.3中,Dataset API是放在contrib包中的:
tf.contrib.data.Dataset
而在TensorFlow 1.4中,Dataset API已經從contrib包中移除,變成了核心API的一員:
tf.data.Dataset
此前,在TensorFlow中讀取數據一般有兩種方法:
- 使用placeholder讀內存中的數據
- 使用queue讀硬盤中的數據(這種方式,可以參考原作者之前的一篇文章:十圖詳解TensorFlow數據讀取機制)
Dataset API同時支持從內存和硬盤的讀取,相比之前的兩種方法在語法上更加簡潔易懂。此外,如果想要用到TensorFlow新出的Eager模式,就必須要使用Dataset API來讀取數據。
三、基本使用
1、一維數據集示範基本使用
Google官方給出的Dataset API中的類圖:
在初學時,我們只需要關註兩個最重要的基礎類:Dataset和Iterator。
Dataset可以看作是相同類型“元素”的有序列表。在實際使用時,單個“元素”可以是向量,也可以是字符串、圖片,甚至是tuple或者dict。
數據集對象實例化:
dataset = tf.data.Dataset.from_tensor_slices(數據)
叠代器對象實例化(非Eager模式下):
iterator = dataset.make_one_shot_iterator()
one_element = iterator.get_next()
綜合起來效果如下,
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 2.0 3.0 4.0 5.0
讀取結束異常:
如果一個dataset中元素被讀取完了,再嘗試sess.run(one_element)的話,就會拋出tf.errors.OutOfRangeError異常,這個行為與使用隊列方式讀取數據的行為是一致的。
在實際程序中,可以在外界捕捉這個異常以判斷數據是否讀取完,綜合以上三點請參考下面的代碼:
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(config=config) as sess: try: while True: print(sess.run(one_element)) except tf.errors.OutOfRangeError: print("end!")
輸出:1.0 2.0 3.0 4.0 5.0 end!
2、高維數據集使用
tf.data.Dataset.from_tensor_slices真正作用是切分傳入Tensor的第一個維度,生成相應的dataset,即第一維表明數據集中數據的數量,之後切分batch等操作都以第一維為基礎。
dataset = tf.data.Dataset.from_tensor_slices(np.random.uniform(size=(5, 2)))
傳入的數值是一個矩陣,它的形狀為(5, 2),tf.data.Dataset.from_tensor_slices就會切分它形狀上的第一個維度,最後生成的dataset中一個含有5個元素,每個元素的形狀是(2, ),即每個元素是矩陣的一行。
dataset = tf.data.Dataset.from_tensor_slices(np.random.uniform(size=(5, 2))) iterator = dataset.make_one_shot_iterator() one_element = iterator.get_next() with tf.Session(config=config) as sess: try: while True: print(sess.run(one_element)) except tf.errors.OutOfRangeError: print("end!")
[0.09787406 0.71672957] [0.25681324 0.81974072] [0.35186046 0.39362398] [0.75228199 0.6534702 ] [0.39695169 0.9341708 ] end!
3、字典使用
在實際使用中,我們可能還希望Dataset中的每個元素具有更復雜的形式,如每個元素是一個Python中的元組,或是Python中的詞典。例如,在圖像識別問題中,一個元素可以是{“image”: image_tensor, “label”: label_tensor}的形式,這樣處理起來更方便,
註意,image_tensor、label_tensor和上面的高維向量一致,第一維表示數據集中數據的數量。相較之下,字典中每一個key值可以看做數據的一個屬性,value則存儲了所有數據的該屬性值。
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)) }) iterator = dataset.make_one_shot_iterator() one_element = iterator.get_next() with tf.Session(config=config) as sess: try: while True: print(sess.run(one_element)) except tf.errors.OutOfRangeError: print("end!")
{‘a‘: 1.0, ‘b‘: array([0.31721037, 0.33378767])} {‘a‘: 2.0, ‘b‘: array([0.99221946, 0.65894961])} {‘a‘: 3.0, ‘b‘: array([0.98405468, 0.11478854])} {‘a‘: 4.0, ‘b‘: array([0.95311317, 0.57432678])} {‘a‘: 5.0, ‘b‘: array([0.46067428, 0.19716722])} end!
4、復雜的tuple組合數據
類似的,可以使用組合的特征進行拼接,
dataset = tf.data.Dataset.from_tensor_slices( (np.array([1.0, 2.0, 3.0, 4.0, 5.0]), np.random.uniform(size=(5, 2))) ) iterator = dataset.make_one_shot_iterator() one_element = iterator.get_next() with tf.Session(config=config) as sess: try: while True: print(sess.run(one_element)) except tf.errors.OutOfRangeError: print("end!")
(1.0, array([6.55877282e-04, 6.63244735e-01])) (2.0, array([0.04756927, 0.44968581])) (3.0, array([0.97841076, 0.06465231])) (4.0, array([0.46639246, 0.39146086])) (5.0, array([0.61085016, 0.61609538])) end!
四、數據集處理方法
Dataset支持一類特殊的操作:Transformation。一個Dataset通過Transformation變成一個新的Dataset。通常我們可以通過Transformation完成數據變換,打亂,組成batch,生成epoch等一系列操作。
常用的Transformation有:
- map
- batch
- shuffle
- repeat
map
和python中的map類似,map接收一個函數,Dataset中的每個元素都會被當作這個函數的輸入,並將函數返回值作為新的Dataset,
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) # <----- iterator = dataset.make_one_shot_iterator() one_element = iterator.get_next() with tf.Session(config=config) as sess: try: while True: print(sess.run(one_element)) except tf.errors.OutOfRangeError: print("end!")
輸出:2.0 3.0 4.0 5.0 6.0 end!
註意map函數可以使用num_parallel_calls參數加速(第五部分有介紹)。
batch
batch就是將多個元素組合成batch,如上所說,按照輸入元素第一個維度,
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)) }) dataset = dataset.batch(2) # <----- iterator = dataset.make_one_shot_iterator() one_element = iterator.get_next() with tf.Session(config=config) as sess: try: while True: print(sess.run(one_element)) except tf.errors.OutOfRangeError: print("end!")
{‘a‘: array([1., 2.]), ‘b‘: array([[0.87466134, 0.21519021], [0.6123372 , 0.95722733]])} {‘a‘: array([3., 4.]), ‘b‘: array([[0.76964374, 0.22445015], [0.08313089, 0.60531841]])} {‘a‘: array([5.]), ‘b‘: array([[0.37901654, 0.3955096 ]])} end!
shuffle
shuffle的功能為打亂dataset中的元素,它有一個參數buffersize,表示打亂時使用的buffer的大小,建議舍的不要太小,一般是1000:
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)) }) dataset = dataset.shuffle(buffer_size=5) # <----- iterator = dataset.make_one_shot_iterator() one_element = iterator.get_next() with tf.Session(config=config) as sess: try: while True: print(sess.run(one_element)) except tf.errors.OutOfRangeError: print("end!")
{‘a‘: 3.0, ‘b‘: array([0.82048268, 0.39821839])} {‘a‘: 4.0, ‘b‘: array([0.42775421, 0.36749283])} {‘a‘: 1.0, ‘b‘: array([0.09588742, 0.01954797])} {‘a‘: 2.0, ‘b‘: array([0.10992948, 0.24416772])} {‘a‘: 5.0, ‘b‘: array([0.15447616, 0.09005545])} end!
repeat
repeat的功能就是將整個序列重復多次,主要用來處理機器學習中的epoch,假設原先的數據是一個epoch,使用repeat(2)就可以將之變成2個epoch:
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)) }) dataset = dataset.repeat(2) # <----- iterator = dataset.make_one_shot_iterator() one_element = iterator.get_next() with tf.Session(config=config) as sess: try: while True: print(sess.run(one_element)) except tf.errors.OutOfRangeError: print("end!")
{‘a‘: 1.0, ‘b‘: array([0.85180201, 0.1703507 ])} {‘a‘: 2.0, ‘b‘: array([0.37874819, 0.81303628])} {‘a‘: 3.0, ‘b‘: array([0.99560094, 0.56446562])} {‘a‘: 4.0, ‘b‘: array([0.86341794, 0.69984075])} {‘a‘: 5.0, ‘b‘: array([0.85026424, 0.74761098])} {‘a‘: 1.0, ‘b‘: array([0.85180201, 0.1703507 ])} {‘a‘: 2.0, ‘b‘: array([0.37874819, 0.81303628])} {‘a‘: 3.0, ‘b‘: array([0.99560094, 0.56446562])} {‘a‘: 4.0, ‘b‘: array([0.86341794, 0.69984075])} {‘a‘: 5.0, ‘b‘: array([0.85026424, 0.74761098])} end!
註意,如果直接調用repeat()的話,生成的序列就會無限重復下去,沒有結束,因此也不會拋出tf.errors.OutOfRangeError異常。
五、模擬讀入磁盤圖片與對應label
考慮一個簡單,但同時也非常常用的例子:讀入磁盤中的圖片和圖片相應的label,並將其打亂,組成batch_size=32的訓練樣本,在訓練時重復10個epoch
# 函數的功能時將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)
在這個過程中,dataset經歷三次轉變:
- 運行dataset = tf.data.Dataset.from_tensor_slices((filenames, labels))後,dataset的一個元素是(filename, label)。filename是圖片的文件名,label是圖片對應的標簽。
- 之後通過map,將filename對應的圖片讀入,並縮放為28x28的大小。此時dataset中的一個元素是(image_resized, label)
-
最後,dataset.shuffle(buffersize=1000).batch(32).repeat(10)的功能是:在每個epoch內將圖片打亂組成大小為32的batch,並重復10次。最終,dataset中的一個元素是(image_resized_batch, label_batch),image_resized_batch的形狀為(32, 28, 28, 3),而label_batch的形狀為(32, ),接下來我們就可以用這兩個Tensor來建立模型了。
help(tf.data.Dataset.map)
可見:
Help on function map in module tensorflow.python.data.ops.dataset_ops: map(self, map_func, num_parallel_calls=None) Maps `map_func` across this datset. Args: map_func: A function mapping a nested structure of tensors (having shapes and types defined by `self.output_shapes` and `self.output_types`) to another nested structure of tensors. num_parallel_calls: (Optional.) A `tf.int32` scalar `tf.Tensor`, representing the number elements to process in parallel. If not specified, elements will be processed sequentially. Returns: A `Dataset`.
由此可見map作為讀取處理的關鍵步驟,是可以多線程加速的。
六、更多的Dataset創建方法
除了tf.data.Dataset.from_tensor_slices外,目前Dataset API還提供了另外三種創建Dataset的方式:
- tf.data.TextLineDataset():這個函數的輸入是一個文件的列表,輸出是一個dataset。dataset中的每一個元素就對應了文件中的一行。可以使用這個函數來讀入CSV文件。
- tf.data.FixedLengthRecordDataset():這個函數的輸入是一個文件的列表和一個record_bytes,之後dataset的每一個元素就是文件中固定字節數record_bytes的內容。通常用來讀取以二進制形式保存的文件,如CIFAR10數據集就是這種形式。
- tf.data.TFRecordDataset():顧名思義,這個函數是用來讀TFRecord文件的,dataset中的每一個元素就是一個TFExample。
它們的詳細使用方法可以參閱文檔:Module: tf.data
七、更多的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中,實現更為靈活的數據載入,實際上占位符引入了dataset對象創建中,我們可以通過feed來控制數據集合的實際情況。
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) print(value) assert i == value
輸出:0 1 2 3 4 5 6 7 8 9
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})
可見,在上面程序中,feed也遵循著類似字典一樣的規則,創建兩個占位符(keys),給data_holder去feed數據文件,給label_holder去feed標簽文件。
reinitializable iterator和feedable iterator相比initializable iterator更復雜,也更加少用,如果想要了解它們的功能,可以參閱官方介紹,這裏就不再贅述了。
八、總結
在非Eager模式下,Dataset中讀出的一個元素一般對應一個batch的Tensor,我們可以使用這個Tensor在計算圖中構建模型。
在Eager模式下,Dataset建立Iterator的方式有所不同,此時通過讀出的數據就是含有值的Tensor,方便調試。
『TensorFlow』數據讀取類_data.Dataset