GitHub Repository: tensorflow/docs-l10n
Path: blob/master/site/ja/guide/data.ipynb
²⁵¹¹⁵ views

Kernel: Python 3

Copyright 2018 The TensorFlow Authors.

Licensed under the Apache License, Version 2.0 (the "License");

In [ ]:

#@title Licensed under the Apache License, Version 2.0 (the "License"); { display-mode: "form" }
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

tf.data: TensorFlow 入力パイプラインの構築

tf.data API を使用すると、単純で再利用可能なピースから複雑な入力パイプラインを構築することができます。たとえば、画像モデルのパイプラインでは、分散ファイルシステムのファイルからデータを集め、各画像にランダムな摂動を適用し、ランダムに選択された画像を訓練用のバッチとして統合することがあります。また、テキストモデルのパイプラインでは、未加工のテキストデータからシンボルを抽出し、そのシンボルをルックアップテーブルとともに埋め込み識別子に変換し、異なる長さのシーケンスをまとめてバッチ処理することもあります。tf.data API は、大量のデータを処理し、別のデータ形式から読み取り、こういった複雑な変換の実行を可能にします。

tf.data API は、要素のシーケンスを表す tf.data.Dataset 抽象を導入します。シーケンス内の各要素は 1 つ以上のコンポーネントで構成されています。たとえば、画像パイプラインの場合、要素は画像とそのラベルを表す一組のテンソルコンポーネントを持つ単一のトレーニングサンプルである場合があります。

データセットを作成するには、次の 2 つの方法があります。

データソースによって、メモリまたは 1 つ以上のファイルに格納されたデータから Dataset を作成する。
データ変換によって、1 つ以上の tf.data.Dataset オブジェクトからデータセットを作成する。

In [ ]:

import tensorflow as tf

In [ ]:

import pathlib
import os
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np

np.set_printoptions(precision=4)

基本的な仕組み

入力パイプラインを作成するには、データソースから着手する必要があります。たとえば、メモリ内のデータから Dataset を作成する場合、tf.data.Dataset.from_tensors() または tf.data.Dataset.from_tensor_slices() を使用できます。推奨される TFRecord 形式で入力データがファイル内に格納されている場合は、tf.data.TFRecordDataset() を使用することもできます。

Dataset オブジェクトを作成したら、tf.data.Dataset オブジェクトに対してチェーンメソッド呼び出しを行い、新しい Dataset オブジェクトに変換することができます。たとえば、Dataset.map などの要素ごとの変換を適用したり、Dataset.batch() などの複数の要素に対する変換を適用したりすることができます。変換の全リストについては、tf.data.Dataset のドキュメントをご覧ください。

Dataset オブジェクトは Python イテラブルです。そのため、for ループを使って、その要素を消費することができます。

In [ ]:

dataset = tf.data.Dataset.from_tensor_slices([8, 3, 0, 8, 2, 1])
dataset

In [ ]:

for elem in dataset:
  print(elem.numpy())

または、iter を使って Python イテレータを明示的に作成し、next を使って要素を消費することもできます。

In [ ]:

it = iter(dataset)

print(next(it).numpy())

また、reduce 変換を使ってデータセットの要素を消費することもできます。この場合、単一の結果を得るためにすべての要素が減らされます。次の例では、整数のデータセットの合計を計算するために reduce 変換を使用する方法を説明しています。

In [ ]:

print(dataset.reduce(0, lambda state, value: state + value).numpy())

データセットの構造

データセットは、要素ごとに同じ（ネスト）構造のコンポーネントを持つ一連の要素を生成し、構造の各コンポーネントは tf.TypeSpec で表現可能な、tf.Tensor、tf.sparse.SparseTensor、tf.RaggedTensor、tf.TensorArray、tf.data.Dataset など、あらゆる型を持つことができます。

要素の（ネスト）構造を表現するために使用される Python コンストラクトには、tuple、dict、NamedTuple、および OrderedDict があります。特に list は、データセット要素の構造を表現するコンストラクトとして有効ではありません。これは、初期の tf.data ユーザーに、list 入力（code7}tf.data.Dataset.from_tensors に渡される入力など）がテンソルとして自動的に圧縮されることと、list 出力（ユーザー定義関数の戻り値など）が tuple に強制されることを強く希望していたためです。その結果、list 入力を構造として扱う場合には tuple に変換し、単一コンポーネントの list 出力が必要な場合は tf.stack を使って明示的に圧縮する必要があります。

各要素コンポーネントの型を検査するには、Dataset.element_spec プロパティを使用できます。このプロパティは tf.TypeSpec オブジェクトのネストされた構造を、要素の構造を一致させて返します。これは、単一のコンポーネント、コンポーネントのタプル、またはコンポーネントのネストされたタプルである場合があります。以下に例を示します。

In [ ]:

dataset1 = tf.data.Dataset.from_tensor_slices(tf.random.uniform([4, 10]))

dataset1.element_spec

In [ ]:

dataset2 = tf.data.Dataset.from_tensor_slices(
   (tf.random.uniform([4]),
    tf.random.uniform([4, 100], maxval=100, dtype=tf.int32)))

dataset2.element_spec

In [ ]:

dataset3 = tf.data.Dataset.zip((dataset1, dataset2))

dataset3.element_spec

In [ ]:

# Dataset containing a sparse tensor.
dataset4 = tf.data.Dataset.from_tensors(tf.SparseTensor(indices=[[0, 0], [1, 2]], values=[1, 2], dense_shape=[3, 4]))

dataset4.element_spec

In [ ]:

# Use value_type to see the type of value represented by the element spec
dataset4.element_spec.value_type

Dataset 変換では、あらゆる構造のデータセットがサポートされています。Dataset.map を使用し、各要素に関数を適用する Dataset.filter 変換を適用すると、要素の構造によって関数の引数が判定されます。

