パイプラインの実行

Vertex AI Pipelines を使用すると、Kubeflow Pipelines SDK または TensorFlow Extended を使用して構築された ML パイプラインをサーバーレスで実行できます。このドキュメントでは、ML パイプラインの実行方法について説明します。

テンプレート ギャラリーで事前に作成されたテンプレートを使用して、パイプライン実行を作成することもできます。テンプレート ギャラリーについて詳しくは、テンプレート ギャラリーの既成のテンプレートを使用するをご覧ください。

始める前に

Vertex AI Pipelines でパイプラインを実行する前に、次の手順で Google Cloud プロジェクトと開発環境を設定します。

  1. パイプラインを構築します。

  2. Vertex AI SDK for Python を使用してパイプラインを実行するには、Vertex SDK をインストールします。

パイプライン実行を作成する

Google Cloud Console または Python を使用して ML パイプラインを実行する手順は次のとおりです。

コンソール

Google Cloud コンソールを使用して ML パイプラインを実行するには、次の手順を行います。

  1. Google Cloud コンソールの [Vertex AI] セクションで、[パイプライン] ページに移動します。

    [パイプライン] に移動

  2. [リージョン] プルダウン リストで、パイプライン実行を作成するリージョンを選択します。

  3. [ 実行を作成] をクリックして、[パイプライン実行の作成] ペインを開きます。

  4. [実行の詳細] セクションで、次の操作を行います。

    1. [実行のソース] をクリックします。次のオプションが用意されています。

      • 既存のパイプラインから選択: 既存のパイプライン テンプレートに基づいてパイプライン実行を作成するには、[既存のパイプラインから選択] をクリックし、次の詳細を入力します。

        1. パイプラインまたはコンポーネント定義ファイルを含む [リポジトリ] を選択します。

        2. [パイプラインまたはコンポーネント] と [バージョン] を選択します。

        3. パイプライン実行を一意に識別できるように [実行名] を指定します。

      • テンプレート ギャラリーのパイプラインを選択: テンプレート ギャラリーの Google が作成したパイプライン テンプレートに基づいてパイプライン実行を作成するには、[テンプレート ギャラリーのパイプラインを選択する] をクリックし、次の詳細を入力します。

        1. [テンプレート ギャラリーのパイプライン] リストで、パイプライン テンプレートを選択します。

        2. 省略可: パイプライン実行を一意に識別するデフォルトの実行名を変更します。

      • ファイルをアップロード: コンパイル済みのパイプライン定義をアップロードするには、[ファイルをアップロード] をクリックして、次の詳細を入力します。

        1. [参照] をクリックしてファイル セレクタを開きます。実行するコンパイル済みパイプライン YAML ファイルに移動し、パイプラインを選択して [開く] をクリックします。

        2. [パイプライン名またはコンポーネント名] には、デフォルトでパイプライン定義で指定された名前が表示されます。必要に応じて、別のパイプライン名を指定します。

        3. パイプライン実行を一意に識別できるように [実行名] を指定します。

      • Cloud Storage からインポート: Cloud Storage からパイプライン定義ファイルをインポートするには、[Cloud Storage からインポート] をクリックして、次の詳細を入力します。

        1. [参照] をクリックしてパイプライン定義オブジェクトを含む Cloud Storage バケットに移動し、ファイルを選択して [選択] をクリックします。

        2. [パイプラインまたはコンポーネントの名前] を指定します。

        3. パイプライン実行を一意に識別できるように [実行名] を指定します。

    2. 省略可: 繰り返しのパイプライン実行をスケジュールするには、次のように [実行スケジュール] を指定します。

      1. [繰り返し] を選択します。

      2. [開始時間] で、スケジュールがアクティブになる時間を指定します。

        • スケジュール作成の直後に初回実行をスケジュールするには、[今すぐ] を選択します。

        • 最初の実行を特定の日時に行うようにスケジュールするには、[オン] を選択します。

      3. [頻度] フィールドで、unix-cron に基づく Cron スケジュール式を使用して、パイプラインの実行をスケジュールして実行する頻度を指定します。

      4. [終了] で、スケジュールを終了するタイミングを指定します。

        • スケジュールによってパイプラインが無期限に実行されることを示すには、[無期限] を選択します。

        • スケジュールが特定の日時で終了することを示すには、[オン] を選択し、スケジュールの終了日時を指定します。

      5. 省略可: パイプラインの実行にカスタム サービス アカウント、顧客管理の暗号鍵（CMEK）、ピアリングされた VPC ネットワークのいずれかを指定するには、[詳細オプション] をクリックして次の操作を行います。

        • サービス アカウントを指定するには、[サービス アカウント] プルダウン リストからサービス アカウントを選択します。

          サービス アカウントを指定しない場合、Vertex AI Pipelines はデフォルトの Compute Engine サービス アカウントを使用してパイプラインを実行します。

          詳細については、Vertex AI Pipelines で使用するサービス アカウントの構成をご覧ください。

        • CMEK を使用するには、[顧客管理の暗号鍵を使用] を選択します。[顧客管理の暗号鍵を選択] プルダウン リストが表示されます。[顧客管理の暗号鍵を選択] プルダウン リストで、使用する鍵を選択します。

        • このパイプライン実行でピアリングされた VPC ネットワークを使用するには、[ピアリングした VPC ネットワーク] ボックスに VPC ネットワーク名を入力します。

    3. [続行] をクリックします。

  5. [ランタイムの構成] セクションで、次のようにパイプライン実行を構成します。

    1. [Cloud Storage のロケーション] で [参照] をクリックし、パイプライン出力アーティファクトを保存する Cloud Storage バケットを選択して、[選択] をクリックします。

    2. 省略可: パイプライン実行に失敗ポリシーとキャッシュを構成するには、[詳細オプション] をクリックして次の操作を行います。

      • [失敗ポリシー] で、パイプライン全体の失敗ポリシーを指定します。パイプラインの障害ポリシーの詳細を確認する。

        • 1 つのタスクが失敗した後もタスクのスケジュール設定を続行するようにパイプラインを構成するには、[すべてのステップを完了まで実行する] を選択します。このオプションは、デフォルトで選択されています。

        • 1 つのタスクが失敗するとパイプラインが失敗するように構成するには、[1 つのステップが失敗したら、すぐにこの実行を失敗扱いにする] を選択します。

      • [キャッシュの構成] で、パイプライン全体のキャッシュ構成を指定します。

        • パイプラインのタスクにタスクレベルのキャッシュ構成を使用するには、[タスクレベルのキャッシュ構成をオーバーライドしない] を選択します。

        • パイプライン内のすべてのタスクのキャッシュを有効にして、タスクレベルのキャッシュ構成をオーバーライドするには、[すべてのステップでキャッシュからの読み取りを有効にする（最短時間）] を選択します。

        • パイプライン内のすべてのタスクのキャッシュを無効にして、タスクレベルのキャッシュ構成をオーバーライドするには、[すべてのステップでキャッシュからの読み取りを無効にする（最短時間）] を選択します。

    3. 省略可: パイプラインにパラメータがある場合は、[パイプライン パラメータ] で、パイプライン実行パラメータを指定します。

  6. パイプライン実行を作成するには、[送信] をクリックします。

