TensorFlow：tf.data.Dataset 的用法 (二)

上一篇文章已經介紹了 tf.data.Dataset 這個 API，也找出我最常用的兩個方式，一個是 tf.data.Dataset.from_tensor_slices，另一個是 tf.data.Dataset.from_generator。

但是我在複習相關程式碼過程中，發現其實只要用tf.data.Dataset.from_tensor_slices，就能滿足我所有需求了。

資料讀取的情況，我目前大致分成三種：

1. 先用 NumPy Array 進行前處理，再轉成 Dataset 形式。
2. 先把 NumPy Array 轉成 Dataset 形式，再使用 map() 函數進行前處理。
3. 建立資料路徑清單，將清單轉成 Dataset 形式，再用 map() 函數進行資料讀取以及一些前處理。

那麼第一種其實是上一篇的最後一個範例，有興趣可以點連結去看 (點我)，另外兩個反而是我真的花較多時間測試，因為如果可以把前處理寫進 Dataset 裡，代表我可以更有效處理我的資料，也可以更進一步的去實現第三種讀取資料的方式了。

那為何需要用路徑去讀取呢？答案其實很簡單，如果今天資料量多到記憶體放不下，那勢必就無法很好的丟資料給模型訓練，甚至無法訓練都有可能。但今天如果只用路徑 (其實就是字串)，所佔的記憶體容量絕對比影像資料要小很多，這也是為什麼如果能將一連串的路徑字串清單建成 Dataset，再來批次讀取的原因，這樣就算個人電腦等級的配備，也是有機會訓練較龐大的影像資料庫。

那這篇文章就先講我如何實作第二種方式，並搭配 zip()、map() 函式來進行資料整合及資料擴增。

NumPy Array convert to Dataset

首先，先講 zip() 函式如何整合 Dataset。這邊方式和上一篇的最後一個範例有點類似，但我是把影像和標記分別用 from_tensor_slices() 建立出 Dataset 物件，再用 zip() 給結合起來。

• 先匯入相關的模組

import tensorflow as tf
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.datasets import mnist
from tensorflow.data import Dataset

• 設定 Dataset 參數

batch_size = 100
split = 0.8

• 讀取 NumPy Array，並進行前處理 (正規化、one-hot encoding)。

### keras.datasets.mnist  (NumPy Array)
(tra_im, tra_lb), (tes_im, tes_lb) = mnist.load_data()
# 正規化
tra_im_norm = tra_im / 255.0
tes_im_norm = tes_im / 255.0
# one-hot encoding
tra_lb_onehot = to_categorical(tra_lb)
tes_lb_onehot = to_categorical(tes_lb)

• 製作 Dataset 物件

# 分離訓練資料  -->  [0.8, 0.2] = [train, valid]
split_idx = int(len(tra_im)*split)

# training data
tra_ds_im = Dataset.from_tensor_slices(tra_im_norm[:split_idx])    # 影像 Dataset
tra_ds_lb = Dataset.from_tensor_slices(tra_lb_onehot[:split_idx])  # 標記 Dataset
tra_ds = Dataset.zip((tra_ds_im, tra_ds_lb))  # 影像、標記整合成一個 Dataset
tra_ds = tra_ds.batch(batch_size)  # 設定 Dataset 批次大小
tra_ds = tra_ds.shuffle(split_idx) # 打亂 Dataset

# validation data
val_ds_im = Dataset.from_tensor_slices(tra_im_norm[split_idx:])    # 影像 Dataset
val_ds_lb = Dataset.from_tensor_slices(tra_lb_onehot[split_idx:])  # 標記 Dataset
val_ds = Dataset.zip((val_ds_im, val_ds_lb))  # 影像、標記整合成一個 Dataset
val_ds = val_ds.batch(batch_size)  # 設定 Dataset 批次大小
val_ds = val_ds.shuffle(len(tra_im)-split_idx) # 打亂 Dataset

