Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
tensorflow
GitHub Repository: tensorflow/docs-l10n
 Path: blob/master/site/ja/tfx/tutorials/transform/simple.ipynb
25118 views
Kernel: Python 3
#@title Licensed under the Apache License, Version 2.0 (the "License");
# 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.

TensorFlow Transform を使用したデータの前処理

TensorFlow Extended (TFX) の特徴エンジニアリングコンポーネント

注：この例は、Jupyter スタイルのノートブックで今すぐ実行できます。セットアップは必要ありません。「Google Colab で実行」をクリックするだけです

この Colab ノートブックの例では、TensorFlow Transform（tf.Transform）を使用して、簡単にデータを前処理する方法を紹介します。ここではモデルのトレーニングと本番環境での推論のサービングの両方に同じコードを使用します。

TensorFlow Transform は、トレーニングデータセットのフルパスを必要とする特徴の作成など、TensorFlow の入力データを前処理するためのライブラリです。たとえば、TensorFlow Transform を使用すると、次のことができます。

  • 平均と標準偏差を使用して入力値を正規化する

  • すべての入力値に対して語彙を生成することにより、文字列を整数に変換する

  • 観測されたデータ分布に基づいて、浮動小数点数をバケットに割り当てることにより、浮動小数点数を整数に変換する

TensorFlow には、単一のサンプルまたはサンプルのバッチに対する操作のサポートが組み込まれています。tf.Transformは、これらの機能を拡張して、トレーニングデータセット全体のフルパスをサポートします。

The output of tf.Transform is exported as a TensorFlow graph which you can use for both training and serving. Using the same graph for both training and serving can prevent skew, since the same transformations are applied in both stages.

Pip のアップグレード

ローカルで実行する場合にシステム Pip をアップグレードしないようにするには、Colab で実行していることを確認してください。もちろん、ローカルシステムは個別にアップグレードできます。

try:
  import colab
  !pip install --upgrade pip
except:
  pass

TensorFlow Transform のインストール

!pip install -q -U tensorflow_transform

# This cell is only necessary because packages were installed while python was
# running. It avoids the need to restart the runtime when running in Colab.
import pkg_resources
import importlib

importlib.reload(pkg_resources)

インポート

import pathlib
import pprint
import tempfile

import tensorflow as tf
import tensorflow_transform as tft

import tensorflow_transform.beam as tft_beam
from tensorflow_transform.tf_metadata import dataset_metadata
from tensorflow_transform.tf_metadata import schema_utils

データ: ダミーデータを作成する

簡単な例として、いくつかの簡単なダミーデータを作成します。

  • raw_data は前処理する最初の生データです

  • raw_data_metadata にはraw_data の各列の型を示すスキーマが含まれています。この例では非常に簡単です。

raw_data = [
      {'x': 1, 'y': 1, 's': 'hello'},
      {'x': 2, 'y': 2, 's': 'world'},
      {'x': 3, 'y': 3, 's': 'hello'}
  ]

raw_data_metadata = dataset_metadata.DatasetMetadata(
    schema_utils.schema_from_feature_spec({
        'y': tf.io.FixedLenFeature([], tf.float32),
        'x': tf.io.FixedLenFeature([], tf.float32),
        's': tf.io.FixedLenFeature([], tf.string),
    }))

変換：前処理関数を作成する

前処理関数は、tf.Transform の最も重要な概念です。前処理関数では、データセットの変換が実際に行われます。テンソルのディクショナリーを受け入れて返します。ここで、テンソルは Tensor または SparseTensor を意味します。通常、前処理関数の中心となる API 呼び出しには 2 つの主要なグループがあります。

  1. TensorFlow 演算子: テンソルを受け入れて返す関数。通常は TensorFlow 演算子を意味します。これらは、生データを一度に 1 つの特徴ベクトルで変換されたデータに変換するグラフに TensorFlow 演算子を追加します。これらは、トレーニングとサービングの両方で、すべてのサンプルで実行されます。

  2. Tensorflow Transform アナライザー/マッパー: tf.Transform によって提供されるアナライザー/マッパーのいずれか。これらもテンソルを受け入れて返し、通常は Tensorflow 演算子と Beam 計算の組み合わせを含みますが、TensorFlow 演算子とは異なり、分析中はビームパイプラインでのみ実行され、トレーニングデータセット全体を通じた処理が必要になります。Beam 計算は（トレーニング前、分析中に）1 回だけ実行され、通常はトレーニングデータセット全体を処理します。tf.constant テンソルが作成され、グラフに追加されます。たとえば、 tft.min は、トレーニングデータセットのテンソルの最小値を計算します。