Vertex AI SDK for Python

Vertex AI SDK for Python を使用して ML パイプラインを実行する手順は次のとおりです。次のコードサンプルを実行する前に、認証を設定する必要があります。

認証を設定する

認証を設定するには、サービス アカウント キーを作成して、サービス アカウント キーへのパスを示す環境変数を設定する必要があります。

  1. サービス アカウントを作成します。

    1. Google Cloud コンソールで [サービス アカウントの作成] ページに移動します。

      [サービス アカウントの作成] に移動

    2. [サービス アカウント名] フィールドに名前を入力します。
    3. 省略可: [サービス アカウントの説明] フィールドに説明を入力します。
    4. [作成] をクリックします。
    5. [ロールを選択] フィールドをクリックします。[すべてのロール] で、[Vertex AI] > [Vertex AI ユーザー] を選択します。

    6. [完了] をクリックして、サービス アカウントを作成します。

      ブラウザ ウィンドウは閉じないでください。次のステップでこれを使用します。

  2. 認証に使用するサービス アカウント キーを作成します。

    1. Google Cloud コンソールで、作成したサービス アカウントのメールアドレスをクリックします。
    2. [キー] をクリックします。
    3. [鍵を追加]、[新しい鍵を作成] の順にクリックします。
    4. [作成] をクリックします。JSON キーファイルがパソコンにダウンロードされます。
    5. [閉じる] をクリックします。
  3. 新しいサービス アカウントに、パイプラインの実行に使用するサービス アカウントへのアクセスを許可します。
    1. [] をクリックしてサービス アカウントのリストに戻ります。

    2. パイプラインの実行に使用するサービス アカウントの名前をクリックします。[サービス アカウントの詳細] ページが表示されます。

      ガイドの手順に沿って Vertex AI Pipelines 用のプロジェクトを構成した場合、これは、細かな権限を持つサービス アカウントの構成で作成したサービス アカウントと同じになります。それ以外の場合、Vertex AI は Compute Engine のデフォルト サービス アカウントを使用してパイプラインを実行します。Compute Engine のデフォルトのサービス アカウントの名前は PROJECT_NUMBER-compute@developer.gserviceaccount.com です。

    3. [権限] タブをクリックします。
    4. [アクセス権を付与] をクリックします。[プリンシパルの追加] パネルが表示されます。
    5. [新しいプリンシパル] ボックスに、前の手順で作成したサービス アカウントのメールアドレスを入力します。
    6. [ロール] プルダウン リストで、[サービス アカウント] > [サービス アカウント ユーザー] を選択します。
    7. [保存] をクリックします。

  4. 環境変数 GOOGLE_APPLICATION_CREDENTIALS を、サービス アカウント キーが含まれる JSON ファイルのパスに設定します。この変数は現在の shell セッションにのみ適用されるため、新しいセッションを開く場合は、変数を再度設定します。

    例: Linux または macOS

    [PATH] をサービス アカウント キーが含まれる JSON ファイルのパスに置き換えます。

    export GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    次に例を示します。

    export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

    例: Windows

    [PATH] は、サービス アカウント キーが保存されている JSON ファイルのパスに置き換え、[FILE_NAME] はファイル名に置き換えます。

    PowerShell を使用する場合:

    $env:GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    次に例を示します。

    $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\[FILE_NAME].json"

    コマンド プロンプトを使用する場合:

    set GOOGLE_APPLICATION_CREDENTIALS=[PATH]