# testing data
tes_ds_im = Dataset.from_tensor_slices(tes_im_norm)    # 影像 Dataset
tes_ds_lb = Dataset.from_tensor_slices(tes_lb_onehot)  # 標記 Dataset
tes_ds = Dataset.zip((tes_ds_im, tes_ds_lb))  # 影像、標記整合成一個 Dataset
tes_ds = tes_ds.batch(batch_size)  # 設定 Dataset 批次大小

這邊可能和官方範例方式不太一樣，似乎也比較複雜，但我主要是確認如果我要分開處理影像或是標記資料，其實是可行的。

Dataset with Data Augmentation

搭配 map() 來設定資料擴增的函數。

• 先來定義一些資料擴增的 function，定義的功能如下：

(1) 隨機水平翻轉

def flip_h(x):
    x = tf.image.random_flip_left_right(x)
    return x

(2) 隨機垂直翻轉

def flip_v(x):
    x = tf.image.random_flip_up_down(x)
    return x

(3) 隨機旋轉

def rotate(x):
    k = tf.random.uniform([], 1, 4, tf.int32)
    x = tf.image.rot90(x, k)
    return x

(4) 隨機色調

def hue(x, val=0.08):  # 色調
    x = tf.image.random_hue(x, val)
    return x

(5) 隨機亮度

def brightness(x, val=0.05):  # 亮度
    x = tf.image.random_brightness(x, val)
    return x

(6) 隨機飽和度

def saturation(x, minval=0.6, maxval=1.6):  # 飽和度
    x = tf.image.random_saturation(x, minval, maxval)
    return x

(7) 隨機對比度

def contrast(x, minval=0.7, maxval=1.3):  # 對比度
    x = tf.image.random_contrast(x, minval, maxval)
    return x

(8) 隨機縮放

def zoom(x, scale_minval=0.5, scale_maxval=1.5):
    height, width, channel = x.shape
    scale = tf.random.uniform([], scale_minval, scale_maxval)
    new_size = (scale*height, scale*width)
    x = tf.image.resize(x, new_size)
    x = tf.image.resize_with_crop_or_pad(x, height, width)
    return x

• 撰寫輸入函數

這個自訂 function 會去接收 Dataset 作為引數，我還設定 **kwargs 作為可變關鍵字引數，主要是用來設定資料擴增的選項。

def parse_fn(dataset, **kwargs):
    if kwargs:
        print("Data Augmentation!!!")
        for k, v in kwargs.items():
            print("%15s:"%k, v)
    else:
        print("Not Data Augmentation!!!")
    print()

    # 分離 dataset
    x = dataset["image"]
    y = dataset["label"]

    # 對影像進行正規化，及增加影像通道
    # 因為 MNIST 是灰階影像，所以要自行加上通道，也就是第三軸
    # 從 sahpe 來看就是變成 (28, 28)  ==>  (28, 28, 1)
    x = tf.cast(x, tf.float32) / 255.0
    x = tf.expand_dims(x, axis=-1)

    # 從 kwargs 來判斷哪些擴增需要執行
    if kwargs.get("flip_h", None):
        x = tf.cond(tf.random.uniform((), 0, 1) > 0.5,
                    lambda: flip_h(x), lambda: x)
    if kwargs.get("flip_v", None):
        x = tf.cond(tf.random.uniform((), 0, 1) > 0.5,
                    lambda: flip_v(x), lambda: x)
    if kwargs.get("rotate", None):
        x = tf.cond(tf.random.uniform((), 0, 1) > 0.5,
                    lambda: rotate(x), lambda: x)
    if kwargs.get("hue", None):
        x = tf.cond(tf.random.uniform((), 0, 1) > 0.5,
                    lambda: hue(x), lambda: x)
    if kwargs.get("brightness", None):
        x = tf.cond(tf.random.uniform((), 0, 1) > 0.5,
                    lambda: brightness(x), lambda: x)
    if kwargs.get("saturation", None):
        x = tf.cond(tf.random.uniform((), 0, 1) > 0.5,
                    lambda: saturation(x), lambda: x)
    if kwargs.get("contrast", None):
        x = tf.cond(tf.random.uniform((), 0, 1) > 0.5,
                    lambda: contrast(x), lambda: x)
    if kwargs.get("zoom_scale", None):
        x = tf.cond(tf.random.uniform((), 0, 1) > 0.5,
                    lambda: zoom(x), lambda: x)

    # 對標記進行 one-hot encoding
    y = tf.one_hot(y, 10)
    return {"image": x}, {"label": y} # 回傳資料 (個人喜歡採用 dict 形式)

