杂项 如何在Keras中使用数据生成器(翻译)

[Syize]

文章类型

完全原创 —— 转载请标注作者

翻译自斯坦福大学网站上的一篇文章，A detailed example of how to use data generators with Keras。主要介绍了在面对大量数据时，如何通过编写数据生成器(DataGenerator)来实现动态加载数据的功能。

目的​

你是否有过由于训练数据量过大导致内存消耗太多，因此想找一个魔法般的功能来解决这个问题并且不影响已有的代码？在这个大数据的时代，"大数据"变得越来越重要。

有时候即使你拥有顶配的设备，设备的内存可能也无法存放下所有的训练数据。这个时候，我们就需要寻找其他的解决方法。本篇文章将展示如何借助 Python 中高度封装的深度学习库Keras实现实时并发的数据生成功能，并将该功能嵌入到你的深度学习模型中。

前言​

在阅读这篇文章之前，也许你的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)

这篇文章所涉及到的所有更改都是关于修改第5行加载数据的代码的。可以看出，原始代码存在的问题是有可能无法将所有训练使用的样本一次性全部加载进内存中。

为了达到最终的目的，让我们一步一步来介绍如何搭建一个数据生成器。顺便一提，后面的代码都被尽量编写的结构清晰明了，也许很适合用在你自己的项目中。你可以直接复制粘贴，并自己补全其中省略的部分。

声明​

在开始之前，让我们做一些声明以防止在后文中出现有歧义的地方。

假定后面所说的ID代表了Python中的一个字符串，该字符串对应了一个样本。并且为了清晰的分辨不同的样本以及它们的标签，数据被以下列的方式存储：

1. 一个名叫partition的字典，其中包括
   • partition['train']：存储训练集及样本对应的ID
   • partition['validation']：存储测试集样本对应的ID
2. 一个名叫labels的字典，ID作为其中的key，ID对应的样本的标签作为value，可以通过labels[ID]的形式获取到样本对应的标签。

举个例子，假设我们的训练集有三个样本，对应的ID分别是'id-1', 'id-2', 'id-3'，对应的标签分别是0, 1, 2；测试集有一个样本，ID为id-4，标签为1。这种情况下，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代码和自定义类会写在不同的Python文件中，文件树如下

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

其中data/是存放数据集的目录。

最后的最后，本文中的代码编写尽量保证了泛用性和最简化，因此你应该可以轻易的将其更改以适用于你自己的数据集。

Data generator​

现在，我们真正开始编写Python类DataGenerator，并将其用于Keras模型中来实时提供训练数据。

首先我们需要编写类的初始化函数，我们将DataGenerator设计为keras.utils.Sequence的子类以便可以使用其某些便捷的特性例如多进程。

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()

初始化函数接受很多重要的参数，例如数据的ID(list_IDs)和对应的标签labels，数据的维度大小dim(例如三维、大小是32×32×32的数据为dims=(32, 32))，数据的通道数n_channels(处理图像数据等多通道数据时很有用)，数据的种类数n_classes，训练使用的数据块(batch)的大小batch_size，以及生成数据时是否打乱数据shuffle。

最后调用的方法on_epoch_end会在类初始化以及每次训练周期(epoch)结束时调用，用来生成取数据ID使用的索引。如果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)

打乱索引是为了在每次训练周期向模型喂数据时保证喂的数据是不那么相似的，能够保证最后训练出来的模型更加健壮。

另外一个关键的函数是__data_generation，这个函数的工作便是生成每次训练使用的batch，它接受一个参数list_IDs_temp，这个列表包含了用于生成batch的样本的ID。

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)

这个函数会实时的读取对应样本的npy文件。由于我们的代码对多进程友好，因此你可以在这里做更多更复杂的工作而无需担心数据的生成成为训练过程中的瓶颈(感觉瓶颈变成磁盘IO了)。

这里我们还使用了Keras的keras.utils.to_categorical函数来把存储在y中的数字标签转化为了适用于分类的二进制的形式(举个例子，对于6分类的问题，标签3对应的形式为001000)。

下面我们将上面提到的三个部分整合到一起我们首先需要实现获取训练时当前batch的索引的功能(即这是所有batch中的第几个batch)。这个功能我们通过魔术方法__len__实现。

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

通常这个长度的计算公式如下

这样在每次训练周期模型最多遍历所有batch一次。

当模型利用index取对应的batch时，DataGenerator需要可以返回对应的batch，这个功能通过魔术方法__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方法，该方法可以接受我们构造的生成器，接下来的事情Keras会帮我们搞定！

另外需要注意的是，我们将use_multiprocessing设置为True来启用了多进程的特性，使用的进程数在workers那里进行了指定。如果你的workers设置的足够高，那么你的CPU就会努力的工作，这样训练的瓶颈就只会在GPU对神经网络的误差进行传播的过程中(而不是数据的生成)。(还有磁盘IO啦)

结论​

没了，就这些。现在运行你的Keras脚本，你就会发现你的CPU和GPU同时在工作。你还可以到GitHub上看一看将此生成器应用到具体个例上的代码，data generation和Keras脚本都在哟。

本文章搬运自 Syizeのblog