デベロッパー

ML ワークフローにおけるデータドリフトのイベントトリガー検出

2021年3月29日

https://storage.googleapis.com/gweb-cloudblog-publish/images/055-GC-AI-ML-Blog-Header-3.max-2800x2800.max-2600x2600.jpg

Google Cloud Japan Team

※この投稿は米国時間 2021 年 3 月 16 日に、Google Cloud blog に投稿されたものの抄訳です。

ML ワークフローでは、特定のモデルを 1 回トレーニングしデプロイするだけでは不十分なことがよくあります。最初はモデルの精度が望ましいレベルであっても、予測リクエストに使用されるデータが（おそらく、時間の経過に応じて）最初にモデルのトレーニングに使用したデータとまったく違うものになっている場合、精度が変わってくることがあります。

モデルの再トレーニングに使用できる新しいデータが登場した場合、データ「ドリフト」を分析する手法により、そのドリフトが再トレーニングを必要とする程度に異常であるかどうかを判断するとよいでしょう。

さらに、新しいデータが到着したときに、そのような分析（そして、必要に応じたトレーニングパイプラインの再実行）を自動的にトリガーできると便利です。

今回のブログ投稿では、Kubeflow Pipelines（KFP）を使用してこうしたシナリオに対応する方法を解説するサンプルノートブックを取り上げます。

このサンプルノートブックでは、データセットのバージョンが新しくなるたびにドリフトの統計情報をチェックし、その情報を使用してモデルの（再）トレーニングを行うかどうかを判断するパイプラインを構築する方法と、新しいデータが到着したときにパイプラインジョブのイベントドリブンデプロイを構成する方法を説明しています（今回のサンプルでは、新しいデータセットでのモデルの完全な再トレーニングをご紹介します。ここでは説明していませんが、別のシナリオとして、新しいデータで既存のモデルを調整するというものもあります）。

このノートブックの基盤となっているのは、KFP のトレーニングとサービングパイプラインについて説明した以前のブログ投稿で取り上げたサンプルです。これに加えて、主要なコンセプトが新たに 2 つ追加されています。

このサンプルでは、データセットの統計情報を取得し、新旧のバージョンのデータセット間のドリフトを検出するパイプラインコンポーネントを構築するための TensorFlow Data Validation（TFDV）ライブラリの使い方を説明しています。また、ドリフトに関する情報を使用して新しいデータでモデルを再トレーニングするかどうかを判断する方法も示しています。
このサンプルでは、Cloud Functions（GCF）の関数から Kubeflow Pipelines 実行の開始をイベントによってトリガーする方法を示しています。この関数の実行は、ファイルを特定の Cloud Storage（GCS）バケットに追加することでトリガーされます。

この機械学習タスクでは、表形式のデータセットを使用してロンドンの自転車レンタルに関する情報を気象データと結び付け、レンタル時間を予測するように Keras モデルをトレーニングします。データセットとモデルのアーキテクチャに関する背景情報については、こちらとこちらのブログ投稿、ならびに関連する README を参照してください。

https://storage.googleapis.com/gweb-cloudblog-publish/images/CleanShot_2021-02-26_at_11.30.132x.max-210.max-2100x2100.png

TFDV ベースのコンポーネントを使用して「データドリフト」を検出するパイプライン実行。

サンプルノートブックの実行

このサンプルノートブックには、Google Cloud Platform（GCP）のアカウントとプロジェクト、そして理想的には GPU を使用するための割り当てが必要です。また、ノートブックで詳しく説明されているように、AI Platform Pipelines（ホストされている Kubeflow Pipelines）のインストールも必要です（つまり、KFP を Google Kubernetes Engine（GKE）にインストールします）。インストールの完了後に、いくつかの構成を追加で行います。

このノートブックの実行には、Colab（直接開きます）または AI Platform Notebooks（直接開きます）を使用します。

TFDV ベースの KFP コンポーネントの作成

最初のステップは、パイプラインで使用する TFDV コンポーネントの構築です。

注: このサンプルでは、トレーニングデータは CSV 形式のファイルで GCS にあります。このため、CSV ファイルを処理する TFDV の機能を利用できます。TFDV ライブラリは TFRecords 形式のファイルも処理できます。

両方の TFDV KFP パイプラインコンポーネントを「軽量」な Python 関数ベースのコンポーネントとして定義します。各コンポーネントについて関数を定義してから、その関数で kfp.components.func_to_container_op() を呼び出し、.yaml 形式で再利用可能なコンポーネントを構築します。

この仕組みについて、もう少し見てみましょう（詳細については、ノートブックを参照）。

