パイプライン テンプレートの作成、アップロード、使用

パイプライン テンプレートは、ワークフロー定義を公開するためのリソースです。1 人のユーザーまたは複数のユーザーで複数回再利用できます。

Kubeflow Pipelines SDK レジストリ クライアントは、Artifact Registry などの互換性のあるレジストリ サーバーで使用できる新しいクライアント インターフェースで、Kubeflow Pipelines(KFP)テンプレートのバージョン管理に使用できます。詳細については、Kubeflow Pipelines SDK レジストリ クライアントでテンプレートを使用するをご覧ください。

このページでは、次の方法について説明します。

  • KFP パイプライン テンプレートを作成する
  • Kubeflow Pipelines SDK レジストリ クライアントを使用して、テンプレートをパイプライン テンプレート リポジトリにアップロードする
  • Kubeflow Pipelines クライアントでテンプレートを使用する

始める前に

パイプラインを構築して実行する前に、Google Cloud コンソールを使用して、次の手順で Google Cloud プロジェクトと開発環境を設定します。

  1. v2 以降の Kubeflow Pipelines SDK をインストールします。
    (省略可)インストールする前に、次のコマンドを実行して、現在インストールされている Kubeflow Pipelines SDK のバージョンを確認します。

      pip freeze | grep kfp
    
  2. v1.15.0 以降の Vertex AI SDK for Python をインストールします。
    (省略可)インストールする前に、次のコマンドを実行して、現在インストールされている Vertex AI SDK for Python のバージョンを確認します。

      pip freeze | grep google-cloud-aiplatform
    
  3. (省略可)390.0.0 以降の Google Cloud CLI をインストールします。

  4. Artifact Registry API を有効にします

権限の構成

Vertex AI Pipelines 用の gcloud CLI プロジェクトをまだ設定していない場合は、Vertex AI Pipelines 用の Google Cloud プロジェクトの構成の手順を行います。

さらに、Artifact Registry をテンプレート レジストリとして使用するために、事前定義された次の Identity and Access Management 権限を割り当ててください。

  • roles/artifactregistry.admin: リポジトリの作成と管理を行うには、このロールを割り当てます。
  • roles/artifactregistry.repoAdmin または roles/artifactregistry.writer: リポジトリ内のテンプレートを管理するには、これらのロールのいずれかを割り当てます。
  • roles/artifactregistry.reader: リポジトリからテンプレートをダウンロードするには、このロールを割り当てます。
  • roles/artifactregistry.reader: テンプレートからパイプライン実行を作成するには、このロールを Vertex AI Pipelines に関連付けられたサービス アカウントに割り当てます。

Artifact Registry の Identity and Access Management 事前定義ロールの詳細については、Artifact Registry 事前定義ロールをご覧ください。

次のサンプルを使用してロールを割り当てます。

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=PRINCIPAL \
    --role=ROLE

次のように置き換えます。

  • PROJECT_ID: パイプラインを作成するプロジェクト。
  • PRINCIPAL: 権限を追加するプリンシパル。
  • ROLE: プリンシパルに付与する Identity and Access Management ロール。

次の詳細については、Artifact Registry のドキュメントロールと権限をご覧ください。

Artifact Registry にリポジトリを作成する

次に、Artifact Registry でパイプライン テンプレートのリポジトリを作成します。

コンソール

  1. Google Cloud コンソールで Vertex AI の [パイプライン] を開きます。

    Vertex AI の [パイプライン] に移動

  2. [テンプレート] タブをクリックします。

  3. [リポジトリを選択] ペインを開くには、[リポジトリを選択] をクリックします。

  4. [リポジトリを作成] をクリックします。

  5. リポジトリ名として quickstart-kfp-repo を指定します。

  6. [形式] で [Kubeflow Pipelines] を選択します。

  7. [ロケーション タイプ] で [リージョン] を選択します。

  8. [リージョン] プルダウン リストで、us-central1 を選択します。

  9. [作成] をクリックします。

Google Cloud CLI

次のコマンドを実行してリポジトリを作成します。

