A detailed example of how to use data generators with Keras

Motivation
你是否曾經加載一個需要消耗大量記憶體的數據集，讓你希望有一個神奇的技巧可以無縫地處理它？ 隨著我們能夠利用不斷增長的數據量，大型數據集正日益成為我們生活的一部分。

我們必須記住，在某些情況下，即使是最先進的配置也沒有足夠的內存空間來處理數據，就像我們過去那樣。 這就是為什麼我們需要找到其他方法來有效地完成這項任務的原因。 在這篇部落格中，我們將向您展示如何在多個核心上實時生成數據集，並立即將其提供給您的深度學習模型。


本教程中使用的框架是Python的高級軟件包Keras提供的框架，可以在TensorFlow或Theano的GPU安裝之上使用。

Tutorial


Previous situation
在閱讀本文之前，您的Keras腳本可能如下所示：

import numpy as np
from keras.models import Sequential

# Load entire dataset
X, y = np.load('some_training_set_with_labels.npy')

# Design model
model = Sequential()
[...] # Your architecture
model.compile()

# Train model on your dataset
model.fit(x=X, y=y)

本文是有關加載整個數據集的程式碼做修改。 實際上，這項任務可能會導致問題，因為所有訓練樣本可能無法同時適應記憶體。

為了做到這一點，讓我們深入一步一步建立一個適合這種情況的數據生成器。 順便說一句，下面的代碼是一個很好的框架，可用於您自己的項目; 您可以複製/貼上以下代碼並填充在空白處。

Notations
在開始之前，讓我們先了解一些在處理大型數據集時特別有用的組織技巧。

設ID為Python字符串，用於標識數據集的給定樣本。 追蹤樣本及其標籤的好方法是採用以下框架：

1.創建一個名為partition的dictionary，其中你收集：

（i）在partition['train']中訓練ID的list
（ii）在partition['validation']中驗證ID的list

2.創建一個名為labels的字典，其中對於數據集的每個ID，關聯的標籤由labels[ID]給出

例如，假設我們的訓練集包含id-1，id-2和id-3，各自的標籤為0,1和2，驗證集包含id-4，標籤為1.在這種情況下，Python變量partition和labels看起來像

>>> partition
{'train': ['id-1', 'id-2', 'id-3'], 'validation': ['id-4']}

以及

>>> labels
{'id-1': 0, 'id-2': 1, 'id-3': 2, 'id-4': 1}

此外，為了模組化，我們將在單獨的文件中編寫Keras代碼和自定義類別，以便您的文件夾看起來像

folder/
├── my_classes.py
├── keras_script.py
└── data/

其中data/被假定為包含數據集的文件夾。

最後，值得注意的是，本教程中的代碼旨在實現通用和最小化，以便您可以輕鬆地將其用於您自己的數據集。

Data generator
現在，讓我們詳細了解如何設置Python class DataGenerator，它將用於向Keras模型提供即時數據傳輸。

首先，讓我們編寫class的初始化函數。 我們使後者繼承keras.utils.Sequence的屬性，以便我們可以利用像是multiprocessing之類的好功能。

def __init__(self, list_IDs, labels, batch_size=32, dim=(32,32,32), n_channels=1,
             n_classes=10, shuffle=True):
    'Initialization'
    self.dim = dim
    self.batch_size = batch_size
    self.labels = labels
    self.list_IDs = list_IDs
    self.n_channels = n_channels
    self.n_classes = n_classes
    self.shuffle = shuffle
    self.on_epoch_end()

我們把關於數據的相關資訊作為引數，例如尺寸大小（例如，長度為32的體積將具有dim =（32,32,32）），通道數，類別數，批量大小，或決定我們是否需要改變我們的數據。 我們還存儲重要資訊，例如標籤和我們希望在每次傳遞時生成的ID list。

這裡，方法on_epoch_end在每個epoch的最開始和結束時被觸發一次。 如果shuffle參數設置為True，我們將在每次傳遞時獲得新的順序洗牌（否則只保留線性方案）。

def on_epoch_end(self):
  'Updates indexes after each epoch'
  self.indexes = np.arange(len(self.list_IDs))
  if self.shuffle == True:
      np.random.shuffle(self.indexes)

將範例輸入分類器的順序洗牌是有幫助的，因此epochs之間的批次看起來不相似。 這樣做最終會使我們的模型更加健壯。

另一種最關鍵的生成過程核心的方法是：生成批量數據。 負責此任務的私有方法稱為__data_generation，並將目標批次的ID list作為引數。

def __data_generation(self, list_IDs_temp):
  'Generates data containing batch_size samples' # X : (n_samples, *dim, n_channels)
  # Initialization
  X = np.empty((self.batch_size, *self.dim, self.n_channels))
  y = np.empty((self.batch_size), dtype=int)

  # Generate data
  for i, ID in enumerate(list_IDs_temp):
      # Store sample
      X[i,] = np.load('data/' + ID + '.npy')

      # Store class
      y[i] = self.labels[ID]

  return X, keras.utils.to_categorical(y, num_classes=self.n_classes)

