TensorFlow データ検証を開始する

TensorFlow Data Validation (TFDV) は、トレーニング データとサービング データを次の目的で分析できます。

• 記述統計量の計算 ,

• スキーマを推測する

• データの異常を検出します。

コア API は、各機能をサポートし、その上に構築され、ノートブックのコンテキストで呼び出すことができる便利なメソッドを備えています。

記述データ統計の計算

TFDV は、存在する特徴とそれらの値分布の形状に関してデータの概要をすばやく提供する記述統計を計算できます。ファセットの概要などのツールを使用すると、これらの統計を簡潔に視覚化して参照しやすくすることができます。

たとえば、 path TFRecord形式のファイル (タイプtensorflow.Exampleのレコードを保持する) を指しているとします。次のスニペットは、TFDV を使用した統計の計算を示しています。

    stats = tfdv.generate_statistics_from_tfrecord(data_location=path)

戻り値はDatasetFeatureStatisticsListプロトコル バッファです。サンプルのノートブックには、 Facets Overviewを使用した統計の視覚化が含まれています。

    tfdv.visualize_statistics(stats)

前の例では、データがTFRecordファイルに格納されていることを前提としています。 TFDV は CSV 入力形式もサポートしており、他の一般的な形式に拡張できます。利用可能なデータ デコーダーは、こちらで見つけることができます。さらに、TFDV はtfdv.generate_statistics_from_dataframeユーティリティ関数を、pandas DataFrame として表されるメモリ内データを持つユーザーに提供します。

デフォルトのデータ統計セットの計算に加えて、TFDV はセマンティック ドメイン (画像、テキストなど) の統計も計算できます。セマンティック ドメイン統計の計算を有効にするには、 enable_semantic_domain_statsが True に設定されたtfdv.StatsOptionsオブジェクトをtfdv.generate_statistics_from_tfrecordに渡します。

Google Cloud での実行

内部的には、TFDV はApache Beamのデータ並列処理フレームワークを使用して、大規模なデータセットに対する統計の計算をスケーリングします。 TFDV とより深く統合したいアプリケーション (例: データ生成パイプラインの最後に統計生成をアタッチする、カスタム形式でデータの統計を生成する) の場合、API は統計生成用の Beam PTransform も公開します。

Google Cloud で TFDV を実行するには、TFDV wheel ファイルをダウンロードして Dataflow ワーカーに提供する必要があります。次のように、ホイール ファイルを現在のディレクトリにダウンロードします。

pip download tensorflow_data_validation \
  --no-deps \
  --platform manylinux2010_x86_64 \
  --only-binary=:all:

次のスニペットは、Google Cloud での TFDV の使用例を示しています。

import tensorflow_data_validation as tfdv
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions, SetupOptions

PROJECT_ID = ''
JOB_NAME = ''
GCS_STAGING_LOCATION = ''
GCS_TMP_LOCATION = ''
GCS_DATA_LOCATION = ''
# GCS_STATS_OUTPUT_PATH is the file path to which to output the data statistics
# result.
GCS_STATS_OUTPUT_PATH = ''

PATH_TO_WHL_FILE = ''


# Create and set your PipelineOptions.
options = PipelineOptions()

# For Cloud execution, set the Cloud Platform project, job_name,
# staging location, temp_location and specify DataflowRunner.
google_cloud_options = options.view_as(GoogleCloudOptions)
google_cloud_options.project = PROJECT_ID
google_cloud_options.job_name = JOB_NAME
google_cloud_options.staging_location = GCS_STAGING_LOCATION
google_cloud_options.temp_location = GCS_TMP_LOCATION
options.view_as(StandardOptions).runner = 'DataflowRunner'

setup_options = options.view_as(SetupOptions)
# PATH_TO_WHL_FILE should point to the downloaded tfdv wheel file.
setup_options.extra_packages = [PATH_TO_WHL_FILE]

tfdv.generate_statistics_from_tfrecord(GCS_DATA_LOCATION,
                                       output_path=GCS_STATS_OUTPUT_PATH,
                                       pipeline_options=options)

この場合、生成された統計プロトコルは、 GCS_STATS_OUTPUT_PATHに書き込まれた TFRecord ファイルに格納されます。