後述のコマンドデータを使用する前に、次のように置き換えます。

  • LOCATION: リポジトリを作成するロケーションまたはリージョン(例: us-central1

gcloud artifacts repositories create コマンドを実行します。

Linux、macOS、Cloud Shell

gcloud artifacts repositories create quickstart-kfp-repo --location=LOCATION_ID --repository-format=KFP

Windows(PowerShell)

gcloud artifacts repositories create quickstart-kfp-repo --location=LOCATION_ID --repository-format=KFP

Windows(cmd.exe)

gcloud artifacts repositories create quickstart-kfp-repo --location=LOCATION_ID --repository-format=KFP
 

テンプレートを作成する

次のコードサンプルを使用して、単一のコンポーネントを含むパイプラインを定義します。KFP を使用してパイプラインを定義する方法については、パイプラインの構築をご覧ください。

from kfp import dsl
from kfp import compiler

@dsl.component()
def hello_world(text: str) -> str:
    print(text)
    return text

@dsl.pipeline(name='hello-world', description='A simple intro pipeline')
def pipeline_hello_world(text: str = 'hi there'):
    """Pipeline that passes small pipeline parameter string to consumer op."""

    consume_task = hello_world(
        text=text)  # Passing pipeline parameter as argument to consumer op

compiler.Compiler().compile(
    pipeline_func=pipeline_hello_world,
    package_path='hello_world_pipeline.yaml')

サンプルを実行すると、compiler.Compiler().compile(...) ステートメントによって「hello-world」パイプラインが hello_world_pipeline.yaml という名前のローカル YAML ファイルにコンパイルされます。

テンプレートをアップロードする

コンソール

  1. Google Cloud コンソールで Vertex AI の [パイプライン] を開きます。

    Vertex AI の [パイプライン] に移動

  2. [アップロード] をクリックして、[パイプラインまたはコンポーネントのアップロード] ペインを開きます。

  3. [リポジトリ] プルダウン リストで、quickstart-kfp-repo リポジトリを選択します。

  4. パイプライン テンプレートの名前を指定します。

  5. [ファイル] フィールドで [選択] をクリックして、コンパイル済みのパイプライン テンプレート YAML をローカル ファイル システムから選択してアップロードします。

  6. パイプライン テンプレートをアップロードすると、[テンプレート] ページに表示されます。

    [テンプレート] に移動

Kubeflow Pipelines SDK クライアント

  1. Kubeflow Pipelines SDK レジストリ クライアントを構成するには、次のコマンドを実行します。

    from kfp.registry import RegistryClient
    
    client = RegistryClient(host=f"https://us-central1-kfp.pkg.dev/PROJECT_ID/quickstart-kfp-repo")
    
  2. コンパイルされた YAML ファイルを Artifact Registry のリポジトリにアップロードします。

    templateName, versionName = client.upload_pipeline(
      file_name="hello_world_pipeline.yaml",
      tags=["v1", "latest"],
      extra_headers={"description":"This is an example pipeline template."})
    
  3. テンプレートがアップロードされたことを確認するには:

    1. Google Cloud コンソールで Vertex AI の [パイプライン] を開きます。

      Vertex AI の [パイプライン] に移動

    2. [テンプレート] タブをクリックします。

    3. [リポジトリを選択] をクリックします。

    4. リストから quickstart-kfp-repo リポジトリを選択し、[選択] をクリックします。

    5. アップロードされたテンプレート パッケージ hello-world がリストに表示されます。

    6. パイプライン テンプレートのバージョンのリストを表示するには、hello-world テンプレートをクリックします。

    7. パイプライン トポロジを表示するには、バージョンをクリックします。

Vertex AI でテンプレートを使用する

パイプライン テンプレートを Artifact Registry のリポジトリにアップロードすると、Vertex AI Pipelines で使用する準備が整います。

テンプレートのステージング バケットを作成する

パイプライン テンプレートを使用する前に、パイプライン実行をステージングするための Cloud Storage バケットを作成する必要があります。

バケットを作成するには、パイプライン アーティファクト用の Cloud Storage バケットを構成するの手順を行い、次のコマンドを実行します。

STAGING_BUCKET="gs://BUCKET_NAME"

BUCKET_NAME は、作成したバケットの名前に置き換えます。

テンプレートからパイプライン実行を作成する

Vertex AI SDK for Python または Google Cloud コンソールを使用して、Artifact Registry のテンプレートから実行するパイプラインを作成します。

コンソール

  1. Google Cloud コンソールで Vertex AI の [パイプライン] を開きます。

    Vertex AI の [パイプライン] に移動

  2. [テンプレート] タブをクリックします。

  3. [リポジトリを選択] ペインを開くには、[リポジトリを選択] をクリックします。

  4. quickstart-kfp-repo リポジトリを選択し、[選択] をクリックします。

  5. hello-world パッケージをクリックします。

  6. 4f245e8f9605 バージョンの横にある [実行を作成] をクリックします。

  7. [ランタイムの構成] をクリックします。

  8. [Cloud Storage のロケーション] に次のように入力します。

    gs://BUCKET_NAME
    

    BUCKET_NAME は、パイプライン実行のステージング用に作成したバケットの名前に置き換えます。

  9. [送信] をクリックします。

Vertex AI SDK for Python

次のサンプルを使用して、テンプレートからパイプライン実行を作成します。

from google.cloud import aiplatform

# Initialize the aiplatform package
aiplatform.init(
    project="PROJECT_ID",
    location='us-central1',
    staging_bucket=STAGING_BUCKET)

# Create a pipeline job using a version ID.
job = aiplatform.PipelineJob(
    display_name="hello-world-latest",
    template_path="https://us-central1-kfp.pkg.dev/PROJECT_ID/quickstart-kfp-repo/hello-world@SHA256_TAG" + \
        versionName)

# Alternatively, create a pipeline job using a tag.
job = aiplatform.PipelineJob(
    display_name="hello-world-latest",
    template_path="https://us-central1-kfp.pkg.dev/PROJECT_ID/quickstart-kfp-repo/hello-world/TAG")

job.submit()

次のように置き換えます。

  • PROJECT_ID: このパイプラインが実行される Google Cloud プロジェクト。

  • SHA256_TAG: テンプレート バージョンの sha256 ハッシュ値。

  • TAG: テンプレートのバージョンタグ。

作成したパイプライン実行を表示する

特定のパイプライン バージョンによって作成された実行を、Vertex AI SDK for Python で表示します。

コンソール

  1. Google Cloud コンソールで Vertex AI の [パイプライン] を開きます。

    Vertex AI の [パイプライン] に移動

  2. [テンプレート] タブをクリックします。

  3. [リポジトリを選択] をクリックします。

  4. リストから quickstart-kfp-repo リポジトリを選択し、[選択] をクリックします。

  5. hello-world パイプライン テンプレートのバージョンのリストを表示するには、hello world テンプレートをクリックします。

  6. パイプライン実行を表示するバージョンをクリックします。

  7. 選択したバージョンのパイプライン実行を表示するには、[実行を表示] をクリックして、[実行] タブをクリックします。

Vertex AI SDK for Python

パイプライン実行を一覧表示するには、次の例のように pipelineJobs.list コマンドを実行します。

  from google.cloud import aiplatform

  # To filter all runs created from a specific version
  filter = 'template_uri:"https://us-central1-kfp.pkg.dev/PROJECT_ID/quickstart-kfp-repo/hello-world/*" AND ' + \
           'template_metadata.version="%s"' % versionName
  aiplatform.PipelineJob.list(filter=filter)

  # To filter all runs created from a specific version tag
  filter = 'template_uri="https://us-central1-kfp.pkg.dev/PROJECT_ID/quickstart-kfp-repo/hello-world/latest"'
  aiplatform.PipelineJob.list(filter=filter)

  # To filter all runs created from a package
  filter = 'template_uri:"https://us-central1-kfp.pkg.dev/PROJECT_ID/quickstart-kfp-repo/hello-world/*"'
  aiplatform.PipelineJob.list(filter=filter)

  # To filter all runs created from a repo
  filter = 'template_uri:"https://us-central1-kfp.pkg.dev/PROJECT_ID/quickstart-kfp-repo/*"'
  aiplatform.PipelineJob.list(filter=filter)

Kubeflow Pipelines SDK レジストリ クライアントでテンプレートを使用する

Kubeflow Pipelines SDK レジストリ クライアントと Artifact Registry を併用して、パイプライン テンプレートをダウンロードして使用できます。

  • リポジトリ内のリソースを一覧表示するには、次のコマンドを実行します。

    templatePackages = client.list_packages()
    templatePackage = client.get_package(package_name = "hello-world")
    
    versions = client.list_versions(package_name="hello-world")
    version = client.get_version(package_name="hello-world", version=versionName)
    
    tags = client.list_tags(package_name = "hello-world")
    tag = client.get_tag(package_name = "hello-world", tag="latest")
    

    使用可能なメソッドとドキュメントの完全なリストについては、Artifact Registry GitHub リポジトリproto ファイルをご覧ください。

  • テンプレートをローカル ファイル システムにダウンロードするには、次のコマンドを実行します。

    # Sample 1
    filename = client.download_pipeline(
      package_name = "hello-world",
      version = versionName)
    # Sample 2
    filename = client.download_pipeline(
      package_name = "hello-world",
      tag = "v1")
    # Sample 3
    filename = client.download_pipeline(
      package_name = "hello-world",
      tag = "v1",
      file_name = "hello-world-template.yaml")
    

Artifact Registry REST API を使用する

以下では、Artifact Registry REST API を使用して Artifact Registry リポジトリのパイプライン テンプレートを管理する方法について説明します。

Artifact Registry REST API を使用してパイプライン テンプレートをアップロードする

パイプライン テンプレートをアップロードするには、このセクションで説明するパラメータ値を使用して HTTP リクエストを作成します。ここで

  • PROJECT_ID は、このパイプラインが実行される Google Cloud プロジェクトです。
  • REPO_ID は Artifact Registry リポジトリの ID です。

curl リクエストの例

curl -X POST \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    -F tags=v1,latest \
    -F content=@pipeline_spec.yaml \
    https://us-central1-kfp.pkg.dev/PROJECT_ID/REPO_ID

アップロード リクエストの作成

リクエストは、HTTP または HTTPS マルチパート リクエストです。リクエスト ヘッダーに認証トークンを含める必要があります。詳細については、gcloud auth print-access-token をご覧ください。

リクエストのペイロードは、pipeline_spec.yaml ファイル(または .zip パッケージ)の内容になります。推奨されるサイズの上限は 10 MiB です。

パッケージ名は、pipeline_spec.yaml ファイルの pipeline_spec.pipeline_info.name エントリから取得されます。パッケージ名はパッケージを一意に識別するもので、バージョンが変わっても変更されません。4~128 文字の長さで、^[a-z0-9][a-z0-9-]{3,127}$ という正規表現と一致する必要があります。

パッケージ tags は、最大 8 つのタグを含むカンマ区切りリストです。各タグは、正規表現 ^[a-zA-Z0-9\-._~:@+]{1,128}$ と一致する必要があります。

タグが存在し、すでにアップロードされているパイプラインを参照している場合は、現在アップロードされているパイプラインを参照するようにタグが更新されます。たとえば、latest タグがアップロード済みのパイプラインを参照していて、--tag=latest で新しいバージョンをアップロードすると、以前にアップロードされたパイプラインから latest タグが削除され、同じタグがアップロード中の新しいパイプラインに割り当てられます。

アップロードするパイプラインが、以前にアップロードしたパイプラインと同一である場合、アップロードは成功します。アップロードされたパイプラインのメタデータ(バージョンタグを含む)は、アップロード リクエストのパラメータ値と一致するように更新されます。

アップロード レスポンス

アップロード リクエストが成功した場合は、HTTP OK ステータスが返されます。レスポンスの本文は次のとおりです。

{packageName}/{versionName=sha256:abcdef123456...}

ここで、versionName は 16 進数の文字列としてフォーマットされた pipeline_spec.yaml の sha256 ダイジェストです。

Artifact Registry REST API を使用してパイプライン テンプレートをダウンロードする

パイプライン テンプレートをダウンロードするには、このセクションで説明するパラメータ値を使用して HTTP リクエストを作成します。

  • PROJECT_ID は、このパイプラインが実行される Google Cloud プロジェクトです。
  • REPO_ID は Artifact Registry リポジトリの ID です。
  • PACKAGE_ID は、アップロードされたテンプレートのパッケージ ID です。
  • TAG は、バージョンタグです。
  • VERSION は、sha256:abcdef123456... 形式のテンプレート バージョンです。

標準の Artifact Registry のダウンロードでは、次のようにダウンロード リンクを作成する必要があります。

url = https://us-central1-kfp.pkg.dev/PROJECT_ID/REPO_ID/PACKAGE_ID/VERSION
url = https://us-central1-kfp.pkg.dev/PROJECT_ID/REPO_ID/PACKAGE_ID/TAG

curl リクエストの例

curl -X GET \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    https://us-central1-kfp.pkg.dev/PROJECT_ID/REPO_ID/PACKAGE_ID/VERSION

次の例のように、VERSIONTAG に置き換えると、同じテンプレートをダウンロードできます。

curl -X GET \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    https://us-central1-kfp.pkg.dev/PROJECT_ID/REPO_ID/PACKAGE_ID/TAG

ダウンロード レスポンス

ダウンロード リクエストが成功した場合は、HTTP OK ステータスが返されます。レスポンスの本文は pipeline_spec.yaml ファイルの内容です。