注意: 前処理関数をサービング推論に適用する場合、トレーニング中にアナライザーにより作成された定数は変更されません。データに傾向または季節性の要素がある場合は、それに応じて計画します。

Note: The preprocessing_fn is not directly callable. This means that calling preprocessing_fn(raw_data) will not work. Instead, it must be passed to the Transform Beam API as shown in the following cells.

def preprocessing_fn(inputs):
    """Preprocess input columns into transformed columns."""
    x = inputs['x']
    y = inputs['y']
    s = inputs['s']
    x_centered = x - tft.mean(x)
    y_normalized = tft.scale_to_0_1(y)
    s_integerized = tft.compute_and_apply_vocabulary(s)
    x_centered_times_y_normalized = (x_centered * y_normalized)
    return {
        'x_centered': x_centered,
        'y_normalized': y_normalized,
        's_integerized': s_integerized,
        'x_centered_times_y_normalized': x_centered_times_y_normalized,
    }

構文

これで、すべてをまとめて Apache Beam を使用して実行する準備がほぼ整いました。

Apache Beam は、特別な構文を使用して変換を定義および呼び出します。たとえば、次の行をご覧ください。

result = pass_this | 'name this step' >> to_this_call

メソッド to_this_call が呼び出され、pass_this というオブジェクトが渡されます。この演算は、スタックトレースで name this step と呼ばれます。to_this_call の呼び出しの結果は、result に返されます。 頻繁にパイプラインのステージは次のようにチェーンされます。

result = apache_beam.Pipeline() | 'first step' >> do_this_first() | 'second step' >> do_this_last()

そして、新しいパイプラインで始まったので、以下のように続行します。

next_result = result | 'doing more stuff' >> another_function()

すべてを提供する

これで、データを変換する準備が整いました。ダイレクトランナーで Apache Beam を使用し、次の 3 つの入力を提供します。

  1. raw_data - 上で作成した生の入力データ

  2. raw_data_metadata - 生データのスキーマ

  3. preprocessing_fn - 変換を行うために作成した関数

def main(output_dir):
  # Ignore the warnings
  with tft_beam.Context(temp_dir=tempfile.mkdtemp()):
    transformed_dataset, transform_fn = (  # pylint: disable=unused-variable
        (raw_data, raw_data_metadata) | tft_beam.AnalyzeAndTransformDataset(
            preprocessing_fn))

  transformed_data, transformed_metadata = transformed_dataset  # pylint: disable=unused-variable

  # Save the transform_fn to the output_dir
  _ = (
      transform_fn
      | 'WriteTransformFn' >> tft_beam.WriteTransformFn(output_dir))

  return transformed_data, transformed_metadata

output_dir = pathlib.Path(tempfile.mkdtemp())

transformed_data, transformed_metadata = main(str(output_dir))

print('\nRaw data:\n{}\n'.format(pprint.pformat(raw_data)))
print('Transformed data:\n{}'.format(pprint.pformat(transformed_data)))

答えは正しいでしょうか？

以前は、これを行うために tf.Transform を使用しました。

x_centered = x - tft.mean(x)
y_normalized = tft.scale_to_0_1(y)
s_integerized = tft.compute_and_apply_vocabulary(s)
x_centered_times_y_normalized = (x_centered * y_normalized)

  • x_centered - [1, 2, 3] の入力では、x の平均は 2 であり、x からその平均を引いて x 値を 0 に設定します。この場合、正しい結果は [-1.0, 0.0, 1.0] になります。

  • y_normalized - y 値を 0 から 1 の間でスケーリングしています。入力は [1, 2, 3] であったため、正しい結果は [0.0, 0.5, 1.0] になります。

  • s_integerized - 文字列を語彙のインデックスにマップしています。語彙には 2 つの単語（「hello」と「world」）しかありませんでした。したがって、["hello", "world", "hello"] を入力する場合、正しい結果は [0, 1, 0] になります。このデータでは「hello」が最も頻繁に発生するため、「hello」が語彙の最初のエントリになります。

  • x_centered_times_y_normalized - 乗算を使用して x_centered と y_normalized を掛け合わせることにより、新しい機能を作成しています。これにより、元の値ではなく結果が乗算され、[-0.0, 0.0, 1.0] の新しい結果が正しいことに注意してください。