在數據生成期間，此代碼從其對應的文件ID.npy中讀取每個範例的NumPy陣列。 由於我們的代碼是多核的，請注意您可以執行更複雜的操作（例如，從源文件進行計算），而不必擔心數據生成成為訓練過程中的瓶頸。

另外，請注意我們使用Keras的keras.utils.to_categorical函數將我們存儲在y中的數字標籤轉換為二進制形式（例如，在6級問題中，第三個標籤對應於[0 0 1 0 0 0]） 適合分類。

現在，我們將所有這些組件組合在一起。 每次呼叫都請求批次索引在0和批次總數之間，後者在__len__方法中指定。

def __len__(self):
  'Denotes the number of batches per epoch'
  return int(np.floor(len(self.list_IDs) / self.batch_size))

通常的做法是將此值設置為

$$\lfloor \frac{\#\text{samples}}{\text{batch size}}\rfloor$$

這樣模型每個epoch最多可以看到一次訓練樣本。

現在，當特定的索引對應批次被呼叫時，生成器執行__getitem__方法來生成它。

def __getitem__(self, index):
  'Generate one batch of data'
  # Generate indexes of the batch
  indexes = self.indexes[index*self.batch_size:(index+1)*self.batch_size]

  # Find list of IDs
  list_IDs_temp = [self.list_IDs[k] for k in indexes]

  # Generate data
  X, y = self.__data_generation(list_IDs_temp)

  return X, y

與我們在本節中描述的步驟相對應的完整代碼如下所示。

import numpy as np
import keras

class DataGenerator(keras.utils.Sequence):
    'Generates data for Keras'
    def __init__(self, list_IDs, labels, batch_size=32, dim=(32,32,32), n_channels=1,
                 n_classes=10, shuffle=True):
        'Initialization'
        self.dim = dim
        self.batch_size = batch_size
        self.labels = labels
        self.list_IDs = list_IDs
        self.n_channels = n_channels
        self.n_classes = n_classes
        self.shuffle = shuffle
        self.on_epoch_end()

    def __len__(self):
        'Denotes the number of batches per epoch'
        return int(np.floor(len(self.list_IDs) / self.batch_size))

    def __getitem__(self, index):
        'Generate one batch of data'
        # Generate indexes of the batch
        indexes = self.indexes[index*self.batch_size:(index+1)*self.batch_size]

        # Find list of IDs
        list_IDs_temp = [self.list_IDs[k] for k in indexes]

        # Generate data
        X, y = self.__data_generation(list_IDs_temp)

        return X, y

    def on_epoch_end(self):
        'Updates indexes after each epoch'
        self.indexes = np.arange(len(self.list_IDs))
        if self.shuffle == True:
            np.random.shuffle(self.indexes)

    def __data_generation(self, list_IDs_temp):
        'Generates data containing batch_size samples' # X : (n_samples, *dim, n_channels)
        # Initialization
        X = np.empty((self.batch_size, *self.dim, self.n_channels))
        y = np.empty((self.batch_size), dtype=int)

        # Generate data
        for i, ID in enumerate(list_IDs_temp):
            # Store sample
            X[i,] = np.load('data/' + ID + '.npy')

            # Store class
            y[i] = self.labels[ID]

        return X, keras.utils.to_categorical(y, num_classes=self.n_classes)

Keras script
現在，我們必須相應地修改我們的Keras腳本，以便它接受我們剛剛創建的生成器。

import numpy as np

from keras.models import Sequential
from my_classes import DataGenerator

# Parameters
params = {'dim': (32,32,32),
          'batch_size': 64,
          'n_classes': 6,
          'n_channels': 1,
          'shuffle': True}

# Datasets
partition = # IDs
labels = # Labels

# Generators
training_generator = DataGenerator(partition['train'], labels, **params)
validation_generator = DataGenerator(partition['validation'], labels, **params)

# Design model
model = Sequential()
[...] # Architecture
model.compile()

# Train model on dataset
model.fit_generator(generator=training_generator,
                    validation_data=validation_generator,
                    use_multiprocessing=True,
                    workers=6)

如您所見，我們從model中呼叫fit_generator方法而不是fit，我們只需將訓練生成器作為參數之一。 Keras負責其餘的事情！

請注意，我們的執行允許使用fit_generator的multiprocessing引數，其中n_workers中指定的threads數是平行生成批處理的threads數。 足夠多的工作人員確保有效地管理CPU計算，換句話說，瓶頸實際上是神經網絡在GPU上的前向和後向操作（而不是數據生成）。

Conclusion
就是這個！ 您現在可以使用該命令執行Keras腳本

python3 keras_script.py

你會發現在訓練階段，數據由CPU平行生成，然後直接輸入GPU。

您可以在GitHub上的特定範例中找到此策略的完整範例，其中可以使用數據生成代碼以及Keras腳本。



參考
https://stanford.edu/~shervine/blog/keras-how-to-generate-data-on-the-fly?fbclid=IwAR30U2ngm-7XyotCV-1nzTMbekusR8TNUkq-Wi14Sv5blPjcbN342se3TH4