In [ ]:

dataset1 = tf.data.Dataset.from_tensor_slices(
    tf.random.uniform([4, 10], minval=1, maxval=10, dtype=tf.int32))

dataset1

In [ ]:

for z in dataset1:
  print(z.numpy())

In [ ]:

dataset2 = tf.data.Dataset.from_tensor_slices(
   (tf.random.uniform([4]),
    tf.random.uniform([4, 100], maxval=100, dtype=tf.int32)))

dataset2

In [ ]:

dataset3 = tf.data.Dataset.zip((dataset1, dataset2))

dataset3

In [ ]:

for a, (b,c) in dataset3:
  print('shapes: {a.shape}, {b.shape}, {c.shape}'.format(a=a, b=b, c=c))

入力データの読み取り

NumPy 配列を消費する

その他の例については、NumPy 配列を読み込むチュートリアルをご覧ください。

すべての入力データがメモリに収容できる場合、このデータから Dataset を作成するには、データを tf.Tensor オブジェクトに変換してから Dataset.from_tensor_slices を使用するのが最も簡単な方法です。

In [ ]:

train, test = tf.keras.datasets.fashion_mnist.load_data()

In [ ]:

images, labels = train
images = images/255

dataset = tf.data.Dataset.from_tensor_slices((images, labels))
dataset

注意: 上記のコードスニペットは、features 配列と labels 配列を tf.constant() 演算として TesorFlow グラフに埋め込みます。これは小さなデータセットではうまく機能しますが、配列の内容が何度もコピーされるため、メモリを浪費してしまい、tf.GraphDef プロトコルバッファの 2GB 制限に達してしまう可能性があります。

Python ジェネレータの消費

tf.data.Dataset として簡単に統合できるもう 1 つの一般的なデータソースは、Python ジェネレータです。

注意: 便利なアプローチではあるものの、これには移植性と拡張性の制限があります。ジェネレータを作成した Python プロセスで実行する必要があり、それでも、Python GIL の制約を受けます。

In [ ]:

def count(stop):
  i = 0
  while i<stop:
    yield i
    i += 1

In [ ]:

for n in count(5):
  print(n)

Dataset.from_generator コンストラクタは Python ジェネレータを完全に機能する tf.data.Dataset に変換します。

このコンストラクタは、イテレータではなく、呼び出し可能オブジェクトを入力として取ります。このため、ジェネレータが最後に達するとそれを再開することができます。オプションで args 引数を取ることができ、呼び出し可能オブジェクトの引数として渡されます。

tf.data は内部で tf.Graph を構築するため output_types 引数は必須であり、グラフのエッジには、tf.dtype が必要です。

In [ ]:

ds_counter = tf.data.Dataset.from_generator(count, args=[25], output_types=tf.int32, output_shapes = (), )

In [ ]:

for count_batch in ds_counter.repeat().batch(10).take(10):
  print(count_batch.numpy())

output_shapes 引数は必須ではありませんが、多数の TensorFlow 演算では、不明な階数のテンソルはサポートされていないため、強く推奨されます。特定の軸の長さが不明または可変である場合は、output_shapes で None として設定してください。

また、output_shapes と output_types は、ほかのデータセットのメソッドと同じネスト規則に従うことに注意することが重要です。

以下は、この両方の側面を示すジェネレーターの例です。配列のタプルを返し、2 つ目の配列は、不明な長さを持つベクトルです。

In [ ]:

def gen_series():
  i = 0
  while True:
    size = np.random.randint(0, 10)
    yield i, np.random.normal(size=(size,))
    i += 1

In [ ]:

for i, series in gen_series():
  print(i, ":", str(series))
  if i > 5:
    break

最初の出力は int32 で、2 つ目は float32 です。

最初の項目はスカラーの形状 () で、2 つ目は長さが不明なベクトルの形状 (None,) です。

In [ ]:

ds_series = tf.data.Dataset.from_generator(
    gen_series, 
    output_types=(tf.int32, tf.float32), 
    output_shapes=((), (None,)))

ds_series

これで、通常の tf.data.Dataset のように使用できるようになりました。可変形状を持つデータセットをバッチ処理する場合は、Dataset.padded_batch を使用する必要があります。

In [ ]:

ds_series_batch = ds_series.shuffle(20).padded_batch(10)

ids, sequence_batch = next(iter(ds_series_batch))
print(ids.numpy())
print()
print(sequence_batch.numpy())

より現実的な例として、preprocessing.image.ImageDataGenerator を tf.data.Dataset としてラップしてみましょう。

まず、データをダウンロードします。

In [ ]:

flowers = tf.keras.utils.get_file(
    'flower_photos',
    'https://storage.googleapis.com/download.tensorflow.org/example_images/flower_photos.tgz',
    untar=True)

image.ImageDataGenerator を作成します。

In [ ]:

img_gen = tf.keras.preprocessing.image.ImageDataGenerator(rescale=1./255, rotation_range=20)

In [ ]:

images, labels = next(img_gen.flow_from_directory(flowers))

In [ ]:

print(images.dtype, images.shape)
print(labels.dtype, labels.shape)

In [ ]:

ds = tf.data.Dataset.from_generator(
    lambda: img_gen.flow_from_directory(flowers), 
    output_types=(tf.float32, tf.float32), 
    output_shapes=([32,256,256,3], [32,5])
)

ds.element_spec

In [ ]:

for images, labels in ds.take(1):
  print('images.shape: ', images.shape)
  print('labels.shape: ', labels.shape)

TFRecord データの消費

エンドツーエンドの例については、TFRcords を読み込むチュートリアルをご覧ください。

