スタブ エグゼキュータを使用したパイプラインのテスト

導入

このチュートリアルを進めるには、 template.ipynbチュートリアルをステップ 6まで完了する必要があります。

このドキュメントでは、ゴールデン テスト データを使用して偽のアーティファクトを生成するBaseStubExecuctorを使用して TensorFlow Extended (TFX) パイプラインをテストする手順を説明します。これは、ユーザーがテストしたくないエグゼキュータを置き換えて、実際のエグゼキュータの実行時間を節約できるようにすることを目的としています。スタブ エグゼキュータは、TFX Python パッケージのtfx.experimental.pipeline_testing.base_stub_executorの下に提供されます。

このチュートリアルはtemplate.ipynbチュートリアルの拡張機能として機能するため、シカゴ市がリリースしたTaxi Trips データセットも使用します。スタブ エグゼキュータを使用する前に、コンポーネントを変更してみることを強くお勧めします。

1. パイプライン出力を Google Cloud Storage に記録します。

まず、スタブ エグゼキューターが記録された出力からアーティファクトをコピーできるように、パイプライン出力を記録する必要があります。

このチュートリアルでは、 template.ipynbがステップ 6 まで完了していることを前提としているため、成功したパイプラインの実行がMLMDに保存されている必要があります。 MLMD の実行情報には、gRPC サーバーを使用してアクセスできます。

ターミナルを開き、次のコマンドを実行します。

  1. 適切な認証情報を使用して kubeconfig ファイルを生成します。 bash gcloud container clusters get-credentials $cluster_name --zone $compute_zone --project $gcp_project_id $compute_zoneは gcp エンジンのリージョン、 $gcp_project_idは GCP プロジェクトのプロジェクト ID です。

  2. MLMD に接続するためのポート転送を設定します。 bash nohup kubectl port-forward deployment/metadata-grpc-deployment -n $namespace $port:8080 & $namespaceはクラスターの名前空間で、$port はクラスターの名前空間で、 $portクラスターの名前空間です。ポート転送。

  3. tfx GitHub リポジトリのクローンを作成します。 tfx ディレクトリ内で、次のコマンドを実行します。 

python tfx/experimental/pipeline_testing/pipeline_recorder.py \
--output_dir=gs://<gcp_project_id>-kubeflowpipelines-default/testdata \
--host=$host \
--port=$port \
--pipeline_name=$pipeline_name

$output_dir 、パイプライン出力が記録される Google Cloud Storage 内のパスに設定する必要があるため、 <gcp_project_id>を GCP プロジェクト ID に置き換えてください。

$hostと$port 、MLMD に接続するメタデータ grpc サーバーのホスト名とポートです。 $portポート転送に使用したポート番号に設定する必要があり、ホスト名には「localhost」を設定できます。

template.ipynbチュートリアルでは、パイプライン名はデフォルトで「my_pipeline」に設定されているため、 pipeline_name="my_pipeline"設定します。テンプレート チュートリアルの実行時にパイプライン名を変更した場合は、それに応じて--pipeline_nameを変更する必要があります。

2. Kubeflow DAG Runner でスタブ エグゼキュータを有効にする

まず、 tfx template copy CLI コマンドを使用して、事前定義されたテンプレートがプロジェクト ディレクトリにコピーされていることを確認します。コピーしたソースファイルのうち、以下の2つのファイルを編集する必要があります。

  1. kubeflow_dag_runner.py が配置されているディレクトリにstub_component_launcher.pyというファイルを作成し、次の内容をそこに配置します。

    from tfx.experimental.pipeline_testing import base_stub_component_launcher
from pipeline import configs

class StubComponentLauncher(
    base_stub_component_launcher.BaseStubComponentLauncher):
  pass

# GCS directory where KFP outputs are recorded
test_data_dir = "gs://{}/testdata".format(configs.GCS_BUCKET_NAME)
# TODO: customize self.test_component_ids to test components, replacing other
# component executors with a BaseStubExecutor.
test_component_ids = ['Trainer']
StubComponentLauncher.initialize(
    test_data_dir=test_data_dir,
    test_component_ids=test_component_ids)

  2. コンポーネント ID をテスト対象のコンポーネント ID のリストに設定します (つまり、他のコンポーネントのエグゼキューターは BaseStubExecutor に置き換えられます)。

  3. kubeflow_dag_runner.pyを開きます。先ほど追加したStubComponentLauncherクラスを使用するには、次の import ステートメントを先頭に追加します。

    import stub_component_launcher

  4. kubeflow_dag_runner.pyで、 StubComponentLauncherクラスをKubeflowDagRunnerConfigのsupported_launcher_classに追加して、スタブ エグゼキューターの起動を有効にします。

    runner_config = kubeflow_dag_runner.KubeflowDagRunnerConfig(
    supported_launcher_classes=[
        stub_component_launcher.StubComponentLauncher
    ],

3. スタブエグゼキュータを使用してパイプラインを更新して実行します

スタブ エグゼキューターを使用して、変更されたパイプライン定義で既存のパイプラインを更新します。

tfx pipeline update --pipeline-path=kubeflow_dag_runner.py \
  --endpoint=$endpoint --engine=kubeflow

$endpoint KFP クラスターのエンドポイントに設定する必要があります。

次のコマンドを実行して、更新されたパイプラインの新しい実行実行を作成します。

tfx run create --pipeline-name $pipeline_name --endpoint=$endpoint \
  --engine=kubeflow

掃除中

コマンドfg使用してバックグラウンドでポート転送にアクセスし、ctrl-C を押して終了します。 gsutil -m rm -R $output_dirを使用して、記録されたパイプライン出力が含まれるディレクトリを削除できます。

このプロジェクトで使用されているすべての Google Cloud リソースをクリーンアップするには、チュートリアルに使用したGoogle Cloud プロジェクトを削除します。

あるいは、次の各コンソールにアクセスして、個々のリソースをクリーンアップすることもできます。 - Google Cloud Storage - Google Container Registry - Google Kubernetes Engine