Tensorflowデータ検証を開始する

Tensorflow Data Validation（TFDV）は、トレーニングとサービスのデータを分析して次のことを行うことができます。

• 記述計算の統計情報を、

• 推論スキーマを、

• 検出データの異常を。

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

記述データ統計の計算

TFDVは、記述を計算することができます統計存在している機能とその値分布の形状の観点からデータの簡単な概要を提供します。以下のようなツールのファセットの概要は、簡単にブラウジングのため、これらの統計の簡潔な可視化を提供することができます。

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

    stats = tfdv.generate_statistics_from_tfrecord(data_location=path)

返される値であるDatasetFeatureStatisticsListプロトコルバッファ。例えば、ノートブックは、使用して統計の可視化が含まファセットの概要：

    tfdv.visualize_statistics(stats)

前の例では、データが保存されていることを前提としTFRecordファイル。 TFDVはCSV入力形式もサポートしており、他の一般的な形式にも拡張できます。あなたは、利用可能なデータデコーダを見つけることができるここに。また、TFDVが提供tfdv.generate_statistics_from_dataframeパンダデータフレームとして表されるメモリ内データとのユーザのための効用関数を。

TFDVは、デフォルトのデータ統計セットを計算するだけでなく、セマンティックドメイン（画像、テキストなど）の統計も計算できます。セマンティックドメイン統計の計算を有効にするには、合格tfdv.StatsOptionsを持つオブジェクトenable_semantic_domain_statsにTrueに設定tfdv.generate_statistics_from_tfrecord 。

GoogleCloudで実行

内部的には、TFDVは使用アパッチビーム大規模なデータセット上統計の計算をスケーリングするのデータ並列処理フレームワーク。 TFDVとのより深い統合を希望するアプリケーションの場合（例えば、データ生成パイプラインの最後に統計の生成を添付したカスタム形式のデータの統計情報を生成する）、APIはまた、統計作成のためのビームPTransformが公開されています。

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

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

次のスニペットは、GoogleCloudでの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)

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

NOTEのいずれかを呼び出すときtfdv.generate_statistics_...機能（例えば、 tfdv.generate_statistics_from_tfrecord Googleクラウド上に）、あなたが提供しなければならないoutput_path 。 Noneを指定すると、エラーが発生する場合があります。

データからスキーマを推測する

スキーマは、データの期待される特性を説明しています。これらのプロパティの一部は次のとおりです。

• どの機能が存在すると予想されるか
• それらのタイプ
• 各例の機能の値の数
• すべての例にわたる各機能の存在
• 機能の予想されるドメイン。

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

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

    schema = tfdv.infer_schema(stats)

一般に、TFDVは保守的なヒューリスティックを使用して、スキーマを特定のデータセットに過剰適合させないように、統計から安定したデータプロパティを推測します。ことを強く推測されるスキーマを見直し、TFDVのヒューリスティックが見逃している可能性があるデータについてのドメイン知識をキャプチャするために、必要に応じてそれを洗練することをお勧めします。

デフォルトでは、 tfdv.infer_schemaあれば、必要な機能のそれぞれの形状を推測value_count.min等しいvalue_count.max機能するために。設定し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)

結果は、のインスタンスである異常のプロトコルバッファと統計は、スキーマと一致しないすべてのエラーを説明しています。例えば、のデータと仮定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)

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

TFDVも提供validate_instanceスキーマと照合する際の個々の例では、異常を示すかどうかを識別するための機能。この関数を使用するには、例は、特徴名を特徴値のnumpy配列にマッピングするdictである必要があります。あなたは、デコーダで使用することができますtfx_bslにデコード、シリアライズtf.train.Exampleこの形式に秒。例えば：

   import tensorflow_data_validation as tfdv
   import tfx_bsl
   import pyarrow as pa
   decoder = tfx_bsl.coders.example_coder.ExamplesToRecordBatchDecoder()
   example = decoder.DecodeBatch([serialized_tfexample])
   options = tfdv.StatsOptions(schema=schema)
   anomalies = tfdv.validate_instance(example, options)

同様validate_statistics 、結果としてのインスタンスである異常の例では、指定されたスキーマと一致しないエラーを記述するプロトコルバッファ。

スキーマ環境

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

環境は、このような要件を表現するために使用することができます。特に、スキーマの機能は、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)

これを修正するには、すべての機能のデフォルト環境を「トレーニング」と「サービス」の両方に設定し、「ヒント」機能をサービス環境から除外する必要があります。

    # 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)

データセットがスキーマに設定された期待に適合するかどうかをチェックと同様、結果ものインスタンスである異常のプロトコルバッファと訓練となるデータセットとの間のスキューを記載しています。例えば、サービングデータは、特徴と有意に多くの例として含まれていると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)

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

計算データの統計に、TFDVはいくつか提供して便利な方法様々な形式で入力されたデータを処理するための（例えばTFRecordのtf.train.Example 、CSVなど）。データ形式がこのリストにない場合は、入力データを読み取るためのカスタムデータコネクタを作成し、データ統計を計算するためにTFDVコアAPIに接続する必要があります。

TFDVのデータの統計を計算するためのコアAPIは、あるビームPTransform入力例（入力例のバッチは次のように表されるのバッチのPCollectionかかり矢印RecordBatch）、および単一含有PCollection出力DatasetFeatureStatisticsListプロトコルバッファ。

カスタムデータコネクタを実装した後はアローRecordBatchにご入力例バッチは、あなたがそれを接続する必要があることtfdv.GenerateStatisticsデータの統計情報を計算するためのAPI。テイクTFRecordのtf.train.Example例えばS」。我々は提供DecodeTFExampleのデータコネクタを、以下でそれを接続する方法の一例であるtfdv.GenerateStatistics API。

import tensorflow_data_validation as tfdv
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 out the examples from input files.
    | 'ReadData' >> beam.io.ReadFromTFRecord(file_pattern=DATA_LOCATION)
    # 2. Convert examples to Arrow RecordBatches, which represent example
    #    batches.
    | 'DecodeData' >> tf_example_decoder.DecodeTFExample()
    # 3. Invoke TFDV `GenerateStatistics` API to compute the data statistics.
    | 'GenerateStatistics' >> tfdv.GenerateStatistics()
    # 4. Materialize the generated data statistics.
    | 'WriteStatsOutput' >> WriteStatisticsToTFRecord(OUTPUT_LOCATION))

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

TFDVは、データのスライス全体の統計を計算するように構成できます。スライスは、矢印を取り込むスライス機能提供することによって有効にすることができRecordBatchと出力形式のタプルのシーケンス(slice key, record batch) 。 TFDVにする簡単な方法を提供特徴量ベースのスライシング関数生成の一部として提供することができるtfdv.StatsOptions統計を計算するときに。

スライスが有効になっている場合、出力DatasetFeatureStatisticsListのプロトが複数含まDatasetFeatureStatisticsのPROTOS、各スライスの一つ。各スライスは次のように設定されている固有の名前で識別されDatasetFeatureStatisticsプロトにおけるデータセット名。デフォルトでは、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])