tf.data API では多様なファイル形式がサポートされているため、メモリに収まらない大規模なデータセットを処理することができます。たとえば、TFRecord ファイル形式は、単純なレコード指向のバイナリー形式で、多くの TensorFlow アプリケーションでデータの訓練に使用されています。tf.data.TFRecordDataset クラスでは、入力パイプラインの一部として 1 つ以上の TFRecord ファイルの内容をストリーミングすることができます。

以下の例では、French Street Name Signs（FSNS）から取得したテストファイルを使用しています。

In [ ]:

# Creates a dataset that reads all of the examples from two files.
fsns_test_file = tf.keras.utils.get_file("fsns.tfrec", "https://storage.googleapis.com/download.tensorflow.org/data/fsns-20160927/testdata/fsns-00000-of-00001")

TFRecordDataset イニシャライザの filenames 引数は、文字列、文字列のリスト、または文字列の tf.Tensor のいずれかです。そのため、訓練と検証プロセスに使用するファイルが 2 セットある場合、入力引数として filenames を取って、データセットを生成するファクトリーメソッドを作成することができます。

In [ ]:

dataset = tf.data.TFRecordDataset(filenames = [fsns_test_file])
dataset

多数の TensorFlow プロジェクトは、TFRecord ファイルでシリアル化された tf.train.Example レコードを使用します。これらのレコードは、検査前にデコードされている必要があります。

In [ ]:

raw_example = next(iter(dataset))
parsed = tf.train.Example.FromString(raw_example.numpy())

parsed.features.feature['image/text']

テキストデータの消費

エンドツーエンドの例については、テキストを読み込むチュートリアルをご覧ください。

多くのデータセットは 1 つ以上のテキストファイルとして分散されていますが、tf.data.TextLineDataset を使用してそれらのファイルから簡単に行を抽出することができます。1 つ以上のファイル名を指定すると、TextLineDataset によって、ファイルの行ごとに 1 つの文字列値の要素が生成されます。

In [ ]:

directory_url = 'https://storage.googleapis.com/download.tensorflow.org/data/illiad/'
file_names = ['cowper.txt', 'derby.txt', 'butler.txt']

file_paths = [
    tf.keras.utils.get_file(file_name, directory_url + file_name)
    for file_name in file_names
]

In [ ]:

dataset = tf.data.TextLineDataset(file_paths)

以下は、最初のファイルの数行です。

In [ ]:

for line in dataset.take(5):
  print(line.numpy())

複数のファイルの行を交互に抽出するには、Dataset.interleave を使用するとより簡単にファイルを混ぜ合わせて抽出できます。以下は、各変換の 1 行目から 3 行目です。

In [ ]:

files_ds = tf.data.Dataset.from_tensor_slices(file_paths)
lines_ds = files_ds.interleave(tf.data.TextLineDataset, cycle_length=3)

for i, line in enumerate(lines_ds.take(9)):
  if i % 3 == 0:
    print()
  print(line.numpy())

ファイルがヘッダー行で始まる場合やコメントが含まれている場合なども含め、TextLineDataset は、デフォルトで各ファイルの行を抽出しますが、Dataset.skip() または Dataset.filter() 変換を使用することで、こういった行を取り除くことができます。以下では、最初の行をスキップし、フィルタをかけて生存者のみを検索しています。

In [ ]:

titanic_file = tf.keras.utils.get_file("train.csv", "https://storage.googleapis.com/tf-datasets/titanic/train.csv")
titanic_lines = tf.data.TextLineDataset(titanic_file)

In [ ]:

for line in titanic_lines.take(10):
  print(line.numpy())

In [ ]:

def survived(line):
  return tf.not_equal(tf.strings.substr(line, 0, 1), "0")

survivors = titanic_lines.skip(1).filter(survived)

In [ ]:

for line in survivors.take(10):
  print(line.numpy())

CSV データの消費

その他の例については、CSV ファイルを読み込むチュートリアルとPandas DataFrames を読み込むチュートリアルをご覧ください。

CSV ファイル形式は、表形式のデータをプレーンテキストで保管するために広く使用される形式です。

以下に例を示します。

In [ ]:

titanic_file = tf.keras.utils.get_file("train.csv", "https://storage.googleapis.com/tf-datasets/titanic/train.csv")

In [ ]:

df = pd.read_csv(titanic_file)
df.head()

データがメモリに収まる場合は、同じ Dataset.from_tensor_slices メソッドを辞書に使用し、このデータをインポートしやすくすることができます。

In [ ]:

titanic_slices = tf.data.Dataset.from_tensor_slices(dict(df))

for feature_batch in titanic_slices.take(1):
  for key, value in feature_batch.items():
    print("  {!r:20s}: {}".format(key, value))

さらに拡張性の高い手法として、必要に応じてディスクから読み込むことができます。

tf.data モジュールには、RFC 4180 に準拠する 1 つ以上の CSV ファイルからレコードを抽出するメソッドがあります。

tf.data.experimental.make_csv_dataset 関数は、一連の CSV ファイルを読み取るための高度なインターフェースで、使用しやすいように、列の型推論や、バッチ処理とシャッフルといった多数の機能をサポートしています。

In [ ]:

titanic_batches = tf.data.experimental.make_csv_dataset(
    titanic_file, batch_size=4,
    label_name="survived")

In [ ]:

for feature_batch, label_batch in titanic_batches.take(1):
  print("'survived': {}".format(label_batch))
  print("features:")
  for key, value in feature_batch.items():
    print("  {!r:20s}: {}".format(key, value))

列のサブセットのみが必要である場合には、select_columns 引数を使用できます。

In [ ]:

titanic_batches = tf.data.experimental.make_csv_dataset(
    titanic_file, batch_size=4,
    label_name="survived", select_columns=['class', 'fare', 'survived'])

In [ ]:

for feature_batch, label_batch in titanic_batches.take(1):
  print("'survived': {}".format(label_batch))
  for key, value in feature_batch.items():
    print("  {!r:20s}: {}".format(key, value))

また、より細かい制御を行えるように、experimental.CsvDataset という下位クラスもあります。列の型推論はサポートされていないため、各列の型を指定する必要があります。

In [ ]:

titanic_types  = [tf.int32, tf.string, tf.float32, tf.int32, tf.int32, tf.float32, tf.string, tf.string, tf.string, tf.string]
dataset = tf.data.experimental.CsvDataset(titanic_file, titanic_types , header=True)

for line in dataset.take(10):
  print([item.numpy() for item in line])

この下位インターフェースでは、一部の列が空である場合に、列の型の代わりに使用するデフォルト値を指定することができます。

In [ ]:

%%writefile missing.csv
1,2,3,4
,2,3,4
1,,3,4
1,2,,4
1,2,3,
,,,

In [ ]:

# Creates a dataset that reads all of the records from two CSV files, each with
# four float columns which may have missing values.

record_defaults = [999,999,999,999]
dataset = tf.data.experimental.CsvDataset("missing.csv", record_defaults)
dataset = dataset.map(lambda *items: tf.stack(items))
dataset

In [ ]:

for line in dataset:
  print(line.numpy())

デフォルトでは、CsvDataset はファイルのすべての行のすべての列を生成するようになっていますが、ファイルが無視すべきヘッダー行で開始している場合や、入力に不要な列がある場合など、このデフォルトの動作が望ましくないことがあります。こういった行とフィールドについては、それぞれ header 引数と select_cols 引数を使用することで取り除くことができます。

In [ ]:

# Creates a dataset that reads all of the records from two CSV files with
# headers, extracting float data from columns 2 and 4.
record_defaults = [999, 999] # Only provide defaults for the selected columns
dataset = tf.data.experimental.CsvDataset("missing.csv", record_defaults, select_cols=[1, 3])
dataset = dataset.map(lambda *items: tf.stack(items))
dataset

In [ ]:

for line in dataset:
  print(line.numpy())

ファイルのセットの消費

多くのデータセットは一連のファイルセットとして分散されており、各ファイルは 1 つの例です。

In [ ]:

flowers_root = tf.keras.utils.get_file(
    'flower_photos',
    'https://storage.googleapis.com/download.tensorflow.org/example_images/flower_photos.tgz',
    untar=True)
flowers_root = pathlib.Path(flowers_root)

注意: これらの画像のライセンスは CC-BY に帰属します。詳細は、LICENSE.txt を参照してください。

ルートディレクトリには、各クラスのディレクトリが含まれます。

In [ ]:

for item in flowers_root.glob("*"):
  print(item.name)

各クラスディレクトリのファイルは例です。

In [ ]:

list_ds = tf.data.Dataset.list_files(str(flowers_root/'*/*'))

for f in list_ds.take(5):
  print(f.numpy())

tf.io.read_file 関数を使用してデータを読み取り、パスからラベルを抽出して (image, label) のペアを返します。

In [ ]:

def process_path(file_path):
  label = tf.strings.split(file_path, os.sep)[-2]
  return tf.io.read_file(file_path), label

labeled_ds = list_ds.map(process_path)

In [ ]:

for image_raw, label_text in labeled_ds.take(1):
  print(repr(image_raw.numpy()[:100]))
  print()
  print(label_text.numpy())

データセット要素のバッチ処理

単純なバッチ処理

最も単純なバッチ処理の形態は、n 個の連続するデータセット要素を 1 つの要素にスタックする方法です。Dataset.batch() 変換によってこれを実行することができますが、 tf.stack() 演算子と同じ制約が伴い、要素の各コンポーネントに適用されます。つまり、各コンポーネント i に対し、すべての要素にまったく同じ形状のテンソルが必要です。

In [ ]:

inc_dataset = tf.data.Dataset.range(100)
dec_dataset = tf.data.Dataset.range(0, -100, -1)
dataset = tf.data.Dataset.zip((inc_dataset, dec_dataset))
batched_dataset = dataset.batch(4)

for batch in batched_dataset.take(4):
  print([arr.numpy() for arr in batch])

tf.data は形状情報を伝搬しようとしますが、最後のバッチがいっぱいになっていない可能性があるため、Dataset.batch のデフォルト設定は不明なバッチサイズとなります。形状の None に注意してください。

In [ ]:

batched_dataset

drop_remainder 引数を使用して最後のバッチを無視し、完全に形状を伝搬させます。

In [ ]:

batched_dataset = dataset.batch(7, drop_remainder=True)
batched_dataset

パディングによるテンソルのバッチ処理

上記のレシピは、すべてのテンソルが同じサイズである場合に機能しますが、多くのモデル（シーケンスモデルなど）では、さまざまなサイズの入力データが使用されています（異なる長さのシーケンスなど）。このケースに対応するには、Dataset.padded_batch 変換を使い、パディングされている可能性のある 1 つ以上の次元を指定することで、異なる形状のテンソルをバッチ処理することができます。

In [ ]:

dataset = tf.data.Dataset.range(100)
dataset = dataset.map(lambda x: tf.fill([tf.cast(x, tf.int32)], x))
dataset = dataset.padded_batch(4, padded_shapes=(None,))

for batch in dataset.take(2):
  print(batch.numpy())
  print()

Dataset.padded_batch 変換では、各コンポーネントに異なるパディングを設定できます。このパディングは可変長（上記の例では None で指定）または固定長です。また、パディング値をオーバーライドすることも可能で、その場合は、0 となります。

トレーニングワークフロー

複数のエポックの処理

tf.data API では、同一のデータの複数のエポックを主に 2 つの方法で処理することができます。