注: Google Cloud でtfdv.generate_statistics_...関数 (例: tfdv.generate_statistics_from_tfrecord ) を呼び出す場合は、 output_pathを指定する必要があります。 None を指定するとエラーになる場合があります。

データに対するスキーマの推測

スキーマは、期待されるデータのプロパティを記述します。これらのプロパティのいくつかは次のとおりです。

• どの機能が存在すると予想されるか
• 彼らのタイプ
• 各例の特徴の値の数
• すべての例における各機能の存在
• 機能の予想されるドメイン。

つまり、スキーマは「正しい」データに対する期待を記述しているため、データのエラーを検出するために使用できます (後述)。さらに、同じスキーマを使用して、データ変換用のTensorFlow Transformを設定できます。スキーマはかなり静的であることが期待されることに注意してください。たとえば、複数のデータセットが同じスキーマに準拠している可能性がありますが、統計 (上記) はデータセットごとに異なる可能性があります。

スキーマの作成は面倒な作業になる可能性があるため、特に多くの機能を持つデータセットの場合、TFDV は記述統計に基づいてスキーマの初期バージョンを生成する方法を提供します。

    schema = tfdv.infer_schema(stats)

一般に、TFDV は保守的なヒューリスティックを使用して統計から安定したデータ プロパティを推測し、スキーマが特定のデータセットに過剰適合するのを回避します。推論されたスキーマを確認し、必要に応じて改良して、TFDV のヒューリスティックが見逃した可能性のあるデータに関するドメインの知識を取得することを強くお勧めします。

デフォルトでは、 value_count.minその機能のvalue_count.maxと等しい場合、 tfdv.infer_schema必要な各機能の形状を推測します。形状の推定を無効にするには、 infer_feature_shape引数を False に設定します。

スキーマ自体はスキーマ プロトコル バッファとして格納されるため、標準のプロトコル バッファ API を使用して更新/編集できます。 TFDV には、これらの更新を簡単にするためのユーティリティ メソッドもいくつか用意されています。たとえば、単一の値を取る必要な文字列機能payment_type記述する次のスタンザがスキーマに含まれているとします。

feature {
  name: "payment_type"
  value_count {
    min: 1
    max: 1
  }
  type: BYTES
  domain: "payment_type"
  presence {
    min_fraction: 1.0
    min_count: 1
  }
}

例の少なくとも 50% で機能を設定する必要があることを示すには、次のようにします。

    tfdv.get_feature(schema, 'payment_type').presence.min_fraction = 0.5

ノートブックの例には、スキーマの単純な視覚化がテーブルとして含まれており、スキーマにエンコードされた各機能とその主な特性がリストされています。

データのエラーをチェックする

スキーマを指定すると、データセットがスキーマで設定された期待に準拠しているかどうか、またはデータ異常が存在するかどうかを確認できます。 (a) データセットの統計をスキーマと照合することにより、データセット全体の集計で、または (b) 例ごとにエラーをチェックすることにより、データのエラーをチェックできます。

データセットの統計をスキーマと照合する

集計のエラーをチェックするために、TFDV はデータセットの統計をスキーマと照合し、不一致をマークします。例えば：

    # Assume that other_path points to another TFRecord file
    other_stats = tfdv.generate_statistics_from_tfrecord(data_location=other_path)
    anomalies = tfdv.validate_statistics(statistics=other_stats, schema=schema)

結果はAnomaliesプロトコル バッファのインスタンスであり、統計がスキーマと一致しないエラーを示します。たとえば、 other_pathのデータに、スキーマで指定されたドメイン外の機能payment_typeの値を持つ例が含まれているとします。

これにより異常が発生します

   payment_type  Unexpected string values  Examples contain values missing from the schema: Prcard (<1%).

ドメイン外の値が特徴値の 1% 未満の統計で見つかったことを示します。

これが予期されていた場合、スキーマは次のように更新できます。

   tfdv.get_domain(schema, 'payment_type').value.append('Prcard')

異常が本当にデータ エラーを示している場合は、基になるデータをトレーニングに使用する前に修正する必要があります。

このモジュールで検出できるさまざまな異常の種類を次に示します。

サンプル ノートブックには、エラーが検出された機能と各エラーの簡単な説明を一覧表示するテーブルとしての異常の単純な視覚化が含まれています。

例ごとにエラーをチェックする

