テストにパイプライン実行を追加する

Google Cloud コンソール

1. Google Cloud コンソールの [Vertex AI] セクションで、[パイプライン] ページに移動します。
   [パイプライン] に移動
2. [リージョン] プルダウン リストで、パイプラインを作成するリージョンを選択します。
3. [add_box 実行を作成] をクリックして、[パイプライン実行の作成] ペインを開きます。
4. [実行の詳細] に次の内容を指定します。
   • [ファイル] フィールドで [選択] をクリックして、ファイル選択ツールを開きます。 実行するコンパイル済みパイプライン JSON ファイルに移動し、パイプラインを選択して [開く] をクリックします。
   • デフォルトでは、[パイプライン名] にパイプラインの定義で指定した名前が使用されます。必要に応じて、別の [パイプライン名] を指定します。
   • このパイプライン実行を一意に識別できるように実行名を指定します。
5. このパイプラインの実行にカスタム サービス アカウント、顧客管理の暗号鍵、またはピアリングされる VPC ネットワークを使用するように指定するには、[ADVANCED OPTIONS] をクリックします（省略可）。
   詳細オプション（カスタム サービス アカウントなど）を構成する手順は次のとおりです。
   • サービス アカウントを指定するには、[サービス アカウント] プルダウン リストからサービス アカウントを選択します。
     サービス アカウントを指定しない場合、Vertex AI Pipelines はデフォルトの Compute Engine サービス アカウントを使用してパイプラインを実行します。
     詳細については、Vertex AI Pipelines で使用するサービス アカウントの構成をご覧ください。
   • 顧客管理の暗号鍵（CMEK）を使用するには、[顧客管理の暗号鍵を使用する] を選択します。[顧客管理の暗号鍵を選択] プルダウン リストが表示されます。[顧客管理の暗号鍵を選択] プルダウン リストで、使用する鍵を選択します。
   • このパイプライン実行でピアリングされた VPC ネットワークを使用するには、[ピアリングされた VPC ネットワーク] ボックスに VPC ネットワーク名を入力します。
6. [続行] をクリックします。
   [Cloud Storage] のロケーションと [パイプライン パラメータ] ペインが表示されます。
• 必須: Cloud Storage の出力ディレクトリを入力します（例: gs://location_of_directory）。
• （省略可）このパイプライン実行に使用するパラメータを指定します。
7. [送信] をクリックして、パイプライン実行を作成します。
8. パイプラインが送信されると、パイプラインの Google Cloud コンソール テーブルに表示されます。
9. パイプラインに関連付けられた行で、more_vert（もっと見る > [テストに追加]）をクリックします。
• 既存のテストを選択するか、新しく作成します。
• 省略可: テスト実行がテストに関連付けられている場合は、プルダウンに表示されます。既存のテスト実行を選択します。
10. [保存] をクリックします。

Google Cloud コンソールでパイプライン実行とテスト実行を比較する

1. Google Cloud コンソールで、[テスト] ページに移動します。
   [テスト] に移動
   [テスト] ページにテストのリストが表示されます。
2. パイプライン実行を追加するテストを選択します。
   実行のリストが表示されます。
3. 比較する実行を選択して、[比較] をクリックします。
   
4. [実行を追加] ボタンをクリックします。実行のリストが表示されます。
5. 追加するパイプライン実行を選択します。実行が追加されます。

Vertex AI SDK for Python {:#sdk-add-pipeline-run}

以下のサンプルでは、PipelineJob API を使用しています。

from typing import Any, Dict, Optional

from google.cloud import aiplatform


def log_pipeline_job_to_experiment_sample(
    experiment_name: str,
    pipeline_job_display_name: str,
    template_path: str,
    pipeline_root: str,
    project: str,
    location: str,
    parameter_values: Optional[Dict[str, Any]] = None,
):
    aiplatform.init(project=project, location=location)

    pipeline_job = aiplatform.PipelineJob(
        display_name=pipeline_job_display_name,
        template_path=template_path,
        pipeline_root=pipeline_root,
        parameter_values=parameter_values,
    )

    pipeline_job.submit(experiment=experiment_name)

from google.cloud import aiplatform


def log_pipeline_job_sample(
    experiment_name: str,
    run_name: str,
    pipeline_job: aiplatform.PipelineJob,
    project: str,
    location: str,
):
    aiplatform.init(experiment=experiment_name, project=project, location=location)

    aiplatform.start_run(run=run_name, resume=True)

    aiplatform.log(pipeline_job=pipeline_job)