以下に示しているのは、一連の csv ファイルから TFDV 統計情報を生成するために使用する Python 関数です。この関数（そして、それから作成するコンポーネント）は、生成された統計情報ファイルへのパスを出力します。このコンポーネントを使用するパイプラインを定義するとき、このステップの出力を別のパイプラインステップへの入力として使用します。

TFDV は Beam パイプライン（KFP パイプラインとは別のもの）を使用して、統計情報の生成を実装します。構成に応じて、コンポーネントは Direct（ローカル）ランナーまたは Dataflow ランナーを使用できます。大きなデータセットの場合、ローカルではなく Dataflow で Beam パイプラインを実行するとよいでしょう。

読み込んでいます...

from typing import NamedTuple

def generate_tfdv_stats(input_data: str, output_path: str, job_name: str, use_dataflow: str,
                        project_id: str, region:str, gcs_temp_location: str, gcs_staging_location: str,
                        whl_location: str = '', requirements_file: str = 'requirements.txt'
) -> NamedTuple('Outputs', [('stats_path', str)]):

import logging
  import time

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

logging.getLogger().setLevel(logging.INFO)
  logging.info("output path: %s", output_path)
  logging.info("Building pipeline options")
  # Create and set your PipelineOptions.
  options = PipelineOptions()

if use_dataflow == 'true':
    logging.info("using Dataflow")
    if not whl_location:
      logging.warning('tfdv whl file required with dataflow runner.')
      exit(1)
    google_cloud_options = options.view_as(GoogleCloudOptions)
    google_cloud_options.project = project_id
    google_cloud_options.job_name = '{}-{}'.format(job_name, str(int(time.time())))
    google_cloud_options.staging_location = gcs_staging_location
    google_cloud_options.temp_location = gcs_temp_location
    google_cloud_options.region = region
    options.view_as(StandardOptions).runner = 'DataflowRunner'

setup_options = options.view_as(SetupOptions)
    setup_options.extra_packages = [whl_location]
    setup_options.requirements_file = 'requirements.txt'

tfdv.generate_statistics_from_csv(
    data_location=input_data, output_path=output_path,
    pipeline_options=options)

return (output_path, )

この関数を KFP コンポーネントに変換するには、kfp.components.func_to_container_op() を呼び出します。そして、gcr.io/google-samples/tfdv-tests:v1 のように、使用するベースコンテナイメージを渡します。このベースイメージにはすでに TFDV ライブラリがインストールされているため、このコンポーネントに基づいてパイプラインステップを実行する際に、それらを「インライン」でインストールする必要はありません。

読み込んでいます...

2 つ目の TFDV ベースのコンポーネントを構築する際も同じアプローチをとります。このコンポーネントは、統計情報を比較してデータセット間のドリフトを検出するものです。この場合、TFDV ライブラリを使用すると簡単です。回帰モデルに適したドリフトコンパレータを（サンプルのパイプラインで使用されているように）使用し、特定のフィールドセットでドリフトを探します（この場合は、サンプル用途のため 1 つだけです）。

その後、tensorflow_data_validation.validate_statistics() 呼び出しにより、そのフィールドのドリフトの異常が指定されたしきい値を超えているかどうかを判断します。詳細については、TFDV のドキュメントを参照してください。

読み込んでいます...

（この 2 つ目のコンポーネント定義の詳細は、サンプルノートブックに記載されています）。

TFDV コンポーネントを使用するパイプラインの定義

両方の TFDV コンポーネント（データセットの統計情報を生成するものと、データセット間のドリフトを検出するもの）を定義したら、これらのコンポーネントを使用する Kubeflow Pipeline を、トレーニングとサービングのワークフロー向けに構築済みのコンポーネントと関連付けて構築できます。

コンポーネントからパイプラインオペレーションをインスタンス化する

yaml 形式の KFP コンポーネントは共有可能かつ再利用可能です。パイプライン構築のベースとして、いくつかの構築済みのコンポーネント（詳細についてはこちらを参照）を使用します。これらは、基本的な「トレーニング / 評価 / デプロイ」ワークフローをサポートするものです。

既存のコンポーネントを URL 経由で読み込むことで、これらのコンポーネントから以下のようにパイプラインオペレーションをインスタンス化します。

読み込んでいます...

次に、定義されたオペレーションから KFP パイプラインを定義します。ここではパイプライン全体については示しません。詳細については、ノートブックを参照してください。

2 つのパイプラインステップは、統計情報を生成する tfdv_op に基づいています。tfdv1 はテストデータの統計情報、tfdv2 はトレーニングデータの統計情報を生成します。

以下では、tfdv_drift ステップ（tfdv_drift_op に基づくもの）が、tfdv2（トレーニングデータの統計情報）ステップからの出力を入力として受け入れています。

読み込んでいます...

