【Python/Tensorflow】TFRecordsから学習を行う方法

IT関連
2020.11.06

前回の記事の続きに相当する内容です。

【Python/Tensorflow】 TFRecordsとは何ぞや？
TensorFlow(Keras)で大規模データを扱っていると、学習開始前のデータの読み出し(転送)でめっちゃ時間がかかることがあります。普通はGeneratorを利用して解決するのですが、AWSの環境などではそうもいかない場合があります。本記事は、学習におけるI/Oボトルネックを解消する際に用いられるTFRecordについて、自分なりに調べた内容をまとめます。
novnote.com
2020.11.06

TFRecordsとは何か？という部分の説明を行い、実際にファイルとして書き出すところ・書き出した内容を読み出してTensorとして復元するというところまでは解説しました。

ですが、そこから学習を行う場合にはどうするのか？とか、いくつかのデータをひとまとめにしてTFRecordsとして保存するのはどうするのか？といった内容には触れていませんでした。そこで、本記事では複数データのTFRecordsへの書き出し・書き出したファイルからの読み出し、実際に学習を行う際にはどうするのかというのを簡単な多層パーセプトロン(MLP)に適用して確認してみたいと思います。サンプルコードを載せているので、コピペすれば動作確認することが可能です。

本記事の対象読者は以下の通りです。

  • 前回の記事を見て「で、結局学習のコードとはどう繋がるの？」ってなった人
  • 複数のデータを保存できるらしいけど、その読み出しはどうするの？となった人
  • 過去の自分(備忘録的側面もあるので)

今回の記事でもいくつかサンプルコードを用意しますが、実行環境は以下になります。

  • OS：Windows10
  • Pythonのバージョン：3.7.9
  • tensorflowのバージョン：2.1.0

では早速本題に入っていきましょう。

TFRecordsに複数データをまとめて記録する方法

結論から言えば、tf.io.TFRecordWriterを記述したブロックでfor文を回し、データを逐次的に書き込めばOKです。シリアライズされたデータを書き込んでいる上、protocol buffersとして自分で決めた型でデータを書き込んでいくので、順次繋げてデータを書き込んでいっても問題ないということですね。

さて、では早速サンプルコードです。

import numpy as np
import tensorflow as tf

# リストをFloat列のFeatureに変換する関数
def feature_float_list(l):
    return tf.train.Feature(float_list=tf.train.FloatList(value=l))

# 訓練データのnumpy配列と正解データをひとまとめにして、Exampleに詰め込む関数
def array2example(train, label):
    feature = {
        "train": feature_float_list(train.reshape(-1).astype("float32")),
        "label": feature_float_list(label.reshape(-1).astype("float32"))
    }
    return tf.train.Example(features=tf.train.Features(feature=feature))

# メイン
if __name__ == "__main__":
    train_list = [np.random.randn(2, 4) for i in range(10)]
    label_list = np.random.rand(10)

    filepath = "./test_sequence.tfrecords"
    # TFRecords形式に書き出し ループで後ろに追加していく処理
    with tf.io.TFRecordWriter(filepath) as writer:
        for train, label in zip(train_list, label_list):
            example_proto = array2example(train, label)
            writer.write(example_proto.SerializeToString())

訓練データと正解データを用意し、それをProtocol Buffersの形式に沿って構造化し、シリアライズしてTFRecordsファイルに保存しています。今回は、ランダムな値が入っている4×2の配列を10個用意し、それを訓練データ、それに対応づく適当なスカラー値を正解データとして、”train”、”label”のkeyを付けて格納しています。

注意点としては、入力データのシリアライズのために、numpy配列を1次元データに変換しないといけないところです。上記のコードでは”reshape(-1)”を指定しているところですね。

ここまででTFRecordsに複数のデータを保存できたので、次はその読み出しについて見てみましょう。

TFRecords形式で保存した一連のデータを読み出す方法

では、次に先ほど作成したTFRecords形式のデータから、元のTensorを取得する方法について説明します。まずはサンプルコードです。

import numpy as np
import tensorflow as tf
import pprint

# 特徴量の次元(保存したデータによって当然変わる)
feature_dim = 4*2