パイプラインの実行

Vertex AI の PipelineJob を実行するには、PipelineJob オブジェクトを作成してから submit メソッドを呼び出す必要があります。

KFP でサポートされている特別な入力タイプ

パイプライン実行を作成する際に、KFP SDK でサポートされている次のプレースホルダを入力として渡すこともできます。

  • {{$.pipeline_job_name_placeholder}}

  • {{$.pipeline_job_resource_name_placeholder}}

  • {{$.pipeline_job_id_placeholder}}

  • {{$.pipeline_task_name_placeholder}}

  • {{$.pipeline_task_id_placeholder}}

  • {{$.pipeline_job_create_time_utc_placeholder}}

  • {{$.pipeline_root_placeholder}}

詳細については、Kubeflow Pipelines v2 ドキュメントの特殊な入力タイプをご覧ください。

from google.cloud import aiplatform

job = aiplatform.PipelineJob(display_name = DISPLAY_NAME,
                             template_path = COMPILED_PIPELINE_PATH,
                             job_id = JOB_ID,
                             pipeline_root = PIPELINE_ROOT_PATH,
                             parameter_values = PIPELINE_PARAMETERS,
                             enable_caching = ENABLE_CACHING,
                             encryption_spec_key_name = CMEK,
                             labels = LABELS,
                             credentials = CREDENTIALS,
                             project = PROJECT_ID,
                             location = LOCATION,
                             failure_policy = FAILURE_POLICY)

