Helve’s Python memo

Pythonを使った機械学習やデータ分析の備忘録

<Chainer> Iteratorクラスによる学習用ミニバッチ作成

以前の記事で、ChainerのChainクラスとOptimizerを使って最小限のニューラルネットワーク (NN) を実装した(以下のリンクを参照)。
今回は、データセットから学習用ミニバッチを作成してくれるIteratorクラスの動作を確認する。

helve-python.hatenablog.jp

目次

環境

Spyder 3.2.8
Python 3.6.5
NumPy 1.14.3
Chainer 4.2.0

以下では、各ライブラリを以下のようにインポートしていることを前提とする。

import numpy as np
import matplotlib.pyplot as plt
import chainer
import chainer.links as L
import chainer.functions as F
from chainer import optimizers, Chain, dataset, datasets, iterators
import matplotlib.pyplot as plt

Iteratorクラス

Iteratorクラスは、データセットの説明変数と目的変数を束ねたミニバッチ作成や、データ順序のシャッフルを行ってくれるクラスである。
Chainer 4.2.0では、以下3つのIteratorが利用可能である。

SerialIterator 最も単純なIterator
MultiprocessIterator multiprocessingモジュールにより並列化
MultithreadIterator threadingモジュールにより並列化

本記事ではSerialIteratorクラスを扱う。

chainer.iterators.SerialIterator(dataset, batch_size, 
                                 repeat=True, shuffle=True)

パラメータの説明は以下の通り。

dataset:
反復させるデータセット

batch_size: int型
各ミニバッチのサンプル数。

repeat: bool型
Trueの場合、無限に反復させる。
Falseの場合、全てのデータ(1エポック分)を返すと反復を終える。

shuffle: bool型
Trueの場合、ランダムな順序でデータを返す。
Falseの場合、datasetと同じ順序で返す。


SerialIteratorの動作を確認する。

スクリプト

np.random.seed(0)

data = np.arange(3)
data = chainer.Variable(data)
ite  = chainer.iterators.SerialIterator(data, batch_size=2, 
                                        repeat=True, shuffle=True)

for i in range(5):
    print(ite.next())

上のスクリプトでは、データの順序を固定するため、np.random.seed(0)としてNumPyの乱数シードを設定している。
0, 1, 2の3つの数字からなる配列をVariableオブジェクトとして、SerialIteratorにセットする。
バッチ数は2として、データを返す順序はランダムとする。

SerialIteratorオブジェクトitenext()メソッドを実行すると、バッチ数だけデータが返される。

実行結果

[variable(2), variable(1)]
[variable(0), variable(0)]
[variable(2), variable(1)]
[variable(0), variable(1)]
[variable(2), variable(2)]

実行結果を見ると、Variableオブジェクトがbatch_size数だけリスト形式で返されている。
データ(0, 1, 2)を一通り返すと、同じリストであっても継続してデータを返している。

TupleDatasetによるデータセット作成

Iteratorクラスにセットするデータセットを用意する場合、TupleDatasetを用いると便利である。

chainer.datasets.TupleDataset(*datasets)

datasetsは任意の数のデータ変数である。ただし、データは全て同じ長さ(行数)でなければならない。

TupleDatasetの主なメソッドは次の2つ。

__getitem__(index):
indexで指定した要素をタプルで返す。

__len__:
データの長さ(行数)を返す。

例えば、2つのNumPy配列d1, d2を引数として、次のようにTupleDatasetオブジェクトを作る。

d1 = np.array([[1, 2],
               [3, 4],
               [5, 6]], dtype=np.float32)
d2 = np.array([[7],
               [9],
               [8]], dtype=np.float32)

dataset = datasets.TupleDataset(d1, d2)

print(dataset.__len__())
for i in range(3):
	print(dataset.__getitem__(i))

実行結果

3
(array([1., 2.], dtype=float32), array([7.], dtype=float32))
(array([3., 4.], dtype=float32), array([9.], dtype=float32))
(array([5., 6.], dtype=float32), array([8.], dtype=float32))

データの長さは3, また、__getitem__メソッドにより、d1, d2の要素が1行ずつ取り出されている。

次に、TupleDatasetオブジェクトをSerialIteratorにセットしてみる。バッチ数は2とする。

d1 = np.array([[1, 2],
               [3, 4],
               [5, 6]], dtype=np.float32)
d2 = np.array([[7],
               [9],
               [8]], dtype=np.float32)

data_set = datasets.TupleDataset(d1, d2)

ite = iterators.SerialIterator(data_set, batch_size=2, 
                               repeat=True, shuffle=True)

for i in range(5):
    print(ite.next())

実行結果