# TFRecords読み出しの際に、デシリアライズを行うコールバック関数
def parse_example(example):
    # デシリアライズのためのデータ構造定義
    features = {
        "train": tf.io.FixedLenFeature([feature_dim], dtype=tf.float32),
        "label": tf.io.FixedLenFeature([], dtype=tf.float32)
    }
    # デシリアライズ実行
    train_data = tf.io.parse_single_example(example, features=features)
    x = train_data["train"]
    y = train_data["label"]
    return x, y

# データの保存先
filepath = "./test_sequence.tfrecords"

# TFRecordsからのデータ読み出し & デシリアライズ
dataset_train = tf.data.TFRecordDataset([filepath]).map(parse_example)

# 取得結果のチェック
for train_X, train_y in dataset_train:
    pprint.pprint(train_X)
    pprint.pprint(train_y)

"""
【出力結果】
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([ 0.1470189 ,  0.31900916,  0.45194325, -0.72674596,  0.5248739 ,
        1.2797657 ,  1.1793714 ,  0.383066  ], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.32254368>
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([-1.1852002 ,  0.9668531 , -2.449051  ,  0.83119386,  1.2894857 ,
        0.23572999, -0.10557662, -0.74806213], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.31286517>
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([ 1.610198  ,  0.7453617 , -0.1624388 ,  0.7844823 ,  0.9396648 ,
       -0.6712337 , -0.4645068 ,  0.46319363], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.5950493>
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([ 0.38367644, -1.0082129 ,  0.52089405,  0.60254085,  0.55493486,
        2.1865103 ,  0.9362942 , -1.7111706 ], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.1489663>
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([-0.28963235, -0.87757814,  1.0456872 , -0.688191  , -0.8707142 ,
        0.7117175 , -1.0086906 ,  0.22210364], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.64152765>
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([ 1.4525561 ,  0.1256987 ,  0.03918342,  0.8434101 ,  0.30440658,
       -0.96848816, -1.4410765 , -0.9340212 ], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.77477044>
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([ 0.5012486 ,  2.1488936 , -0.78319025, -0.57885784, -0.7437581 ,
        1.484867  ,  0.49208012,  0.0136353 ], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.15695865>
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([-0.23769051,  1.4338529 , -0.30765015,  0.8680623 , -1.6750941 ,
       -1.1952772 ,  1.0512254 , -0.13628246], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.41800562>
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([-1.0444406 ,  0.9778109 , -0.41180775,  1.5427221 ,  0.22119844,
        0.64852273,  0.5101197 , -1.8319397 ], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.8340166>
<tf.Tensor: shape=(8,), dtype=float32, numpy=
array([1.2629781 , 0.5410737 , 1.1205019 , 0.08506348, 0.95039374,
       1.5243149 , 0.58544624, 0.39725098], dtype=float32)>
<tf.Tensor: shape=(), dtype=float32, numpy=0.9453445>
"""

ここでやっていることは、以下の通りです。

  • TFRecords形式でシリアライズされたデータを解析する関数を用意(parse_example)
  • tf.data.TFReocrdDataset()を用いてTFRecordsデータを読み込み
  • 定義したparse_exampleを使用して、読み込んだデータをデシリアライズ

デシリアライズしたデータは、parse_exampleで指定した型の分だけ順次読み出すことができます。順次読み出しているのがfor文で書かれているところになります。

また、tf.data.TFReocrdDataset()で読み出した結果は、tf.data.DatasetというTensorflowが推奨するデータ形式になっています。今回の学習では、この形式でのデータの流し方についても触れます。

tf.data.Dataset  |  TensorFlow v2.16.1
Represents a potentially large set of elements.
www.tensorflow.org

TFRecordsから読みだしたデータで学習を行ってみる

いよいよ本題の学習の部分です。例によって、サンプルコードを最初に載せます。

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense
from tensorflow.keras.optimizers import RMSprop

# 特徴量の次元(保存したデータによって当然変わる)
feature_dim = 4*2

# TFRecords読み出しの際に、デシリアライズを行うコールバック関数
def parse_example(example):
    features = {
        "train": tf.io.FixedLenFeature([feature_dim], dtype=tf.float32),
        "label": tf.io.FixedLenFeature([], dtype=tf.float32)
    }
    train_data = tf.io.parse_single_example(example, features=features)
    x = train_data["train"]
    y = train_data["label"]
    return x, y