job.submit(service_account=SERVICE_ACCOUNT,
           network=NETWORK)

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

  • DISPLAY_NAME: パイプラインの名前。Google Cloud コンソールに表示されます。

  • COMPILED_PIPELINE_PATH: コンパイルされたパイプライン YAML ファイルのパス。ローカルパスまたは Cloud Storage URI を指定できます。

    省略可: コンパイルされたパイプラインの特定のバージョンを指定するには、次のいずれかの形式でバージョンタグを配置します。

    • COMPILED_PIPELINE_PATH:TAG。ここで、TAG はバージョンタグです。

    • COMPILED_PIPELINE_PATH@SHA256_TAG。ここで、SHA256_TAG はパイプライン バージョンの sha256 ハッシュ値です。

  • JOB_ID: （省略可）このパイプライン実行の固有識別子。ジョブ ID が指定されていない場合は、Vertex AI Pipelines により、パイプライン名とジョブの開始時間のタイムスタンプを使用してジョブ ID が作成されます。

  • PIPELINE_ROOT_PATH: （省略可）パイプラインの定義で指定されたパイプライン ルートパスをオーバーライドするには、パイプライン ジョブがアクセスできるパス（Cloud Storage バケット URI など）を指定します。

  • PIPELINE_PARAMETERS: （省略可）この実行に渡すパイプライン パラメータ。たとえば、パラメータ名にディクショナリ キーを使用し、パラメータ値にディクショナリ値を使用して、dict() を作成します。

  • ENABLE_CACHING: （省略可）このパイプライン実行で実行キャッシュを使用するかどうかを指定します。実行キャッシュを使用すると、現在の入力セットに対する出力が確認されているパイプライン タスクを省略することで、コストを削減できます。ENABLE_CACHING 引数を指定しない場合、このパイプライン実行で実行キャッシュが使用されます。実行キャッシュの詳細をご覧ください。

  • CMEK: （省略可）このパイプライン実行に使用する顧客管理の暗号鍵の名前。

  • LABELS: （省略可）この PipelineJob を整理するためにユーザーが定義したラベル。リソースラベルの詳細については、Resource Manager ドキュメントのラベルの作成と管理をご覧ください。

    Vertex AI Pipelines は、パイプライン実行に次のラベルを自動的に付けます。

    vertex-ai-pipelines-run-billing-id: pipeline_run_id

    ここで、pipeline_run_id はパイプライン実行の一意の ID です。

    このラベルは、請求レポートでパイプライン実行によって生成される Google Cloud リソースの使用量を関連付けます。

  • CREDENTIALS: （省略可）PipelineJob の作成に使用するカスタム認証情報。aiplatform.init に設定された認証情報をオーバーライドします。

  • PROJECT_ID: （省略可）パイプラインを実行する Google Cloud プロジェクト。このパラメータを設定しない場合は、aiplatform.init で設定されたプロジェクトが使用されます。

  • LOCATION: （省略可）パイプラインを実行するリージョン。Vertex AI Pipelines が利用可能なリージョンの詳細については、Vertex AI ロケーション ガイドをご覧ください。このパラメータを設定しないと、aiplatform.init で設定されたデフォルトのロケーションが使用されます。

  • FAILURE_POLICY: 省略可。パイプライン全体の障害ポリシーを指定します。次の構成を使用できます。

    • 1 つのタスクが失敗した後、パイプラインが失敗するように構成するには、「fast」と入力します。

    • 1 つのタスクが失敗した後もタスクのスケジュールを継続するようにパイプラインを構成するには、「slow」と入力します。

    このパラメータを設定しないと、エラーポリシーの構成はデフォルトの slow に設定されます。パイプラインの障害ポリシーの詳細を確認する。

  • SERVICE_ACCOUNT: 省略可。このパイプライン実行に使用するサービス アカウントの名前。サービス アカウントを指定しない場合、Vertex AI Pipelines はデフォルトの Compute Engine サービス アカウントを使用してパイプラインを実行します。

  • NETWORK: （省略可）このパイプライン実行に使用する VPC ピアリング ネットワークの名前。