入門 機械学習パイプライン ~Tensorflowで学ぶワークフローの自動化~について学習してみた[4章]

1.データ検証

以前の記事で機械学習ワークフローの最初の段階であるデータの読み込みをCSVで行いました。
パイプラインの基礎や簡易的なApache BEAMの実装、TFXの概要はこの記事で説明しているので、よかったら、みてみてください。

さて、データを読み込むことはできました。

その後に何をするべきでしょうか。
前処理？
いや、ワークフローにおいてはその前に行うことがあります。
それはデータ検証です。

データは逐次送られてきて増えるものです。
今日、モデル作成した際のデータの構造と明日のデータの構造は必ずしも同じわけではありません。

もし、大きく欠損が増えたら？データの統計分布が異なっていたら？（急に男性のデータしか入ってこなくなったとか）

その際はパイプラインを停止させる必要性があります。

じゃあその指標はどのようにして見つければ良いのか？

Pythonにはデータ検証を支援するパッケージである

Tensofflow Data Validation　があります。

新しく収集したデータが、モデルの学習に使ったデータセットとは異なる統計量であることを
ドリフトと言います。

その検知も行ってくれるのでとても便利です。

では早速データを取得していきましょう。

2.データ検証

2.1データの準備

# Initial dataset source
DATASET_URL = "http://bit.ly/building-ml-pipelines-dataset"
# Initial local dataset location
LOCAL_FILE_NAME = "/content/drive/MyDrive/pipeline-tfx/4/consumer_complaints_with_narrative.csv"


def download_dataset(url=DATASET_URL):
    """download_dataset downloads the remote dataset to a local path

    Keyword Arguments:
        url {string} --
            complete url path to the csv data source (default: {DATASET_URL})
        local_path {string} --
            initial local file location (default: {LOCAL_FILE_NAME})
    Returns:
        None
    """
    df = pd.read_csv(url, index_col=0)
    df.to_csv(LOCAL_FILE_NAME)
    logging.info("Download completed.")


download_dataset()

このCSVファイルを使ってデータの統計量を出していきます。
また、必要なライブラリのインストール手順は前の記事でやっているので、うまくインポートできない場合はそちらを参考にしてみてください。

2.2 要約統計量の算出

import tensorflow_data_validation as tfdv
stats = tfdv.generate_statistics_from_csv(
    data_location='/content/drive/MyDrive/pipeline-tfx/4/consumer_complaints_with_narrative.csv',
    delimiter=','
)

この2行で統計量が出せます。とても便利ですね。
可視化するには次の用にコーディングします。

tfdv.visualize_statistics(stats)

数値データについての統計量

カテゴリデータについての統計量

TFDVは自動的にスキーマ情報を取得してきます。
そのため、カテゴリデータの部分に名前などが出てきてしまう場合もあります。

数値データ、カテゴリデータそれぞれどのような値が算出されるのでしょうか。

2.3 スキーマの生成

次は、データのスキーマを生成します。

データスキーマは、データセットを表現するための記述形式です。
スキーマは、データセットに含まれるべき特徴と、各特徴量の型を定義します。

また、データの範囲も定義することもできます。
例えば、外れ値の最小値と最大値、許容される欠損レコード数の閾値などです。

これがとても大事な概念です。
データスキーマを用いることで、新しく入ってきたデータと、学習に使ったデータセットと一致しているか確認することができます。

作成した統計量からスキーマを自動的に作成することができます。
いかがスキーマの作成方法です。

schema = tfdv.infer_schema(stats)
tfdv.display_schema(schema)

Presenceは、特徴が必須(required)か必須ではない(optional)かを意味しています。

Domain情報とは、数値変数の場合は変数の取りうる範囲で、カテゴリカル変数の場合は値の種類（ex. 文字列など）に相当します。

例えば、季節のカラムがあったとして、1~4のデータで春から冬までを表現するとします。
そうした場合、1~4以外の数字が出てきたら異常と判断できるようになるのです。

このスキーマは自動的に算出されたものなので、必ずしも正しいわけではありません。
ドメイン知識を用いてスキーマの更新をする必要があります。

2.4 データセットの比較

学習データと検証データがあったとします。
検証データが学習データのスキーマに準拠しているのかを調べる必要があります。
TFDVを使用すると、その答えをすぐに得られることができます。

では早速行っていきます。

#検証データを学習データに分ける
df.iloc[5000:].to_csv('chapters/data_validation/dataset_1.csv', index=False)
df.iloc[:5000].to_csv('chapters/data_validation/dataset_2.csv', index=False)

train_stats = tfdv.generate_statistics_from_csv(
    data_location='chapters/data_validation/dataset_1.csv',
    delimiter=','
)
val_stats = tfdv.generate_statistics_from_csv(
    data_location='chapters/data_validation/dataset_2.csv',
    delimiter=','
)

tfdv.visualize_statistics(
    lhs_statistics=val_stats,
    rhs_statistics=train_stats,
    lhs_name='VAL_DATASET',
    rhs_name='TRAIN_DATASET'
)

青が検証データであり、オレンジが学習データです。

これを見ると、sub_issueの欠損率が低下しているのがみられます。
また、気になるのが、どちらも欠損値が50パーセントに近いことです。
もし、sub_issueの特徴量が大切であれば、データ収集の方法を修正する必要があります。

2.5 スキーマの更新

先ほど述べた通り、スキーマの変更は基本的にはドメイン知識を用いて、設定を行います。
先ほどのsub_issueについては、「これは90%ないといけない！」ということであれば、自動的に設定したスキーマ情報に「この特徴量は90%ないといけないよ」と明示的に設定させてあげることが可能です。

tfdv.get_feature(schema, 'sub_issue').presence.min_fraction = 0.9

たった1行で行うことができます。

3.TFDVと機械学習パイプラインの統合

これから、TFXを用いて、パイプラインコンポーネントを作成していきます。

TFXはStatisticsGENというパイプラインコンポーネントを提供しています。
このコンポーネントはExampleGenコンポーネントの出力を入力として受け取り、統計量を生成します。
ExampleGenの生成については前の記事で行っております。

前回のコードを含めて記述します。

import os
import tensorflow as tf
from tfx.components import CsvExampleGen
from tfx.orchestration.experimental.interactive.interactive_context import \
InteractiveContext
from tfx.components import StatisticsgGen
from tfx.components import SchemaGen
from tfx.components import ExampleValidator
from tfx.components import ExampleValidator

# メモリ内にMeatadatastoreをセットアップしている
context = InteractiveContext()
# データの取り込み
example_gen = CsvExampleGen("/content/drive/MyDrive/movielens/Untitled Folder")

'''
下から今回のコード
'''
# 統計量の算出
statistics_gen = StatisticsGen(
    example = example_gen.outputs["examples"]
)
# スキーマ情報の抽出
schema_gen = SchemaGen(
statistics=statistics_gen.outputs['statistics'],
infer_feature_shape=True)

# 統計量とスキーマが揃ったら、データセットを検証が行える
example_validator = ExampleValidator(
statistics=statistics_gen.outputs["statistics"],
schema=schema_gen.outputs["schema"]
)
context.run(example_validator)

context.run()

※注意
バージョン管理が複雑なため、うまく実行されない可能性が高いです。
うまく実行されることが確認されましたら、またバージョン等を追記いたします。

４.まとめ

今回は、TFXを用いて、データ検証まで行うことができました。
次回は前処理のフェーズを行っていきたいと思います。
それではまた。