複数のエポック内でデータセットをイテレートする最も簡単な方法は、Dataset.repeat() 変換を使用することです。まず、titanic データのデータセットを作成します。

In [ ]:

titanic_file = tf.keras.utils.get_file("train.csv", "https://storage.googleapis.com/tf-datasets/titanic/train.csv")
titanic_lines = tf.data.TextLineDataset(titanic_file)

In [ ]:

def plot_batch_sizes(ds):
  batch_sizes = [batch.shape[0] for batch in ds]
  plt.bar(range(len(batch_sizes)), batch_sizes)
  plt.xlabel('Batch number')
  plt.ylabel('Batch size')

Dataset.repeat() 変換を引数を使用せずに適用すると、入力が無限に繰り返されます。

Dataset.repeat 変換は、1 つのエポックの最後と次のエポックの始まりを伝達することなく、その引数を連結します。このため、Dataset.repeat の後に適用される Dataset.batch は、エポックの境界をまたぐバッチを生成してしまいます。

In [ ]:

titanic_batches = titanic_lines.repeat(3).batch(128)
plot_batch_sizes(titanic_batches)

エポックを明確に分離する必要がある場合は、repeat の前に Dataset.batch を記述します。

In [ ]:

titanic_batches = titanic_lines.batch(128).repeat(3)

plot_batch_sizes(titanic_batches)

各エポックの最後にカスタム計算（統計の収集など）を実行する場合は、エポックごとにデータセットのイテレーションを再開することが最も簡単です。

In [ ]:

epochs = 3
dataset = titanic_lines.batch(128)

for epoch in range(epochs):
  for batch in dataset:
    print(batch.shape)
  print("End of epoch: ", epoch)

入力データのランダムシャッフル

Dataset.shuffle() 変換は、固定サイズのバッファを維持し、次の要素をそのバッファからランダムに均等して選択します。

注意: buffer_size が大きければより全体的にシャッフルされますが、メモリを多く消費し、より長い時間がかかる可能性があります。これが問題となる場合は、ファイル全体で Dataset.interleave を使用することを検討してください。

効果を確認できるように、データセットにインデックスを追加します。

In [ ]:

lines = tf.data.TextLineDataset(titanic_file)
counter = tf.data.experimental.Counter()

dataset = tf.data.Dataset.zip((counter, lines))
dataset = dataset.shuffle(buffer_size=100)
dataset = dataset.batch(20)
dataset

buffer_size は 100 であり、バッチサイズは 20 であるため、最初のバッチには、120 を超えるインデックスの要素は含まれません。

In [ ]:

n,line_batch = next(iter(dataset))
print(n.numpy())

Dataset.batch と同様に、Dataset.repeat に対する順番は重要です。

Dataset.shuffle は、シャッフルのバッファが空になるまでエポックの最後をシグナルしません。そのため、repeat の後に記述される shuffle は、次のエポックに移動する前のエポックのすべての要素を表示します。

In [ ]:

dataset = tf.data.Dataset.zip((counter, lines))
shuffled = dataset.shuffle(buffer_size=100).batch(10).repeat(2)

print("Here are the item ID's near the epoch boundary:\n")
for n, line_batch in shuffled.skip(60).take(5):
  print(n.numpy())

In [ ]:

shuffle_repeat = [n.numpy().mean() for n, line_batch in shuffled]
plt.plot(shuffle_repeat, label="shuffle().repeat()")
plt.ylabel("Mean item ID")
plt.legend()

ただし、shuffle の前の repeat によって、エポックの境界が混合されてしまいます。

In [ ]:

dataset = tf.data.Dataset.zip((counter, lines))
shuffled = dataset.repeat(2).shuffle(buffer_size=100).batch(10)

print("Here are the item ID's near the epoch boundary:\n")
for n, line_batch in shuffled.skip(55).take(15):
  print(n.numpy())

In [ ]:

repeat_shuffle = [n.numpy().mean() for n, line_batch in shuffled]

plt.plot(shuffle_repeat, label="shuffle().repeat()")
plt.plot(repeat_shuffle, label="repeat().shuffle()")
plt.ylabel("Mean item ID")
plt.legend()

データの前処理

Dataset.map(f) 変換は、指定された関数 f を入力データセットの各要素に適用します。これは、関数型プログラミング言語でリスト（およびその他の構造）に一般的に適用される map() 関数に基づきます。関数 f は、入力内の単一の要素を表す tf.Tensor オブジェクトを取り、新しいデータセット内の単一の要素を表す tf.Tensor オブジェクトを返します。この実装には、1 つの要素を別の要素に変換する標準的な TensorFlow 演算が使用されています。

このセクションでは、Dataset.map() の一般的な使用例を説明します。

画像データのデコードとサイズ変更

実世界の画像データでニューラルネットワークを訓練する際、異なるサイズのデータを固定サイズでバッチ処理できるように、一般的なサイズに変換しなければならないことがよくあります。

花のファイル名のデータセットを再構築します。

In [ ]:

list_ds = tf.data.Dataset.list_files(str(flowers_root/'*/*'))

データセットの要素を操作する関数を記述します。

In [ ]:

# Reads an image from a file, decodes it into a dense tensor, and resizes it
# to a fixed shape.
def parse_image(filename):
  parts = tf.strings.split(filename, os.sep)
  label = parts[-2]

  image = tf.io.read_file(filename)
  image = tf.io.decode_jpeg(image)
  image = tf.image.convert_image_dtype(image, tf.float32)
  image = tf.image.resize(image, [128, 128])
  return image, label

これが機能するかをテストします。

In [ ]:

file_path = next(iter(list_ds))
image, label = parse_image(file_path)

def show(image, label):
  plt.figure()
  plt.imshow(image)
  plt.title(label.numpy().decode('utf-8'))
  plt.axis('off')

show(image, label)

それをデータセットにマッピングします。