有了上面的 function，就可以來製作資料擴增的 Dataset 了。

• 設定 Dataset 參數

batch_size = 100
split = 0.8
augdict = {
    # "flip_h": True,
    # "flip_v": True,
    # "rotate": True,
    "hue": False,
    "saturation": False,
    "contrast": True,
    "brightness": True,
    "zoom_scale": False,
}

• 讀取 NumPy Array，並做成訓練、驗證、測試的 dict。

(tra_im, tra_lb), (tes_im, tes_lb) = mnist.load_data()
#-----------------------------------------------------------------------------#
split_idx = int(len(tra_im)*split)
tra_data = {"image": tra_im[:split_idx], "label": tra_lb[:split_idx]}
val_data = {"image": tra_im[split_idx:], "label": tra_lb[split_idx:]}
tes_data = {"image": tes_im, "label": tes_lb}
#-----------------------------------------------------------------------------#

• 那些 dict 弄成 Dataset，再用 map() 來加入剛剛自行撰寫的 parse_fn()。

這裡有幾點要注意的事：

(1) 由於傳入的是 dict，所以 parse_fn() 也是接收到 dict 的 Dataset。

(2) 因為訓練以及驗證集都有設定資料擴增，可以發現傳入之前，有先用 lambda 匿名函式包裝起來，才能將我的擴增參數 augdict 輸入進 parse_fn()中。

(3) 由於 map() 會去傳遞我的 dataset，所以 lambda 匿名函式需要設定一個輸入參數給他。因為我是傳遞一個 dict 型態的 Dataset，所以參數只有設定一個，我將其命名為 ds。

(4) 擴增參數 augdict 輸入進 parse_fn()中的時候，必須要加上 ** 來解析我的 dict，這樣才會變成是輸入關鍵字引數。

(5) autotune = tf.data.experimental.AUTOTUNE 是用來自動調整 CPU 數量，讓 TensorFlow 自行決定如何使用 CPU 的核心。

autotune = tf.data.experimental.AUTOTUNE
#-----------------------------------------------------------------------------#
tra_ds = Dataset.from_tensor_slices(tra_data)
tra_ds = tra_ds.map(lambda ds: parse_fn(ds, **augdict), num_parallel_calls=autotune)
tra_ds = tra_ds.shuffle(1000).batch(batch_size)
#-----------------------------------------------------------------------------#
val_ds = Dataset.from_tensor_slices(val_data)
val_ds = val_ds.map(lambda ds: parse_fn(ds, **augdict), num_parallel_calls=autotune)
val_ds = val_ds.shuffle(1000).batch(batch_size)
#-----------------------------------------------------------------------------#
tes_ds = Dataset.from_tensor_slices(tes_data)
tes_ds = tes_ds.map(parse_fn, num_parallel_calls=autotune)
tes_ds = tes_ds.batch(batch_size)

結語

這篇文章重點其實是後半段的資料擴增方式，因為在 TensorFlow 讀取資料實在有太多方法，甚至沿用以前的方式也是可以進行訓練，只是有更高效的方法就還是想要嘗試看看，順便將相關的程式記錄起來，也分享自己的程式。接下來就是朝第三篇邁進，雖然來只打算兩篇就好，但實在太多文章、書籍都是直接用現成的資料庫，如果要讀自己的資料都要花上許多時間研究，希望自己的程式可以幫助那些想讀自己資料卻無所適從的人。