TFDV には、データセット全体の統計をスキーマと比較する代わりに、例ごとにデータを検証するオプションも用意されています。 TFDV は、例ごとにデータを検証し、見つかった異常な例の要約統計を生成する機能を提供します。例えば：

   options = tfdv.StatsOptions(schema=schema)
   anomalous_example_stats = tfdv.validate_examples_in_tfrecord(
       data_location=input, stats_options=options)

validate_examples_in_tfrecordが返すanomalous_example_stats DatasetFeatureStatisticsListプロトコル バッファであり、各データセットは特定の異常を示す一連の例で構成されています。これを使用して、特定の異常を示すデータセット内の例の数とそれらの例の特性を判断できます。

スキーマ環境

デフォルトでは、検証は、パイプライン内のすべてのデータセットが単一のスキーマに準拠していることを前提としています。場合によっては、わずかなスキーマのバリエーションを導入する必要があります。たとえば、トレーニング中にはラベルとして使用される機能が必要であり (検証する必要があります)、提供中には欠落しています。

環境は、そのような要件を表現するために使用できます。特に、スキーマ内の機能は、default_environment、in_environment、および not_in_environment を使用して一連の環境に関連付けることができます。

たとえば、ヒント機能がトレーニングでラベルとして使用されているが、提供データにない場合です。環境が指定されていないと、異常として表示されます。

    serving_stats = tfdv.generate_statistics_from_tfrecord(data_location=serving_data_path)
    serving_anomalies = tfdv.validate_statistics(serving_stats, schema)

これを修正するには、すべての機能のデフォルト環境を「TRAINING」と「SERVING」の両方に設定し、SERVING 環境から「tips」機能を除外する必要があります。

    # All features are by default in both TRAINING and SERVING environments.
    schema.default_environment.append('TRAINING')
    schema.default_environment.append('SERVING')

    # Specify that 'tips' feature is not in SERVING environment.
    tfdv.get_feature(schema, 'tips').not_in_environment.append('SERVING')

    serving_anomalies_with_env = tfdv.validate_statistics(
        serving_stats, schema, environment='SERVING')

データのスキューとドリフトの確認

データセットがスキーマで設定された期待に準拠しているかどうかを確認するだけでなく、TFDV は次の検出機能も提供します。

• トレーニング データとサービング データの間の偏り
• 異なる日のトレーニング データ間のドリフト

TFDV は、スキーマで指定されたドリフト/スキュー コンパレータに基づいて、さまざまなデータセットの統計を比較することにより、このチェックを実行します。たとえば、トレーニング データセットとサービング データセット内の「payment_type」機能に偏りがあるかどうかを確認するには、次のようにします。

    # Assume we have already generated the statistics of training dataset, and
    # inferred a schema from it.
    serving_stats = tfdv.generate_statistics_from_tfrecord(data_location=serving_data_path)
    # Add a skew comparator to schema for 'payment_type' and set the threshold
    # of L-infinity norm for triggering skew anomaly to be 0.01.
    tfdv.get_feature(schema, 'payment_type').skew_comparator.infinity_norm.threshold = 0.01
    skew_anomalies = tfdv.validate_statistics(
        statistics=train_stats, schema=schema, serving_statistics=serving_stats)

注 L 無限ノルムは、カテゴリ特徴のスキューのみを検出します。 infinity_normしきい値を指定する代わりに、 skew_comparatorでjensen_shannon_divergenceしきい値を指定すると、数値機能とカテゴリ機能の両方のスキューが検出されます。

データセットがスキーマで設定された期待に準拠しているかどうかをチェックする場合と同様に、結果はAnomaliesプロトコル バッファのインスタンスでもあり、トレーニング データセットとサービス データセットの間のスキューを示します。たとえば、機能payement_typeの値がCashであるサンプルがサービス データにかなり多く含まれているとします。これにより、スキュー異常が発生します。

   payment_type  High L-infinity distance between serving and training  The L-infinity distance between serving and training is 0.0435984 (up to six significant digits), above the threshold 0.01. The feature value with maximum difference is: Cash

異常がトレーニング データとサービング データの間の偏りを本当に示している場合は、モデルのパフォーマンスに直接影響する可能性があるため、さらに調査する必要があります。

サンプル ノートブックには、スキューベースの異常をチェックする簡単な例が含まれています。