In [ ]:

images_ds = list_ds.map(parse_image)

for image, label in images_ds.take(2):
  show(image, label)

任意の Python ロジックの適用

パフォーマンスの理由により、データの前処理には可能な限り TensorFlow 演算を使用してください。ただし、入力データを解析する際は、外部の Python ライブラリを呼び出すと便利な場合があります。Dataset.map 変換で、tf.py_function 演算を使用することができます。

たとえば、ランダムな回転を適用する場合、tf.image モジュールには tf.image.rot90 しかないため、画像を拡張するにはあまり便利とは言えません。

注意: tensorflow_addons には、tensorflow_addons.image.rotate というように、TensorFlow と互換性のある rotate があります。

tf.py_function の動作を実演するために、scipy.ndimage.rotate 関数を代わりに使用してみましょう。

In [ ]:

import scipy.ndimage as ndimage

def random_rotate_image(image):
  image = ndimage.rotate(image, np.random.uniform(-30, 30), reshape=False)
  return image

In [ ]:

image, label = next(iter(images_ds))
image = random_rotate_image(image)
show(image, label)

この関数を Dataset.map と使用する場合、Dataset.from_generator と同じ警告が適用されるため、関数を適用する際に、戻される形状と型を記述する必要があります。

In [ ]:

def tf_random_rotate_image(image, label):
  im_shape = image.shape
  [image,] = tf.py_function(random_rotate_image, [image], [tf.float32])
  image.set_shape(im_shape)
  return image, label

In [ ]:

rot_ds = images_ds.map(tf_random_rotate_image)

for image, label in rot_ds.take(2):
  show(image, label)

`tf.Example` プロトコルバッファメッセージの解析

多くの入力パイプラインは、TFRecord 形式から tf.train.Example プロトコルバッファメッセージを抽出します。各 tf.train.Example レコードには、1 つ以上の「特徴量」が含まれており、入力パイプラインは通常、これらの特徴量をテンソルに変換します。

In [ ]:

fsns_test_file = tf.keras.utils.get_file("fsns.tfrec", "https://storage.googleapis.com/download.tensorflow.org/data/fsns-20160927/testdata/fsns-00000-of-00001")
dataset = tf.data.TFRecordDataset(filenames = [fsns_test_file])
dataset

データを理解するには、tf.data.Dataset の外部で tf.train.Example プロトコルを処理することができます。

In [ ]:

raw_example = next(iter(dataset))
parsed = tf.train.Example.FromString(raw_example.numpy())

feature = parsed.features.feature
raw_img = feature['image/encoded'].bytes_list.value[0]
img = tf.image.decode_png(raw_img)
plt.imshow(img)
plt.axis('off')
_ = plt.title(feature["image/text"].bytes_list.value[0])

In [ ]:

raw_example = next(iter(dataset))

In [ ]:

def tf_parse(eg):
  example = tf.io.parse_example(
      eg[tf.newaxis], {
          'image/encoded': tf.io.FixedLenFeature(shape=(), dtype=tf.string),
          'image/text': tf.io.FixedLenFeature(shape=(), dtype=tf.string)
      })
  return example['image/encoded'][0], example['image/text'][0]

In [ ]:

img, txt = tf_parse(raw_example)
print(txt.numpy())
print(repr(img.numpy()[:20]), "...")

In [ ]:

decoded = dataset.map(tf_parse)
decoded

In [ ]:

image_batch, text_batch = next(iter(decoded.batch(10)))
image_batch.shape

時系列ウィンドウ

エンドツーエンドの時系列の例については、時系列の予測をご覧ください。

時系列データは、時間軸がそのままの状態で編成されることがよくあります。

単純な Dataset.range を使用して実演します。

In [ ]:

range_ds = tf.data.Dataset.range(100000)

通常、このようなデータに基づくでモデルでは、連続したタイムスライスが求められます。

最も単純な手法は、データをバッチ処理することです。

`batch` の使用

In [ ]:

batches = range_ds.batch(10, drop_remainder=True)

for batch in batches.take(5):
  print(batch.numpy())

または、密度の高い予測を 1 ステップ先に行うには、特徴量とラベルを相互に 1 ステップずつシフトします。

In [ ]:

def dense_1_step(batch):
  # Shift features and labels one step relative to each other.
  return batch[:-1], batch[1:]

predict_dense_1_step = batches.map(dense_1_step)

for features, label in predict_dense_1_step.take(3):
  print(features.numpy(), " => ", label.numpy())

固定オフセットの代わりにウィンドウ全体を予測するには、バッチを 2 つに分割することができます。

In [ ]:

batches = range_ds.batch(15, drop_remainder=True)

def label_next_5_steps(batch):
  return (batch[:-5],   # Inputs: All except the last 5 steps
          batch[-5:])   # Labels: The last 5 steps

predict_5_steps = batches.map(label_next_5_steps)

for features, label in predict_5_steps.take(3):
  print(features.numpy(), " => ", label.numpy())

1 つのバッチの特徴量と別のバッチのラベルをオーバーラップできるようにするには、Dataset.zip を使用します。

In [ ]:

feature_length = 10
label_length = 3

features = range_ds.batch(feature_length, drop_remainder=True)
labels = range_ds.batch(feature_length).skip(1).map(lambda labels: labels[:label_length])

predicted_steps = tf.data.Dataset.zip((features, labels))

for features, label in predicted_steps.take(5):
  print(features.numpy(), " => ", label.numpy())

`window` の使用

Dataset.batch を使用することもできますが、より細かい制御が必要となる場合があります。Dataset.window メソッドを使えば完全に制御できるようになりますが、このメソッドは、Datasets の Dataset を返すことに注意する必要があります。詳細は、Dataset の構造をご覧ください。

In [ ]:

window_size = 5