結果の transform_fn を使用する

!ls -l {output_dir}

transform_fn/ ディレクトリには、グラフに組み込まれたすべての定数 tensorflow-transform の分析結果を実装する tf.saved_model が含まれています。

これを tf.saved_model.load で直接ロードすることは可能ですが、これは簡単ではありません。

loaded = tf.saved_model.load(str(output_dir/'transform_fn'))
loaded.signatures['serving_default']

より良いアプローチは、tft.TFTransformOutput を使用してロードすることです。 TFTransformOutput.transform_features_layer メソッドは、変換を適用するために使用できる tft.TransformFeaturesLayer オブジェクトを返します。

tf_transform_output = tft.TFTransformOutput(output_dir)

tft_layer = tf_transform_output.transform_features_layer()
tft_layer

この tft.TransformFeaturesLayer は、バッチ機能のディクショナリを想定しています。したがって、raw_data に List[Dict[str, Any]] から Dict[str, tf.Tensor] を作成します。

raw_data_batch = {
    's': tf.constant([ex['s'] for ex in raw_data]),
    'x': tf.constant([ex['x'] for ex in raw_data], dtype=tf.float32),
    'y': tf.constant([ex['y'] for ex in raw_data], dtype=tf.float32),
}

tft.TransformFeaturesLayer は単独で使用できます。

transformed_batch = tft_layer(raw_data_batch)

{key: value.numpy() for key, value in transformed_batch.items()}

エクスポート

より一般的なユースケースでは、tf.Transform を使用してトレーニングおよび評価データセットに変換を適用します（例については、次のチュートリアルを参照してください）。次に、トレーニング後、モデルをエクスポートする前に、tft.TransformFeaturesLayer を最初のレイヤーとして添付して、tf.saved_model の一部としてエクスポートできるようにします。以下で、具体的な例を説明します。

トレーニングモデルの例

以下は次のようなモデルです。

  1. 変換済みのバッチを取り込み、

  2. すべてを一緒にまとめて、シンプルな (batch, features) 行列にし、

  3. いくつかの Dense レイヤーで実行し、そして

  4. 10 個の線形出力を生成する。

実際のユースケースでは、s_integerized 機能に one-hot を適用します。

このモデルは、tf.Transform によって変換されたデータセットでトレーニングできます。

class StackDict(tf.keras.layers.Layer):
  def call(self, inputs):
    values = [
        tf.cast(v, tf.float32)
        for k,v in sorted(inputs.items(), key=lambda kv: kv[0])]
    return tf.stack(values, axis=1)

class TrainedModel(tf.keras.Model):
  def __init__(self):
    super().__init__(self)
    self.concat = StackDict()
    self.body = tf.keras.Sequential([
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(10),
    ])

  def call(self, inputs, training=None):
    x = self.concat(inputs)
    return self.body(x, training)

trained_model = TrainedModel()

モデルをトレーニングしたと想像してください。

trained_model.compile(loss=..., optimizer='adam')
trained_model.fit(...)

このモデルは、変換された入力で実行されます

trained_model_output = trained_model(transformed_batch)
trained_model_output.shape

エクスポートラッパーの例

上記のモデルをトレーニングし、それをエクスポートしたいとします。

エクスポートされたモデルに変換関数を含める必要があります。

class ExportModel(tf.Module):
  def __init__(self, trained_model, input_transform):
    self.trained_model = trained_model
    self.input_transform = input_transform

  @tf.function
  def __call__(self, inputs, training=None):
    x = self.input_transform(inputs)
    return self.trained_model(x)

export_model = ExportModel(trained_model=trained_model,
                           input_transform=tft_layer)

この結合されたモデルは生データで機能し、トレーニングされたモデルを直接呼び出すのとまったく同じ結果を生成します。

export_model_output = export_model(raw_data_batch)
export_model_output.shape

tf.reduce_max(abs(export_model_output - trained_model_output)).numpy()

この export_model には tft.TransformFeaturesLayer が含まれており、完全に自己完結型です。保存して別の環境に復元しても、まったく同じ結果が得られます。

import tempfile
model_dir = tempfile.mkdtemp(suffix='tft')

tf.saved_model.save(export_model, model_dir)

reloaded = tf.saved_model.load(model_dir)

reloaded_model_output = reloaded(raw_data_batch)
reloaded_model_output.shape

tf.reduce_max(abs(export_model_output - reloaded_model_output)).numpy()