異なる日のトレーニング データ間のドリフトの検出は、同様の方法で行うことができます。

    # Assume we have already generated the statistics of training dataset for
    # day 2, and inferred a schema from it.
    train_day1_stats = tfdv.generate_statistics_from_tfrecord(data_location=train_day1_data_path)
    # Add a drift comparator to schema for 'payment_type' and set the threshold
    # of L-infinity norm for triggering drift anomaly to be 0.01.
    tfdv.get_feature(schema, 'payment_type').drift_comparator.infinity_norm.threshold = 0.01
    drift_anomalies = tfdv.validate_statistics(
        statistics=train_day2_stats, schema=schema, previous_statistics=train_day1_stats)

注 L 無限ノルムは、カテゴリ特徴のスキューのみを検出します。 infinity_normしきい値を指定する代わりに、 skew_comparatorでjensen_shannon_divergenceしきい値を指定すると、数値機能とカテゴリ機能の両方のスキューが検出されます。

カスタム データ コネクタの作成

データ統計を計算するために、TFDV はさまざまな形式の入力データを処理するためのいくつかの便利なメソッドを提供します (例: tf.train.ExampleのTFRecord 、CSV など)。データ形式がこのリストにない場合は、入力データを読み取るためのカスタム データ コネクタを作成し、それを TFDV コア API に接続してデータ統計を計算する必要があります。

データ統計を計算するための TFDV コア API は、入力例のバッチの PCollection を受け取り (入力例のバッチはArrow RecordBatch として表されます)、単一のDatasetFeatureStatisticsListプロトコル バッファーを含む PCollection を出力するBeam PTransformです。

Arrow RecordBatch で入力例をバッチ処理するカスタム データ コネクタを実装したら、データ統計を計算するためにそれをtfdv.GenerateStatistics API に接続する必要があります。たとえば、 tf.train.ExampleのTFRecordを取ります。 tfx_bsl TFExampleRecordデータ コネクタを提供します。以下は、 tfdv.GenerateStatistics API と接続する方法の例です。

import tensorflow_data_validation as tfdv
from tfx_bsl.public import tfxio
import apache_beam as beam
from tensorflow_metadata.proto.v0 import statistics_pb2

DATA_LOCATION = ''
OUTPUT_LOCATION = ''

with beam.Pipeline() as p:
    _ = (
    p
    # 1. Read and decode the data with tfx_bsl.
    | 'TFXIORead' >> (
          tfxio.TFExampleRecord(
              file_pattern=[DATA_LOCATION],
              telemetry_descriptors=['my', 'tfdv']).BeamSource())
    # 2. Invoke TFDV `GenerateStatistics` API to compute the data statistics.
    | 'GenerateStatistics' >> tfdv.GenerateStatistics()
    # 3. Materialize the generated data statistics.
    | 'WriteStatsOutput' >> WriteStatisticsToTFRecord(OUTPUT_LOCATION))

データのスライスに対する統計の計算

TFDV は、データのスライスに対して統計を計算するように構成できます。 Arrow RecordBatch取り込ん(slice key, record batch)形式の一連のタプルを出力するスライス関数を提供することで、スライスを有効にすることができます。 TFDV は、統計を計算するときにtfdv.StatsOptionsの一部として提供できる特徴値ベースのスライス関数を生成する簡単な方法を提供します。

スライスが有効な場合、出力DatasetFeatureStatisticsList proto には、スライスごとに 1 つずつ、複数のDatasetFeatureStatistics proto が含まれます。各スライスは、 DatasetFeatureStatistics proto でデータセット名として設定された一意の名前で識別されます。デフォルトでは、TFDV は構成されたスライスに加えて、データセット全体の統計を計算します。

import tensorflow_data_validation as tfdv
from tensorflow_data_validation.utils import slicing_util

# Slice on country feature (i.e., every unique value of the feature).
slice_fn1 = slicing_util.get_feature_value_slicer(features={'country': None})

# Slice on the cross of country and state feature (i.e., every unique pair of
# values of the cross).
slice_fn2 = slicing_util.get_feature_value_slicer(
    features={'country': None, 'state': None})

# Slice on specific values of a feature.
slice_fn3 = slicing_util.get_feature_value_slicer(
    features={'age': [10, 50, 70]})

stats_options = tfdv.StatsOptions(
    slice_functions=[slice_fn1, slice_fn2, slice_fn3])