windows = range_ds.window(window_size, shift=1)
for sub_ds in windows.take(5):
  print(sub_ds)

Dataset.flat_map メソッドは、データセットのデータセットを取り、単一のデータセットにフラット化することができます。

In [ ]:

 for x in windows.flat_map(lambda x: x).take(30):
   print(x.numpy(), end=' ')

ほぼすべての場合において、最初にデータセットを Dataset.batch 処理します。

In [ ]:

def sub_to_batch(sub):
  return sub.batch(window_size, drop_remainder=True)

for example in windows.flat_map(sub_to_batch).take(5):
  print(example.numpy())

これで、shift 引数がどれくらい各ウィンドウを移動するかがわかりました。

これを合わせて、この関数を記述することができます。

In [ ]:

def make_window_dataset(ds, window_size=5, shift=1, stride=1):
  windows = ds.window(window_size, shift=shift, stride=stride)

  def sub_to_batch(sub):
    return sub.batch(window_size, drop_remainder=True)

  windows = windows.flat_map(sub_to_batch)
  return windows

In [ ]:

ds = make_window_dataset(range_ds, window_size=10, shift = 5, stride=3)

for example in ds.take(10):
  print(example.numpy())

こうすると、前と同じようにラベルを簡単に抽出できるようになります。

In [ ]:

dense_labels_ds = ds.map(dense_1_step)

for inputs,labels in dense_labels_ds.take(3):
  print(inputs.numpy(), "=>", labels.numpy())

リサンプリング

クラスが非常に不均衡なデータセットを使用する場合は、データセットをリサンプリングすることができます。tf.data では、2 つのメソッドを使ってこれを実行することができます。こういった問題の例には、クレジットカード詐欺のデータセットを使用できます。

注意: チュートリアル全文は、不均衡なデータでの分類をご覧ください。

In [ ]:

zip_path = tf.keras.utils.get_file(
    origin='https://storage.googleapis.com/download.tensorflow.org/data/creditcard.zip',
    fname='creditcard.zip',
    extract=True)

csv_path = zip_path.replace('.zip', '.csv')

In [ ]:

creditcard_ds = tf.data.experimental.make_csv_dataset(
    csv_path, batch_size=1024, label_name="Class",
    # Set the column types: 30 floats and an int.
    column_defaults=[float()]*30+[int()])

ここで、クラスの分布を確認してください。非常に歪んでいます。

In [ ]:

def count(counts, batch):
  features, labels = batch
  class_1 = labels == 1
  class_1 = tf.cast(class_1, tf.int32)

  class_0 = labels == 0
  class_0 = tf.cast(class_0, tf.int32)

  counts['class_0'] += tf.reduce_sum(class_0)
  counts['class_1'] += tf.reduce_sum(class_1)

  return counts

In [ ]:

counts = creditcard_ds.take(10).reduce(
    initial_state={'class_0': 0, 'class_1': 0},
    reduce_func = count)

counts = np.array([counts['class_0'].numpy(),
                   counts['class_1'].numpy()]).astype(np.float32)

fractions = counts/counts.sum()
print(fractions)

不均衡なデータを使用して訓練する際の一般的な手法は、データの均衡をとることです。tf.data には、このワークフローを実行するためのメソッドがいくつか含まれています。

データセットのサンプリング

データセットをリサンプリングするための 1 つに、sample_from_datasets を使用する手法が挙げられます。これは、クラスごとに別々の tf.data.Dataset がある場合に、さらに適用できる手法です。

ここでは、フィルタをかけて、クレジットカード詐欺のデータセットからデータセットを生成します。

In [ ]:

negative_ds = (
  creditcard_ds
    .unbatch()
    .filter(lambda features, label: label==0)
    .repeat())
positive_ds = (
  creditcard_ds
    .unbatch()
    .filter(lambda features, label: label==1)
    .repeat())

In [ ]:

for features, label in positive_ds.batch(10).take(1):
  print(label.numpy())

tf.data.Dataset.sample_from_datasets を使用するために、データセットと各データセットのウェイトを渡します。

In [ ]:

balanced_ds = tf.data.Dataset.sample_from_datasets(
    [negative_ds, positive_ds], [0.5, 0.5]).batch(10)

これでデータセットから各クラスのサンプルが 50/50 の等しい確率で得られるようになりました。

In [ ]:

for features, labels in balanced_ds.take(10):
  print(labels.numpy())

棄却リサンプリング

上記の Dataset.sample_from_datasets の手法には、クラスごとに個別の tf.data.Dataset が必要となることが難点です。Dataset.filter を使用することもできますが、すべてのデータが 2 回読み込まれることになってしまいます。

tf.data.Dataset.rejection_resample メソッドは、適用すると、1 回の読み込みでデータセットの均衡をとることができます。要素については、均衡を得るためにドロップされるか繰り返されます。

rejection_resample メソッドは、class_func 引数を取ります。この class_func は、各データセット要素に適用され、均衡を取る目的で、どのクラスに例が属するかを判定するために使用されます。

ここでの目標は、ラベルの分布を均衡化することですが、creditcard_ds はすでに (features, label) ペアになっているため、class_func はそれらのラベルを戻すことができます。

In [ ]:

def class_func(features, label):
  return label

resampler は個別の例を処理するため、そのメソッドを適用する前にデータセットを unbatch する必要があります。

メソッドにはターゲット分布と、オプションとして初期分布の推定が必要です。

In [ ]:

resample_ds = (
    creditcard_ds
    .unbatch()
    .rejection_resample(class_func, target_dist=[0.5,0.5],
                        initial_dist=fractions)
    .batch(10))

rejection_resample メソッドは (class, example) ペアを返します。class は class_func の出力です。この場合、example はすでに (feature, label) ペアであるため、map を使用して、余分なラベルのコピーを取り除きます。