@dsl.pipeline(
  name='bikes_weather_tfdv',
  description='Model bike rental duration given weather'
)
def bikes_weather_tfdv( 
  ... other pipeline params ...
  working_dir: str = 'gs://YOUR/GCS/PATH',  
  data_dir: str = 'gs://aju-dev-demos-codelabs/bikes_weather/', 
  project_id: str = 'YOUR-PROJECT-ID',
  region: str = 'us-central1',
  requirements_file: str = 'requirements.txt',
  job_name: str = 'test',
  whl_location: str = 'tensorflow_data_validation-0.26.0-cp37-cp37m-manylinux2010_x86_64.whl',
  use_dataflow: str = '',
  stats_older_path: str = 'gs://aju-dev-demos-codelabs/bikes_weather_chronological/evaltrain1.pb'
  ):
  ...

tfdv1 = tfdv_op(  # TFDV stats for the test data
    input_data='%stest-*.csv' % (data_dir,),
    output_path='%s/tfdv_expers/%s/eval/evaltest.pb' % (working_dir, dsl.RUN_ID_PLACEHOLDER),
    job_name='%s-1' % (job_name,),
    use_dataflow=use_dataflow,
    project_id=project_id, region=region,
    gcs_temp_location='%s/tfdv_expers/tmp' % (working_dir,), 
    gcs_staging_location='%s/tfdv_expers' % (working_dir,), 
    whl_location=whl_location, requirements_file=requirements_file
    )
  tfdv2 = tfdv_op(  # TFDV stats for the training data
    input_data='%strain-*.csv' % (data_dir,),
    # output_path='%s/%s/eval/evaltrain.pb' % (output_path, dsl.RUN_ID_PLACEHOLDER),
    output_path='%s/tfdv_expers/%s/eval/evaltrain.pb' % (working_dir, dsl.RUN_ID_PLACEHOLDER),
    job_name='%s-2' % (job_name,),
    use_dataflow=use_dataflow,
    project_id=project_id, region=region,
    gcs_temp_location='%s/tfdv_expers/tmp' % (working_dir,), 
    gcs_staging_location='%s/tfdv_expers' % (working_dir,), 
    whl_location=whl_location, requirements_file=requirements_file
    )

# compare generated training data stats with stats from a previous version
  # of the training data set.
  tfdv_drift = tfdv_drift_op(stats_older_path, tfdv2.outputs['stats_path'])
  
  # proceed with training if drift is detected (or if no previous stats were provided)
  with dsl.Condition(tfdv_drift.outputs['drift'] == 'true'):  
    train = train_op(...)
    eval_metrics = eval_metrics_op(...)
    with dsl.Condition(eval_metrics.outputs['deploy'] == 'deploy'):
      serve = serve_op(...)

パイプラインの詳細すべてが示されているわけではありませんが、このパイプライン定義にはいくつかの条件式が含まれていることがわかるでしょう。パイプラインの一部は、「上流」ステップの出力が特定の条件を満たした場合にのみ実行されます。ドリフトの異常が検出された場合、モデルのトレーニングステップを開始します（また、トレーニングが完了したら、その評価指標が特定のしきい値を満たす場合にのみサービング用のモデルをデプロイします）。

以下は、このパイプラインの DAG です。条件式が反映されています。また、テストデータセットの統計情報を生成するためのステップが下流の依存関係を一切提供していない一方で、トレーニングセットの統計情報がドリフト検出ステップ向けの入力として使用されていることがわかります。

https://storage.googleapis.com/gweb-cloudblog-publish/images/bw_tfdv_pipeline.max-1500x1500.max-1200x1200.png

パイプラインの DAG

以下に、進行中のパイプライン実行を示します。

https://storage.googleapis.com/gweb-cloudblog-publish/images/bw_tfdv_pipeline_run.max-2000x2000.max-2000x2000.png

進行中のパイプライン実行。

このパイプラインの実行方法の詳細については、サンプルノートブックを参照してください。

イベントによってトリガーされるパイプライン実行

このパイプラインを定義したら、次に、データセットの更新があったときに自動的にパイプラインを実行するようにすると便利です。これによって、データセットの更新があるたびに、データドリフトと必要に応じたモデルの（再）トレーニングの分析がトリガーされます。

Cloud Functions（GCF）を使用してこれを行う方法をご紹介します。新しいデータが GCS バケットに追加されたときにトリガーされる関数を設定します。

データセットが更新されたときにパイプライン実行をトリガーする GCF 関数を設定する

新しいトレーニングデータを利用できるようになった時点でこのパイプラインの実行を開始する Cloud Functions（GCF）の関数を定義し、デプロイします。GCS の「トリガー」バケットでファイルが作成または修正されるとトリガーされるようにします。