[(array([5., 6.], dtype=float32), array([8.], dtype=float32)), (array([1., 2.], dtype=float32), array([7.], dtype=float32))]
[(array([3., 4.], dtype=float32), array([9.], dtype=float32)), (array([5., 6.], dtype=float32), array([8.], dtype=float32))]
[(array([1., 2.], dtype=float32), array([7.], dtype=float32)), (array([3., 4.], dtype=float32), array([9.], dtype=float32))]
[(array([3., 4.], dtype=float32), array([9.], dtype=float32)), (array([5., 6.], dtype=float32), array([8.], dtype=float32))]
[(array([1., 2.], dtype=float32), array([7.], dtype=float32)), (array([3., 4.], dtype=float32), array([9.], dtype=float32))]

このように、各ite.next()毎に、d1d2の各行が、ネストされたタプルで得られる。
ite.next()の結果をd1, d2毎にまとめるためには、chainer.dataset.concat_examples()を用いる。
(評価関数にデータを与えるときに必要)

dataset.concat_examples(ite.next())

実行結果

(array([[5., 6.],
        [3., 4.]], dtype=float32), 
 array([[8.],
        [9.]], dtype=float32))

実装例

Iteratorを使った学習例を示す。
前回の記事のスクリプトを改良した。

学習するデータの順序をランダムに入れ替える以外、前回と同じ学習条件である。
線形モデルy=x-5を学習させる。
説明変数xは0~9.9まで0.1刻みで100点とする。
また、教師データの目的変数には、平均0, 分散0.1の正規分布に従う信号をノイズとして加えた。
この教師データを100回反復させて学習させる(エポック数100)。
また、教師データを2点ずつ与えるバッチ処理を行う。

import numpy as np
import chainer
import chainer.links as L
import chainer.functions as F
from chainer import optimizers, Chain, dataset, datasets
import matplotlib.pyplot as plt

class MyChain(Chain):
    def __init__(self):
        super(MyChain, self).__init__()
        with self.init_scope():
            self.l1 = L.Linear(None, 3)
            self.l2 = L.Linear(3, 1)

    def __call__(self, x):
        h = F.relu(self.l1(x))
        return self.l2(h)

def lossfun(train_y, pred_y):
    loss = F.mean_squared_error(train_y, pred_y)
    return loss
    

np.random.seed(0)
batch_size = 2    # バッチ数
n_epoch = 100     # エポック数

model = MyChain()
optimizer = optimizers.SGD().setup(model)

x_data = np.arange(0, 10, 0.1, dtype=np.float32).reshape(-1,1) # 説明変数
y_data = x_data-5 + 0.1*np.random.randn(len(x_data)).reshape(-1,1)
y_data = y_data.astype(np.float32) # 目的変数

data_set = datasets.TupleDataset(x_data, y_data)
ite = chainer.iterators.SerialIterator(data_set, batch_size=batch_size, 
                                       repeat=True, shuffle=True)

pred_y_before = model(x_data) # 学習前の予測値

mean_err = [] # 平均予測誤差
for epoch in range(n_epoch):
    err_temp = 0
    
    for i in range(0, x_data.shape[0], batch_size):
        batch = ite.next()
        train_x, train_y = dataset.concat_examples(batch)
        
        pred_y  = model(train_x)
        optimizer.update(lossfun, train_y, pred_y)
        
        err_temp += F.mean_squared_error(train_y, pred_y).data*batch_size
    
    mean_err += [err_temp/x_data.shape[0]]
    
pred_y_after = model(x_data)      # 学習後の予測値


fig, ax = plt.subplots()
ax.plot(x_data, y_data, label="True values")
ax.plot(x_data, pred_y_before.data, label="Before learning")
ax.plot(x_data, pred_y_after.data, label="After learning")
ax.set_xlabel("x")
ax.set_ylabel("y")
ax.legend()
ax.grid()
plt.show()

fig, ax = plt.subplots()
ax.plot(mean_err)
ax.set_xlabel("Epoch")
ax.set_ylabel("Mean squared error")
ax.grid()
plt.show()

学習前後のモデルの予測値は次のようになった(青が教師データ、黄色が学習前の予測値、緑が学習後の予測値)。
学習によって、予測値が教師データに近づいている。
f:id:Helve:20180722155633p:plain

また、エポックごとの予測の二乗平均誤差は以下の通り。
データをシャッフルしなかった前回と比べ、誤差の減少が速くなっている。
f:id:Helve:20180722155642p:plain

参考リンク

https://docs.chainer.org/en/stable/reference/generated/chainer.iterators.SerialIterator.html#chainer.iterators.SerialIterator

この記事ではChainerの実装に焦点を絞っているが、実際にニューラルネットワークの内部で行われている計算を知りたいときには、以下の本が参考になる。