In [ ]:

balanced_ds = resample_ds.map(lambda extra_label, features_and_label: features_and_label)

これでデータセットから各クラスのサンプルが 50/50 の等しい確率で得られるようになりました。

In [ ]:

for features, labels in balanced_ds.take(10):
  print(labels.numpy())

イテレータのチェックポイント

Tensorflow では、トレーニングプロセスが再開する際に、ほとんどの進捗状況を復元するための最新のチェックポイントを復元できるように、チェックポイントの使用がサポートされています。モデルの変数にチェックポイントを設定するだけでなく、データセットのイテレータにチェックポイントを設定することもできます。このため、大規模なデータセットを使用しており、再開されるたびにデータセットの始まりから開始しないようにする場合に役立ちます。ただし、Dataset.shuffle や Dataset.prefetch などの変換には、イテレータ内にバッファ要素が必要となるため、イテレータのチェックポイントが大量になる可能性があります。

チェックポイントにイテレータを含めるには、イテレータを tf.train.Checkpoint コンストラクタに渡します。

In [ ]:

range_ds = tf.data.Dataset.range(20)

iterator = iter(range_ds)
ckpt = tf.train.Checkpoint(step=tf.Variable(0), iterator=iterator)
manager = tf.train.CheckpointManager(ckpt, '/tmp/my_ckpt', max_to_keep=3)

print([next(iterator).numpy() for _ in range(5)])

save_path = manager.save()

print([next(iterator).numpy() for _ in range(5)])

ckpt.restore(manager.latest_checkpoint)

print([next(iterator).numpy() for _ in range(5)])

注意: tf.py_function などの外部の状態に依存するイテレータにチェックポイントを設定することはできません。設定しようとすると、外部の状態に関する問題を示す例外が発生します。

`tf.keras` と `tf.data` を使用する

tf.keras API は、機械学習モデルの作成や実行に関する多くの側面を単純化します。その Model.fit、Model.evaluate、および Model.predict APIは、入力としてのデータセットをサポートします。以下は、簡単なデータセットとモデルのセットアップです。

In [ ]:

train, test = tf.keras.datasets.fashion_mnist.load_data()

images, labels = train
images = images/255.0
labels = labels.astype(np.int32)

In [ ]:

fmnist_train_ds = tf.data.Dataset.from_tensor_slices((images, labels))
fmnist_train_ds = fmnist_train_ds.shuffle(5000).batch(32)

model = tf.keras.Sequential([
  tf.keras.layers.Flatten(),
  tf.keras.layers.Dense(10)
])

model.compile(optimizer='adam',
              loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True), 
              metrics=['accuracy'])

(feature, label) ペアのデータセットを渡すだけで、Model.fit と Model.evaluate を実行できます。

In [ ]:

model.fit(fmnist_train_ds, epochs=2)

Dataset.repeat を呼び出すなどして無限データセットを渡す場合も、steps_per_epoch 引数を渡すだけで完了です。

In [ ]:

model.fit(fmnist_train_ds.repeat(), epochs=2, steps_per_epoch=20)

評価するには、評価のステップ数を渡すことができます。

In [ ]:

loss, accuracy = model.evaluate(fmnist_train_ds)
print("Loss :", loss)
print("Accuracy :", accuracy)

長いデータセットについては、評価するステップ数を設定します。

In [ ]:

loss, accuracy = model.evaluate(fmnist_train_ds.repeat(), steps=10)
print("Loss :", loss)
print("Accuracy :", accuracy)

Model.predict を呼び出す際に、ラベルは必要ありません。

In [ ]:

predict_ds = tf.data.Dataset.from_tensor_slices(images).batch(32)
result = model.predict(predict_ds, steps = 10)
print(result.shape)

ただし、ラベルを含むデータセットを渡した場合、そのラベルは無視されます。

In [ ]:

result = model.predict(fmnist_train_ds, steps = 10)
print(result.shape)

Copyright 2018 The TensorFlow Authors.

tf.data: TensorFlow 入力パイプラインの構築

基本的な仕組み

データセットの構造

入力データの読み取り

NumPy 配列を消費する

Python ジェネレータの消費

TFRecord データの消費

テキストデータの消費

CSV データの消費

ファイルのセットの消費

データセット要素のバッチ処理

単純なバッチ処理

パディングによるテンソルのバッチ処理

トレーニングワークフロー

複数のエポックの処理

入力データのランダムシャッフル

データの前処理

画像データのデコードとサイズ変更

任意の Python ロジックの適用

`tf.Example` プロトコルバッファメッセージの解析

時系列ウィンドウ

`batch` の使用

`window` の使用

リサンプリング

データセットのサンプリング

棄却リサンプリング

イテレータのチェックポイント

`tf.keras` と `tf.data` を使用する

Product

Resources

Company

Copyright 2018 The TensorFlow Authors.

tf.data: TensorFlow 入力パイプラインの構築

基本的な仕組み

データセットの構造

入力データの読み取り

NumPy 配列を消費する

Python ジェネレータの消費

TFRecord データの消費

テキストデータの消費

CSV データの消費

ファイルのセットの消費

データセット要素のバッチ処理

単純なバッチ処理

パディングによるテンソルのバッチ処理

トレーニングワークフロー

複数のエポックの処理

入力データのランダムシャッフル

データの前処理

画像データのデコードとサイズ変更

任意の Python ロジックの適用

tf.Example プロトコルバッファメッセージの解析

時系列ウィンドウ

batch の使用

window の使用

リサンプリング

データセットのサンプリング

棄却リサンプリング

イテレータのチェックポイント

tf.keras と tf.data を使用する

`tf.Example` プロトコルバッファメッセージの解析

`batch` の使用

`window` の使用

`tf.keras` と `tf.data` を使用する