ほとんどの場合、新しいファイルがデータセットに追加されるたびに新しいパイプライン実行を開始する必要はありません。これは、一般に、データセットを構成するファイルのコレクションについては、複数のファイルを一括で追加または更新するためです。よって、「トリガーバケット」をデータセットのバケットにすることはおすすめしません（データが GCS にある場合）。これによって、不要なパイプライン実行がトリガーされてしまうためです。

代わりに、新しいデータのバッチのアップロードが完了した後にパイプライン実行をトリガーします。

このためには、「トリガー」バケットがデータセットファイルの保存に使用されるバケットとは異なるものとするアプローチをとります。このバケットにアップロードされた「トリガーファイル」には、更新されたデータセットのパスと、最後にトレーニングされたモデル向けに生成されたデータ統計情報ファイルへのパスが含まれているはずです。

トリガーファイルは新しいデータのアップロードが完了した後でアップロードされ、このアップロードにより GCF 関数の実行がトリガーされます。その後、この関数がトリガーファイルから新しいデータパスの情報を読み取り、パイプラインジョブを開始します。

GCF 関数を定義する

このプロセスを設定するにはまず、main.py というファイルと、関数の実行前に読み込むライブラリを指定する、同じディレクトリ内の関連要件ファイルで GCF 関数を定義します。この要件ファイルは、KFP SDK をインストールするように求めます。

kfp==1.4

コードは以下のようになります（一部の詳細は省略）。トリガーファイルのコンテンツを解析し、その情報を使用してパイプライン実行を開始します。このコードでは、GCF 関数のアップロード時に設定する複数の環境変数の値を使用します。

読み込んでいます...

import logging
import os

import kfp
from kfp import dsl
from kfp import compiler
from kfp import components

from google.cloud import storage

PIPELINE_PROJECT_ID = os.getenv('PIPELINE_PROJECT_ID')
...etc...

def read_trigger_file(data, context, storage_client):
    """Read the contents of the trigger file and return as string.
    """
    ....
    bucket = storage_client.get_bucket(data['bucket'])
    blob = bucket.get_blob(data['name'])
    trigger_file_string = blob.download_as_string().strip()
    logging.info('trigger file contents: {}'.format(trigger_file_string))
    return trigger_file_string.decode('UTF-8')

def gcs_update(data, context):
    """Background Cloud Function to be triggered by Cloud Storage.
    """

storage_client = storage.Client()
    # get the contents of the trigger file
    trigger_file_string = read_trigger_file(data, context, storage_client)
    trigger_file_info = trigger_file_string.strip().split('\n')
    # then run the pipeline using the given job spec, passing the trigger file contents
    # as parameter values.
    logging.info('running pipeline with id %s...', PIPELINE_ID)
    # create the client object
    client = kfp.Client(host=PIPELINE_HOST)    
    # deploy the pipeline run
    run = client.run_pipeline(EXP_ID, 'bw_tfdv_gcf', pipeline_id=PIPELINE_ID,
                          params={'working_dir': WORKING_DIR,
                                  'project_id': PIPELINE_PROJECT_ID,
                                  'use_dataflow': USE_DATAFLOW,
                                  'data_dir': trigger_file_info[0],
                                  'stats_older_path': trigger_file_info[1]})

logging.info('job response: %s', run)

続けて、以下のように GCF 関数をデプロイします。gcs_update 定義を（main.py から）使用し、トリガーバケットを指定している点に注意してください。また、デプロイの一部として環境変数を設定する方法にも注目してください。

読み込んでいます...

新しいデータを利用できるようになった時点でパイプライン実行をトリガーする

設定した GCF 関数は、ファイルがトリガーバケットに追加（またはトリガーバケットで修正）されたときに実行されます。シンプルな例を挙げると、GCF 関数は以下の形式のトリガーファイルを予期します。1 行目は更新されたデータセットへのパスで、2 行目は以前にトレーニングされたモデルに使用されるデータセットの TFDV 統計情報へのパスです。一般的には、このようなトリガーファイルには、パイプライン実行をパラメータ化する方法を決定するために必要なあらゆる情報を含めることができます。

読み込んでいます...

次のステップ

今回のブログ投稿では、TFDV ライブラリを使用して Kubeflow Pipeline コンポーネントを構築し、データセットの分析とデータドリフトの検出を行う方法をご紹介しました。また、Cloud Functions を介してイベントによってトリガーされるパイプライン実行をサポートする方法もご説明しました。

TFDV を使用して生成された統計情報を可視化し検討する方法については、この投稿では説明していませんが、こちらのサンプルノートブックでその方法が説明されています。

また、Kubeflow Pipelines GitHub リポジトリでサンプルを参照することもできます。

-スタッフデベロッパーアドボケイト Amy Unruh

投稿先