# データの保存先
filepath = "./test_sequence.tfrecords"

# TFRecordsからのデータ読み出し & デシリアライズ
dataset_train = tf.data.TFRecordDataset([filepath]) \ 
                .map(parse_example) \ # デシリアライズ
                .batch(1) \ #バッチサイズ指定(今回は1としているので、ミニバッチなしと同義)
                .repeat(3)  #epoch数を指定 ⇒ -1 を指定すれば何度でもループできる。それでもOK

# keras API を利用して、モデル構築(MLP) & 学習
inputs = Input(shape=(feature_dim,))
X = Dense(16, activation="relu")(inputs)
X = Dense(4, activation="relu")(X)
X = Dense(1, activation="relu")(X)

model = Model(inputs, X)
print(model.summary())

model.compile(
    loss="MSE",
    optimizer=RMSprop()
)

# 学習実行
model.fit(
    x=dataset_train, # 直接dataset_trainを突っ込んでOK
    epochs=3,
    verbose=1,
    steps_per_epoch=10
)
"""
【出力結果】
Model: "model"
_________________________________________________________________
Layer (type)                 Output Shape              Param #
=================================================================
input_1 (InputLayer)         [(None, 8)]               0
_________________________________________________________________
dense (Dense)                (None, 16)                144
_________________________________________________________________
dense_1 (Dense)              (None, 4)                 68
_________________________________________________________________
dense_2 (Dense)              (None, 1)                 5
=================================================================
Total params: 217
Trainable params: 217
Non-trainable params: 0
_________________________________________________________________
None
Train for 10 steps
Epoch 1/3
10/10 [==============================] - 0s 49ms/step - loss: 0.3281
Epoch 2/3
10/10 [==============================] - 0s 2ms/step - loss: 0.3285
Epoch 3/3
10/10 [==============================] - 0s 2ms/step - loss: 0.3281
"""

学習を動かすにはどうするのか、という部分が主眼なので、モデルは適当です。一応、隠れ層のunit数が16と4、出力層のunit数が1つの4層パーセプトロンとしました。

基本的に、tf.data.TFRecordDataset()で読み出したデータはそのままModel.fitメソッドに突っ込んでやればOKです。要するに、tf.data.Datasetの形式のデータをModel.fitで直接受けることができるということですね。なお、これはtensorflow2系の話で、それ以前は自分でiteratorを設定する必要があったようです。

また、tf.data.TFRecordDataset()の部分で、少し変更(追加で使用しているメソッド)があります。

  • .batch(batch_size)というメソッドで、読み出す際のバッチサイズを規定している
  • .repeat(epoch)というメソッドで、何回データ読み出しを繰り返すか指定している

batchメソッドについては、指定しないとValueErrorが起きて動作しません。また、repeatについては適切に設定していないと、途中で学習に失敗します。ただし、「-1」を設定すると無限リピートになるので失敗はしなくなります(ただし、Model.fitメソッドで指定するsteps_per_epochの数が適切でないと、epoch内で複数回同じデータで学習することになる点は注意です)。

以上が、TFReocrds形式のデータを読み出して学習を行う際のサンプルコードと注意点です。

まとめ

今回はTFRecords形式のデータで、複数の訓練データを保存する方法・それを読み出す方法・読み出したデータを使って学習を行う方法についてサンプルコードを交えて解説しました。

動作がイメージできる最小構成を目指したので、正直サンプルコードはしょぼいです。しかし、基本となるエッセンスが詰まっているので、これをベースに自分のやりたいようにデータを保存し、学習に用いることができるはずです。

次回はもともとやりたかった、SparseTensor・SparseFeatureを交えたTFRcordsによるシリアライズを取り扱おうと思います。疎行列によるデータの保存・読み出しが、シリアライズしたデータでできるようになると、データの性質によってはストレージを圧迫せずに済むので実用的かと思います。

今回は以上となります。最後までお読みいただきありがとうございました！

スポンサーリンク

コメント

タイトルとURLをコピーしました