tf.data.Dataset影象預處理詳解
阿新 • • 發佈：2018-11-15
目錄
1、tf.data.Dataset
當訓練集的樣本特別大時， 比較適合tf.data.Dataset作為資料輸入管線，相當方便。然而真正在使用tf.data.Dataset時，還是有許多坑，在這裡寫出來，當作參考。由於我只涉及影象處理，本文只專注於影象預處理相關內容。
本文的第二部分主要講一些講一些Dataset的常用函式；第三部分講了使用Tensorflow原生API來完成圖片預處理的方法；第四部分是使用tf.py_func來完成任意邏輯的預處理；第五部分是例子的完整程式碼。實際上，還有另外一種預處理資料的方法，就是先用不涉及tensorflow的純python程式碼來完成預處理，然後把處理後的資料（比如Numpy陣列）存到硬碟上，然後再使用tf.py_func使用相同的邏輯來讀取處理後的資料，這樣就不用每次訓練都預處理資料了。
參考連結如下：
2、Dataset常用函式
先來看一個例子
# 讀取filename指定的影象，並調整其大小。label是其對應的標籤
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, - tf.data.Dataset.from_tensor_slices((data, labels)):
建立一個Dataset例項。如果函式的引數包含NumPy陣列，並且未啟用Eager Execution，則值將作為一個或多個tf.constant操作嵌入到graph中。 對於大型資料集（> 1 GB），這可能會浪費記憶體並超過graph序列化（儲存圖的時候需要序列化）的位元組限制。 如果函式的引數包含一個或多個大型NumPy陣列，請參考替代方案。 - tf.data.Dataset.map(f, num_parallel_calls)
Dataset.map 轉換通過將函式 f 應用於輸入資料集的每對元素（data， label）來生成新資料集。比如在上面的例子中，就是把（filename， label）中filename指定的影象讀取出來並調整大小。
num_parallel_calls指定使用多少個執行緒來進行map操作。可以設定為CPU的最大核心數目（=multiprocessing.cpu_count()）。如果不指定的話，只使用一個執行緒順序處理資料。 - tf.data.Dataset.batch(batch_size)
這個函式特別重要。 假如輸入影象大小為（227，227，3），模型的輸入shape為（None，227，227，3），其中None是batch_size。如果不呼叫這個函式，那麼從dataset獲取一批資料時，返回的資料shape為（227，227，3），輸入到模型維度肯定匹配不上，就會出現如下類似的錯誤：
Index out of range using input dim 4; input has only 4 dims
或者
Error when checking target: expected softmax to have 2 dimensions, but got array with shape (250,)
如果呼叫了這個函式，再從dataset獲取一批資料時，返回的資料shape為（batch_size，227，227，3），就能和模型的輸入shape匹配上了。 - tf.data.Dataset.repeat(count)
重複這個資料集多少次。如果不傳count這個引數，預設會無限重複這個資料集。加入count=1，那麼當你訓練完一輪之後，就會報錯tensorflow.python.framework.errors_impl.OutOfRangeError: End of sequence。在實際使用中，基本可以不傳count引數，無限重複這個資料集。
常用函式還有tf.data.Dataset.shuffle()，是用來打亂資料集的。另外還需要注意的是：這些函式都是返回呼叫該操作之後的一個Dataset例項，並沒有在本身上應用該操作。
3、影象預處理的第一種方式
首先說下需求，主要是需要訓練一個分類模型。訓練集放在一個txt文字中，每一行是由圖片和標籤組成，一部分如下
data/test/001/001_01_01_051_09.png 0
data/test/001/001_01_01_051_10.png 0
data/test/002/002_01_01_051_19.png 1
data/test/002/002_01_01_051_09.png 1
data/test/003/003_01_01_051_14.png 2
data/test/003/003_01_01_051_03.png 2
data/test/004/004_01_01_051_05.png 3
data/test/004/004_01_01_051_06.png 3
...
現在需要把文字中的圖片和標籤讀入到一個Dataset裡面。
3.1、匯入依賴庫
# coding=utf-8
# 相容python3
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import multiprocessing as mt
import numpy as np
import tensorflow as tf
from tensorflow import keras
3.2、定義常量
# 分類問題，總共有250個類
NUM_CLASSES = 250
# 訓練批次大小
TRAIN_BATCH_SIZE = 128
# 影象每個畫素的每個通道的最大值，對於8點陣圖像，就是255
IMAGE_DEPTH=255
# 包含訓練集的文字
TRAIN_LIST = 'data/train.txt'
3.3、讀取文字中的圖片標籤對
# 讀取由path指定的文字檔案，並返回由很多(圖片路徑,標籤)組成的列表
lists_and_labels = np.loadtxt(path, dtype=str).tolist()
# 打亂下lists_and_labels
np.random.shuffle(lists_and_labels)
# 把圖片路徑和標籤分開
list_files, labels = zip(*[(l[0], int(l[1])) for l in lists_and_labels])
# 如果使用keras構建模型，還需要對標籤進行one_hot編碼，如果使用tensorflow構建的模型，則不需要。
one_shot_labels = keras.utils.to_categorical(labels, NUM_CLASSES).astype(dtype=np.int32)
3.4、例項化Dataset並完成影象預處理
# 定義資料集例項
dataset = tf.data.Dataset.from_tensor_slices((tf.constant(list_files), tf.constant(one_shot_labels)))
# 對沒一對 (image, label)呼叫_parse_image，完成影象的預處理
dataset = dataset.map(_parse_image, num_parallel_calls=mt.cpu_count())
# 設定訓練批次大小。非常重要！！！
dataset = dataset.batch(TRAIN_BATCH_SIZE)
# 無限重複資料集
dataset = dataset.repeat()
# 計算遍歷一遍資料集需要多少步
steps_per_epoch = np.ceil(len(labels) / TRAIN_BATCH_SIZE).astype(np.int32)
return dataset, steps_per_epoch
_parse_image函式需要實現的是：讀取圖片，調整大小，並將影象畫素值的範圍從[0, 255]縮放到[-0.5, 0.5]。_parse_image不能直接呼叫其他庫來實現功能，只能使用tensorflow中預定的操作來完成所需要的功能。實現如下：
def _parse_image(filename, label):
# 讀取並解碼圖片
image_string = tf.read_file(filename)
image_decoded = tf.image.decode_image(image_string)
# 一定要在這裡轉換型別！！！
image_converted = tf.cast(image_decoded, tf.float32)
# 縮放範圍
image_scaled = tf.divide(tf.subtract(image_converted, IMAGE_DEPTH/2), IMAGE_DEPTH)
return image_scaled, label
至此dataset就可以作為model.fit和model.evaluate（keras中）的引數了。
3.5、從Dataset中獲取資料
在使用Dataset作為使用tensorflow編寫的模型的輸入時，需要把資料取出來，作為feed_dict的引數的資料。另外，在使用Dataset作為模型輸入是，需要看看資料預處理的結果對不對，把資料取出來，看看實際的資料是否符合預期。
# 列印dataset的相關資訊
print('shapes:', dataset.output_shapes)
print('types:', dataset.output_types)
print('steps:', steps)
# 獲取一個用來迭代資料的iterator
data_it = dataset.make_one_shot_iterator()
# 定義個獲取下一組資料的操作(operator)
next_data = data_it.get_next()
# 新建Session
with tf.Session() as sess:
# 獲取前10組資料
for i in range(10):
# 獲取一批圖片和對應的標籤
data, label = sess.run(next_data)
# 列印資料的長度，標籤的長度，資料的shape，資料的最大值和最小值
print(len(data), len(label), data.shape, np.min(data), np.max(data))
執行上面的程式，輸出類似於
128 128 (128, 227, 227, 3) -0.5 0.5
128 128 (128, 227, 227, 3) -0.49607846 0.5
128 128 (128, 227, 227, 3) -0.5 0.5
128 128 (128, 227, 227, 3) -0.49607846 0.5
...
3.6、處理需要預測的樣本
預測（predict）樣本時，在預處理圖片時，預處理的操作一定要和訓練時的相同，否則評估或者預測的結果是不對的。在上面的方法中，預處理的程式碼為：
def read_image(filename):
with tf.Session() as sess:
read_op = _parse_image(tf.constant(filename, dtype=tf.string), tf.constant(0))
image, label = sess.run(read_op)
return image
image = read_image('data/test/001/001_01_01_051_09.png')
print('shape: ', image.shape)
在使用model.predict（keras）時，還需要擴充套件image的維度為四維，程式碼如下
# 讀圖片
image = read_image('data/train/022/022_01_01_051_00.png')
# 擴充套件維度為 (1, 227, 227, 3)
image = image[np.newaxis, :]
print(image.shape)
....
# 預測
softmax = model.predict(image, 1)
print(np.argmax(softmax)+1)
4、使用tf.py_func進行圖片預處理
有時候，需要完成特別複雜的預處理的時候，無法使用tensorflow內建的操作完成預處理，就可以使用tf.py_func來完成任意邏輯的預處理。先來個例子：
# coding=utf-8
import cv2
import tensorflow as tf
# 使用OpenCV程式碼來完成讀取圖片，在這個函式裡，你可以使用任意的python庫來完成任意操作
def _read_py_function(filename, label):
image_decoded = cv2.imread(filename.decode(), cv2.IMREAD_UNCHANGED)
return image_decoded, label
# 讀取圖片
def _read_image_caller(filename, label):
return tf.py_func(_read_py_function, [filename, label], [tf.uint8, label.dtype])
# 使用標準TensorFlow操作來調整圖片大小
def _resize_function(image_decoded, label):
# 由於無法從image_decoded推斷shape，所以要先手動設定
image_decoded.set_shape([None, None, None])
# 調整大小
image_resized = tf.image.resize_images(image_decoded, [28, 28])
return image_resized, label
filenames = ["data/train/001/001_01_01_051_04.png", "data/train/001/001_01_01_051_05.png", ]
labels = [0, 37, ]
# 定義dataset物件
dataset = tf.data.Dataset.from_tensor_slices((filenames, labels))
# 呼叫map，完成圖片讀取
dataset = dataset.map(_read_image_caller)
# 再次呼叫map，完成圖片的調整大小的操作
dataset = dataset.map(_resize_function)
# 定義獲取資料的tensorflow操作
next_op = dataset.make_one_shot_iterator().get_next()
with tf.Session() as sess:
for _ in range(len(labels)):
# 獲取下一組image,label影象對
image, label = sess.run(next_op)
print image.shape, label
tf.py_func的作用是把一個普通的python函式包裝（wrap）為tensorflow操作（類似於tf.read_file之類的），主要引數如下
- func: 指定要包裹的普通python函式F。
- inp: F所需要的引數組成的列表。
- Tout: 指定F返回值的型別。
4.1、來個例子
模型輸入的shape為（None，12， 227，227，3），其中的None是批次的大小，12是一個物體模型有12張圖片，（227，227，3）是一張影象的大小。所以預處理的要求是，把一個物體的12張圖片讀進來，完成調整大小，縮放畫素值的範圍到[-0.5, 0.5]，併疊在一起（shape為（12，227，227，3））。物體模型的列表和標籤train.txt如下：
data/train/001/list.txt 0
data/train/002/list.txt 1
data/train/003/list.txt 2
data/train/004/list.txt 3
data/train/005/list.txt 4
...
每一行的一個list.txt指定了一個物體模型的12張圖片，其中的一個如下：
data/train/001/001_01_01_051_14.png
data/train/001/001_01_01_051_19.png
data/train/001/001_01_01_051_18.png
data/train/001/001_01_01_051_10.png
data/train/001/001_01_01_051_07.png
data/train/001/001_01_01_051_16.png
data/train/001/001_01_01_051_04.png
data/train/001/001_01_01_051_17.png
data/train/001/001_01_01_051_13.png
data/train/001/001_01_01_051_15.png
data/train/001/001_01_01_051_11.png
data/train/001/001_01_01_051_05.png
4.2、匯入依賴庫
# coding=utf-8
# 相容python3
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import cv2
import numpy as np
import tensorflow as tf
import multiprocessing as mt
from tensorflow import keras
4.3、定義常量
# 分類問題，總共有40個類
NUM_CLASSES = 40
# 訓練批次大小
TRAIN_BATCH_SIZE = 1
# 影象每個畫素的每個通道的最大值，對於8點陣圖像，就是255
IMAGE_DEPTH=255
# 包含訓練集的文字
TRAIN_LIST = 'data/train.txt'
# 一個物體有12張圖片
NUM_VIEWS = 12
# 一張圖的大小
IMAGE_SHAPE = (227, 227, 3)
4.4、定義Dataset
這裡我這給出定義Dataset的函式
def prepare_dataset(path=''):
# 讀取物體模型列表
lists_and_labels = np.loadtxt(path, dtype=str).tolist()
# 打亂資料
np.random.shuffle(lists_and_labels)
# 分為模型列表和標籤
list_files, labels = zip(*[(l[0], int(l[1])) for l in lists_and_labels])
# 對標籤進行one_hot編碼
one_shot_labels = keras.utils.to_categorical(labels, NUM_CLASSES).astype(dtype=np.int32)
# 定義資料集
dataset = tf.data.Dataset.from_tensor_slices((tf.constant(list_files), tf.constant(one_shot_labels)))
# 讀取每個模型的12張圖片的路徑
dataset = dataset.map(read_object_caller, num_parallel_calls=mt.cpu_count())
# 調整每張圖片的大小，轉換圖片的資料型別為float32,並將12張圖片堆疊到一起
dataset = dataset.map(read_resize_concat, num_parallel_calls=mt.cpu_count())
# 非常重要，記得要設定批次大小
dataset = dataset.batch(TRAIN_BATCH_SIZE)
# 無限重複
dataset = dataset.repeat()
# 計算每次迭代需要多少步
steps_per_epoch = np.ceil(len(labels)/TRAIN_BATCH_SIZE).astype(np.int32)
return dataset, steps_per_epoch
4.5、讀取物體模型列表
我在寫程式碼的時候，讀取一個物體的12張圖片的路徑列表花了很久很久，就是因為不知道tf.py_func這個神器，接下來的程式碼，就是如何讀取一個物體模型的列表。
def read_object_caller(filename, label):
# 使用tf.py_func呼叫一個普通python函式來讀取一個物體的12張圖片路徑
# 注意返回值的型別是[tf.string, label.dtype]。
return tf.py_func(read_object_list, [filename, label], [tf.string, label.dtype])
def read_object_list(filename, label):
# 讀取一個物體模型的列表
image_lists = np.loadtxt(filename.decode(), dtype=str)
# 擷取前NUM_VIEWS個圖片路徑
image_lists = image_lists[:NUM_VIEWS]
# 如果圖片路徑沒有NUM_VIEWS個，丟擲錯誤
if len(image_lists) != NUM_VIEWS:
raise ValueError('There haven\'t %d views in %s ' % (NUM_VIEWS, filename))
# 返回圖片列表與標籤
return image_lists, label
4.5、圖片的預處理操作
def read_resize_concat(image_list, label):
# image_list是物體模型的12張圖片路徑的列表
# 下面這個函式就是處理列表中的每一個影象的函式
def process_one_image(image):
# 讀取圖片並解碼
image_string = tf.read_file(image)
image_decoded = tf.image.decode_image(image_string)
# 由於無法從image_decoded推斷shape，所以要先手動設定，否則resize_images會報錯
image_decoded.set_shape([None, None, None])
# 調整大小
image_resized = tf.image.resize_images(image_decoded, IMAGE_SHAPE[0:2])
# 轉換影象畫素型別
image_converted = tf.cast(image_resized, tf.float32)
# 把畫素值的範圍從[0, 255]縮放到[-0.5, 0.5]
image_scaled = tf.divide(tf.subtract(image_converted, IMAGE_DEPTH / 2), IMAGE_DEPTH)
return image_scaled
# 呼叫tf.map_fn對image_list的每個元素，也就是一張圖片的路徑，呼叫process_one_image函式，完成
# 對一張圖片的預處理，並返回一個處理後的list
image_prepocessed_list = tf.map_fn(process_one_image, image_list, dtype=tf.float32)
# 把12個處理後圖片在維度0上堆疊起來，一張圖片的shape為（227, 227, 3），堆疊後的shape為（12, 227,227,3）
concat = tf.concat(image_prepocessed_list, axis=0)
return concat, label
注意：tf.image.decode_image返回的image_decoded沒有shape，如果直接對image_decoded呼叫tf.image.resize_images，會出現如下錯誤ValueError: 'images' contains no shape.
4.6、從Dataset讀取資料
def inputs_test():
dataset, steps = prepare_dataset(TRAIN_LIST)
print('shapes:', dataset.output_shapes)
print('types:', dataset.output_types)
print('steps:', steps)
next_op = dataset.make_one_shot_iterator().get_next()
with tf.Session() as sess:
for i in range(5):
data, label = sess.run(next_op)
print(len(data), len(label), data.shape, np.min(data), np.max(data))
if __name__ == '__main__':
inputs_test()
4.7、獲取預測樣本
同樣，在預測時，樣本資料需要經過和訓練資料同樣的預處理，程式碼如下：
def process_one_sample(path):
label = 0
# 讀取圖片列表
image_list, _ = read_object_list(path, label)
# 定義處理操作
process_op = read_resize_concat(tf.constant(image_list), tf.constant(label))
# 處理
with tf.Session() as sess:
concat_image, _ = sess.run(process_op)
return concat_image
if __name__ == '__main__':
concat_image = process_one_sample('data/train/004/list.txt')
print(concat_image.shape)
5、兩種方法的完整程式碼
資料我就不提供了，自行準備吧。
5.1、第一種方法
# coding=utf-8
# 相容python3
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import multiprocessing as mt
import numpy as np
import tensorflow as tf
from tensorflow import keras
# 分類問題，總共有250個類
NUM_CLASSES = 250
# 訓練批次大小
TRAIN_BATCH_SIZE = 128
# 影象每個畫素的每個通道的最大值，對於8點陣圖像，就是255
IMAGE_DEPTH = 255
# 包含訓練集的文字
TRAIN_LIST = 'data/train.txt'
def prepare_dataset(path=''):
"""
prepaer dataset using tf.data.Dataset
:param path: the list file like data/train_lists_demo.txt
and data/val_lists_demo.txt
:return: a Dataset object
"""
# read image list files name and labels
lists_and_labels = np.loadtxt(path, dtype=str).tolist()
# shuffle dataset
np.random.shuffle(lists_and_labels)
# split lists an labels
list_files, labels = zip(*[(l[0], int(l[1])) for l in lists_and_labels])
# one_shot encoding on labels
one_shot_labels = keras.utils.to_categorical(labels, NUM_CLASSES).astype