GSP965

概要

パイプラインを使用すると、ML ワークフローを自動化し、再現可能性を確保することができます。Vertex AI は、Google Cloud 全体の ML サービスを統合してシームレスな開発エクスペリエンスを提供します。以前は、AutoML でトレーニングしたモデルやカスタムモデルには、個別のサービスを介してアクセスする必要がありました。Vertex AI は、それらを他の新しいプロダクトとともに 1 つの API へと統合します。また、Vertex AI には Vertex Pipelines などのさまざまな MLOps プロダクトが含まれています。このラボでは、Vertex Pipelines による ML パイプラインの作成方法と実行方法について学びます。

ML パイプラインはなぜ有用か？

詳細に入る前に、なぜパイプラインを使用するのかについて理解しましょう。データの処理、モデルのトレーニング、ハイパーパラメータのチューニング、評価、モデルのデプロイを含む ML ワークフローを構築しているとします。これらのステップにはそれぞれ異なる依存関係があり、ワークフロー全体をモノリスとして扱うと、扱いづらくなる場合があります。また、ML プロセスを拡張する際は、チームの他のメンバーがワークフローを実行し、コーディングに参加できるように、ML ワークフローを共有したいところですが、信頼性と再現性のあるプロセスがなければ困難です。パイプラインでは、ML プロセスの各ステップがそれぞれのコンテナとなります。これにより、ステップを独立して開発し、各ステップからの入力と出力を再現可能な方法で追跡できます。また、新しいトレーニング データが利用可能になったときなど、クラウド環境内の他のイベントに基づいてパイプラインの実行をスケジュールまたはトリガーすることもできます。

目標

このラボでは、次の方法について学びます。

• Kubeflow Pipelines SDK を使用してスケーラブルな ML パイプラインを構築する
• 3 つのステップからなり、入力としてテキストを受け取る簡単なパイプラインを作成して実行する
• AutoML 分類モデルのトレーニング、評価、デプロイを行うパイプラインを作成して実行する
• google_cloud_pipeline_components ライブラリを通して提供される事前構築済みコンポーネントを使用して Vertex AI サービスとやり取りする
• Cloud Scheduler でパイプライン ジョブのスケジュールを設定する

設定と要件

[ラボを開始] ボタンをクリックする前に

こちらの手順をお読みください。ラボの時間は記録されており、一時停止することはできません。[ラボを開始] をクリックするとスタートするタイマーは、Google Cloud のリソースを利用できる時間を示しています。

このハンズオンラボでは、シミュレーションやデモ環境ではなく、実際のクラウド環境を使ってご自身でラボのアクティビティを行うことができます。そのため、ラボの受講中に Google Cloud にログインおよびアクセスするための、新しい一時的な認証情報が提供されます。

このラボを完了するためには、下記が必要です。

• 標準的なインターネット ブラウザ（Chrome を推奨）

注: このラボの実行には、シークレット モードまたはシークレット ブラウジング ウィンドウを使用してください。これにより、個人アカウントと受講者アカウント間の競合を防ぎ、個人アカウントに追加料金が発生することを防ぎます。

• ラボを完了するために十分な時間を確保してください。ラボをいったん開始すると一時停止することはできません。

注: すでに個人の Google Cloud アカウントやプロジェクトをお持ちの場合でも、このラボでは使用しないでください。アカウントへの追加料金が発生する可能性があります。

ラボを開始して Google Cloud コンソールにログインする方法

1. [ラボを開始] ボタンをクリックします。ラボの料金をお支払いいただく必要がある場合は、表示されるポップアップでお支払い方法を選択してください。 左側の [ラボの詳細] パネルには、以下が表示されます。

   • [Google Cloud コンソールを開く] ボタン
   • 残り時間
   • このラボで使用する必要がある一時的な認証情報
   • このラボを行うために必要なその他の情報（ある場合）

2. [Google Cloud コンソールを開く] をクリックします（Chrome ブラウザを使用している場合は、右クリックして [シークレット ウィンドウでリンクを開く] を選択します）。

   ラボでリソースが起動し、別のタブで [ログイン] ページが表示されます。

   ヒント: タブをそれぞれ別のウィンドウで開き、並べて表示しておきましょう。

   注: [アカウントの選択] ダイアログが表示されたら、[別のアカウントを使用] をクリックします。

3. 必要に応じて、下のユーザー名をコピーして、[ログイン] ダイアログに貼り付けます。

   {{{user_0.username | "Username"}}}

   [ラボの詳細] パネルでも [ユーザー名] を確認できます。

4. [次へ] をクリックします。

5. 以下のパスワードをコピーして、[ようこそ] ダイアログに貼り付けます。

   {{{user_0.password | "Password"}}}

   [ラボの詳細] パネルでも [パスワード] を確認できます。

6. [次へ] をクリックします。

   重要: ラボで提供された認証情報を使用する必要があります。Google Cloud アカウントの認証情報は使用しないでください。 注: このラボでご自身の Google Cloud アカウントを使用すると、追加料金が発生する場合があります。

7. その後次のように進みます。

   • 利用規約に同意してください。
   • 一時的なアカウントなので、復元オプションや 2 要素認証プロセスは設定しないでください。
   • 無料トライアルには登録しないでください。

その後、このタブで Google Cloud コンソールが開きます。

注: Google Cloud のプロダクトやサービスのリストを含むメニューを表示するには、左上のナビゲーション メニューをクリックします。

タスク 1. Vertex ノートブック インスタンスを作成する

1. ナビゲーション メニューをクリックします。

2. [Vertex AI] に移動し、[ワークベンチ] に移動します。

3. ノートブック インスタンス ページで、[ユーザー管理のノートブック] タブに移動し、ai-notebook が完全に作成されるまで待ちます。

注: ノートブックが完全に作成されるまで数分かかります。

4. インスタンスが作成されたら、[JUPYTERLAB を開く] を選択します。

ノートブックが作成されたことを確認する

タスク 2. Vertex Pipelines の設定

Vertex Pipelines を使用するためには、いくつかのライブラリを追加でインストールする必要があります。

• Kubeflow Pipelines: これはパイプラインの構築に使用する SDK です。Vertex Pipelines は、Kubeflow Pipelines と TFX の両方で構築されたパイプラインの実行をサポートします。
• Google Cloud パイプライン コンポーネント: このライブラリは、パイプラインの各ステップで Vertex AI サービスとのやり取りを簡単にする事前構築済みコンポーネントを提供します。

ステップ 1: Python ノートブックを作成してライブラリをインストールする

1. ノートブック インスタンスのランチャー メニューから [Python 3] を選択して、ノートブックを作成します。

2. ランチャー メニューにアクセスするには、ノートブック インスタンスの左上にある [+] 記号をクリックします。

3. このラボに必要な両方のサービスをインストールするには、最初にノートブックのセルでユーザーフラグを設定します。

USER_FLAG = "--user"

4. 続いて、ノートブックから次のコードを実行します。

!pip3 install {USER_FLAG} google-cloud-aiplatform==1.0.0 --upgrade !pip3 install {USER_FLAG} kfp google-cloud-pipeline-components==0.1.1 --upgrade !pip3 install -U google-cloud-aiplatform !pip3 uninstall -y shapely pygeos geopandas !pip3 install shapely==1.8.5.post1 pygeos==0.12.0 geopandas==0.10.2 !pip3 install google-cloud-pipeline-components 注: バージョンに関する警告とエラーが表示されますが、無視してかまいません。

5. これらのパッケージをインストールした後、カーネルを再起動する必要があります。

import os if not os.getenv("IS_TESTING"): # インストール後カーネルを自動で再起動 import IPython app = IPython.Application.instance() app.kernel.do_shutdown(True)

6. 最後に、パッケージを正しくインストールしたことを確認します。KFP SDK のバージョンは 1.6 以上である必要があります。

!python3 -c "import kfp; print('KFP SDK version: {}'.format(kfp.__version__))" !python3 -c "import google_cloud_pipeline_components; print('google_cloud_pipeline_components version: {}'.format(google_cloud_pipeline_components.__version__))"

ステップ 2: プロジェクト ID とバケットを設定する

このラボ全体を通じて、自分の Cloud プロジェクト ID と前に作成したバケットを参照します。以下でこれらを格納する変数をそれぞれ作成します。

1. プロジェクト ID がわからない場合は、次のコードを実行して取得できる可能性があります。

import os PROJECT_ID = "" # gcloud から Google Cloud プロジェクト ID を取得 if not os.getenv("IS_TESTING"): shell_output=!gcloud config list --format 'value(core.project)' 2>/dev/null PROJECT_ID = shell_output[0] print("Project ID: ", PROJECT_ID)

2. 次に、バケット名を格納する変数を作成します。

BUCKET_NAME="gs://" + PROJECT_ID + "-bucket"

ステップ 3: ライブラリをインポートする

• このラボ全体を通して使用するライブラリをインポートするために、次のコードを追加します。

from typing import NamedTuple import kfp from kfp import dsl from kfp.v2 import compiler from kfp.v2.dsl import (Artifact, Dataset, Input, InputPath, Model, Output, OutputPath, ClassificationMetrics, Metrics, component) from kfp.v2.google.client import AIPlatformClient from google.cloud import aiplatform from google_cloud_pipeline_components import aiplatform as gcc_aip

ステップ 4: 定数を定義する

• パイプラインを構築する前に行う必要がある最後の作業は、いくつかの定数を定義することです。PIPELINE_ROOT は、パイプラインによって作成されるアーティファクトを書き込む Cloud Storage のパスです。ここではリージョンとして を使用しますが、バケットの作成時に別の region を使用した場合は、次のコード内の REGION 変数を更新します。

PATH=%env PATH %env PATH={PATH}:/home/jupyter/.local/bin REGION="{{{ project_0.default_region | Placeholder value. }}}" PIPELINE_ROOT = f"{BUCKET_NAME}/pipeline_root/" PIPELINE_ROOT

上のコードを実行すると、パイプラインのルート ディレクトリが表示されます。これは、パイプラインからのアーティファクトが書き込まれる Cloud Storage の場所です。これは gs://<bucket_name>/pipeline_root/ という形式になります。

タスク 3. 最初のパイプラインを作成する

• KFP SDK を使用して短いパイプラインを作成します。このパイプラインは ML 関連の処理は何も行いません（それについては後で学習します）。この演習では次の方法について学びます。

  • KFP SDK でカスタム コンポーネントを作成する方法
  • Vertex Pipelines でパイプラインを実行してモニタリングする方法

2 つの出力（商品名と絵文字の説明）を使用して 1 つの文を表示するパイプラインを作成します。このパイプラインは 3 つのコンポーネントで構成されます。

• product_name: このコンポーネントは商品名を入力として受け取り、その文字列を出力として返します。
• emoji: このコンポーネントは絵文字の説明テキストを受け取り、それを絵文字に変換します。たとえば、✨ のテキストコードは「sparkles」です。このコンポーネントでは絵文字ライブラリを使用しており、パイプラインで外部依存関係を管理する方法を見ることができます。
• build_sentence: この最後のコンポーネントは、前の 2 つの出力を使用して、絵文字を含む文を作成します。たとえば、「Vertex Pipelines is ✨」のような出力が得られます。

ステップ 1: Python の関数に基づくコンポーネントを作成する

KFP SDK を使用して、Python の関数に基づくコンポーネントを作成できます。最初に product_name コンポーネントを作成します。このコンポーネントは文字列を入力として受け取り、その文字列を返す単純なものです。

• ノートブックに次のコードを追加します。

@component(base_image="python:3.9", output_component_file="first-component.yaml") def product_name(text: str) -> str: return text

構文を詳しく見ていきます。

• @component デコレータにより、この関数はパイプラインの実行時にコンポーネントへとコンパイルされます。カスタム コンポーネントを記述するときには、これを使用します。
• base_image パラメータは、このコンポーネントが使用するコンテナ イメージを指定します。
• output_component_file パラメータはオプションで、コンパイルしたコンポーネントを書き込む yaml ファイルを指定します。セルの実行後、そのファイルがノートブック インスタンスに書き込まれているのを確認できます。このコンポーネントを他のユーザーと共有したい場合は、生成した yaml ファイルをそのユーザーに送信し、次のコードでそれを読み込んでもらうことができます。

product_name_component = kfp.components.load_component_from_file('./first-component.yaml')

関数定義の後の -> str は、このコンポーネントの出力タイプを指定します。

ステップ 2: 追加で 2 つのコンポーネントを作成する

1. パイプラインを完成させるために、あと 2 つのコンポーネントを作成します。最初のコンポーネントは文字列を入力として受け取り、それに対応する絵文字がある場合は、文字列を絵文字に変換します。渡された入力テキストと結果の絵文字を含むタプルを返します。

@component(base_image="python:3.9", output_component_file="second-component.yaml", packages_to_install=["emoji"]) def emoji( text: str, ) -> NamedTuple( "Outputs", [ ("emoji_text", str), # Return parameters ("emoji", str), ], ): import emoji emoji_text = text emoji_str = emoji.emojize(':' + emoji_text + ':', language='alias') print("output one: {}; output_two: {}".format(emoji_text, emoji_str)) return (emoji_text, emoji_str)

このコンポーネントは前のものよりも多少複雑です。注意点は次のとおりです。

• packages_to_install パラメータは、このコンテナの外部ライブラリ依存関係をコンポーネントに示します。ここでは、emoji というライブラリを使用しています。
• このコンポーネントは Outputs という名前の NamedTuple を返します。このタプルに含まれる各文字列には、emoji_text および emoji というキーがあります。次のコンポーネントで、これらを使用して出力にアクセスします。

2. このパイプラインの最後のコンポーネントは、最初の 2 つのコンポーネントの出力を組み合わせて 1 つの文字列を返します。

@component(base_image="python:3.9", output_component_file="third-component.yaml") def build_sentence( product: str, emoji: str, emojitext: str ) -> str: print("We completed the pipeline, hooray!") end_str = product + " is " if len(emoji) > 0: end_str += emoji else: end_str += emojitext return(end_str)

このコンポーネントは、前のステップで定義した出力を使用することがなぜわかるのだろうと疑問に思われるかもしれません。

重要なポイントです。次のステップで全体をまとめます。

ステップ 3: コンポーネントをパイプラインへとまとめる

ここまでのコンポーネント定義により、パイプラインの定義でステップの作成に使用できるファクトリ関数が作成されました。

1. パイプラインを設定するには、@dsl.pipeline デコレータを使用し、パイプラインの名前と説明を指定して、パイプラインのアーティファクトを書き込むルートパスを指定します。アーティファクトとは、パイプラインで生成される出力ファイルを意味します。これは、この簡単なパイプラインでは生成されませんが、次のパイプラインでは生成されます。

2. 次のコードブロックで、intro_pipeline 関数を定義します。ここでは、最初の各パイプライン ステップへの入力と、ステップを互いにどのように接続するかを指定します。

• product_task は商品名を入力として受け取ります。ここでは「Vertex Pipelines」を渡していますが、これは任意の名前に変更できます。
• emoji_task は絵文字のテキストコードを入力として受け取ります。これも任意の文字列に変更できます。たとえば、「party_face」は 🥳 という絵文字を表します。このコンポーネントと product_task コンポーネントはどちらも、それらに入力を与えるステップがないため、パイプラインを定義するときに手動で入力を指定します。
• パイプラインの最後のステップである consumer_task には、3 つの入力パラメータがあります。
  • product_task の出力。このステップは 1 つの出力のみを生成するため、product_task.output を介してそれを参照できます。
  • emoji_task ステップの emoji 出力。出力パラメータに名前を付けた emoji コンポーネントの定義を参照してください。
  • 同様に、emoji コンポーネントからの emoji_text 名前付き出力。絵文字に対応しないテキストがパイプラインに渡された場合は、このテキストを使用して文を作成します。

@dsl.pipeline( name="hello-world", description="An intro pipeline", pipeline_root=PIPELINE_ROOT, ) # ここでパラメータ `text` と `emoji_str` を変更することでパイプラインの出力を変更可能 def intro_pipeline(text: str = "Vertex Pipelines", emoji_str: str = "sparkles"): product_task = product_name(text) emoji_task = emoji(emoji_str) consumer_task = build_sentence( product_task.output, emoji_task.outputs["emoji"], emoji_task.outputs["emoji_text"], )

ステップ 4: パイプラインをコンパイルして実行する

1. パイプラインを定義したら、次はそれをコンパイルします。次のコードで、パイプラインの実行に使用する JSON ファイルを生成します。

compiler.Compiler().compile( pipeline_func=intro_pipeline, package_path="intro_pipeline_job.json" )

2. 次に、API クライアントをインスタンス化します。

api_client = AIPlatformClient( project_id=PROJECT_ID, region=REGION, )

3. 最後に、パイプラインを実行します。

response = api_client.create_run_from_job_spec( job_spec_path="intro_pipeline_job.json", # pipeline_root=PIPELINE_ROOT # パイプライン定義の一部として PIPELINE_ROOT を定義しなかった場合はこの引数が必要になる )

パイプラインを実行すると、コンソールでパイプラインの実行結果を見るためのリンクが生成されます。完了すると、次のように表示されます。

4. このパイプラインの実行には 5～6 分間かかります。完了したら、build-sentence コンポーネントをクリックして最終出力を表示できます。

KFP SDK と Vertex Pipelines の機能に習熟したことで、他の Vertex AI サービスを使用して ML モデルの作成やデプロイを行うパイプラインを構築できるようになりました。

絵文字のパイプラインの完了を確認する

タスク 4. エンドツーエンドの ML パイプラインを作成する

それでは、最初の ML パイプラインを構築しましょう。このパイプラインでは、UCI Machine Learning の Dry Beans データセットを使用します（出典: KOKLU, M. および OZKAN, I.A.（2020 年）「Multiclass Classification of Dry Beans Using Computer Vision and Machine Learning Techniques」、Computers and Electronics in Agriculture、174、105507。DOI）。

注: このパイプラインは完了まで 2 時間以上かかります。そのため、ラボを完了するのにパイプラインの実行を最後まで待つ必要はありません。パイプライン ジョブが開始されるまでの手順を実行してください。

これは表形式のデータセットです。パイプラインでこのデータセットを使用し、豆をその特徴に基づいて 7 種類に分類する AutoML モデルのトレーニング、評価、デプロイを行います。

このパイプラインは次の処理を行います。

• Vertex AI 内にデータセットを作成する
• AutoML を使用して表形式の分類モデルをトレーニングする
• このモデルの評価指標を取得する
• 評価指標に基づき、Vertex Pipelines で条件付きロジックを使用して、モデルをデプロイするかどうかを決定する
• Vertex Prediction を使用してモデルをエンドポイントにデプロイする

これらの各ステップがコンポーネントとなります。パイプラインのほとんどのステップでは、このラボで前にインポートした google_cloud_pipeline_components ライブラリを介して Vertex AI サービス用の事前構築済みコンポーネントを使用します。

このセクションでは、最初に 1 つのカスタム コンポーネントを定義します。次に、事前構築済みコンポーネントを使用してパイプラインの残りのステップを定義します。事前構築済みコンポーネントにより、モデルのトレーニングやデプロイなど、Vertex AI サービスへのアクセスが簡単になります。

このステップの大部分の時間は、このパイプラインの AutoML トレーニング部分に費やされ、これには約 1 時間かかります。

ステップ 1: モデル評価用のカスタム コンポーネント

ここで定義するカスタム コンポーネントは、モデルのトレーニングの完了後、パイプラインの終わりにかけて使用されます。このコンポーネントはいくつかの処理を実行します。

• トレーニング済みの AutoML 分類モデルから評価指標を取得する
• 指標を解析し、Vertex Pipelines UI にレンダリングする
• 指標をしきい値と比較して、モデルをデプロイするかどうかを決定する

コンポーネントを定義する前に、その入力パラメータと出力パラメータについて理解します。パイプラインは入力として Cloud プロジェクトのメタデータ、結果のトレーニング済みモデル（このコンポーネントは後で定義します）、モデルの評価指標、そして thresholds_dict_str を受け取ります。

thresholds_dict_str は、パイプラインの実行時に定義する値です。この分類モデルの場合、これはモデルをデプロイする ROC 曲線値の下側の領域となります。たとえば、0.95 を渡した場合は、この指標が 95% を超えるときにのみパイプラインでモデルをデプロイすることを意味します。

評価コンポーネントは、モデルをデプロイするかどうかを示す文字列を返します。

• ノートブックのセルに次のコードを追加して、このカスタム コンポーネントを作成します。

@component( base_image="gcr.io/deeplearning-platform-release/tf2-cpu.2-3:latest", output_component_file="tables_eval_component.yaml", # オプション: これを使用して後にコンポーネントを読み込むことができます packages_to_install=["google-cloud-aiplatform"], ) def classif_model_eval_metrics( project: str, location: str, # "region", api_endpoint: str, # "region-aiplatform.googleapis.com", thresholds_dict_str: str, model: Input[Model], metrics: Output[Metrics], metricsc: Output[ClassificationMetrics], ) -> NamedTuple("Outputs", [("dep_decision", str)]): # 戻り値のパラメータ """これは AutoML の表形式分類モデルの評価指標をレンダリングする関数である。 AutoML の表形式トレーニング プロセスによって生成された分類モデル評価を取得し、解析作業を行って、その情報を元にモデルの ROC 曲線と混同行列をレンダリングする。また与えられた指標のしきい値情報を用いて、評価結果と比較を行い、モデルの精度がデプロイするのに十分かどうかを判断する。 """ import json import logging from google.cloud import aiplatform # モデルの評価情報をフェッチ def get_eval_info(client, model_name): from google.protobuf.json_format import MessageToDict response = client.list_model_evaluations(parent=model_name) metrics_list = [] metrics_string_list = [] for evaluation in response: print("model_evaluation") print(" name:", evaluation.name) print(" metrics_schema_uri:", evaluation.metrics_schema_uri) metrics = MessageToDict(evaluation._pb.metrics) for metric in metrics.keys(): logging.info("metric: %s, value: %s", metric, metrics[metric]) metrics_str = json.dumps(metrics) metrics_list.append(metrics) metrics_string_list.append(metrics_str) return ( evaluation.name, metrics_list, metrics_string_list, ) # 与えられた指標のしきい値を用いてモデルの精度がデプロイするのに # 十分かどうかを判断。 def classification_thresholds_check(metrics_dict, thresholds_dict): for k, v in thresholds_dict.items(): logging.info("k {}, v {}".format(k, v)) if k in ["auRoc", "auPrc"]: # より高い方が良い if metrics_dict[k] < v: # しきい値を下回る場合はデプロイしない logging.info( "{} < {}; returning False".format(metrics_dict[k], v) ) return False logging.info("threshold checks passed.") return True def log_metrics(metrics_list, metricsc): test_confusion_matrix = metrics_list[0]["confusionMatrix"] logging.info("rows: %s", test_confusion_matrix["rows"]) # ROC 曲線をロギング fpr = [] tpr = [] thresholds = [] for item in metrics_list[0]["confidenceMetrics"]: fpr.append(item.get("falsePositiveRate", 0.0)) tpr.append(item.get("recall", 0.0)) thresholds.append(item.get("confidenceThreshold", 0.0)) print(f"fpr: {fpr}") print(f"tpr: {tpr}") print(f"thresholds: {thresholds}") metricsc.log_roc_curve(fpr, tpr, thresholds) # 混同行列をロギング annotations = [] for item in test_confusion_matrix["annotationSpecs"]: annotations.append(item["displayName"]) logging.info("confusion matrix annotations: %s", annotations) metricsc.log_confusion_matrix( annotations, test_confusion_matrix["rows"], ) # テキストの指標情報もロギング for metric in metrics_list[0].keys(): if metric != "confidenceMetrics": val_string = json.dumps(metrics_list[0][metric]) metrics.log_metric(metric, val_string) # metrics.metadata["model_type"] = "AutoML Tabular classification" logging.getLogger().setLevel(logging.INFO) aiplatform.init(project=project) # モデルのリソース名を入力 Model Artifact から抽出 model_resource_path = model.uri.replace("aiplatform://v1/", "") logging.info("model path: %s", model_resource_path) client_options = {"api_endpoint": api_endpoint} # リクエストの作成と送信に使うクライアントを初期化する。 client = aiplatform.gapic.ModelServiceClient(client_options=client_options) eval_name, metrics_list, metrics_str_list = get_eval_info( client, model_resource_path ) logging.info("got evaluation name: %s", eval_name) logging.info("got metrics list: %s", metrics_list) log_metrics(metrics_list, metricsc) thresholds_dict = json.loads(thresholds_dict_str) deploy = classification_thresholds_check(metrics_list[0], thresholds_dict) if deploy: dep_decision = "true" else: dep_decision = "false" logging.info("deployment decision is %s", dep_decision) return (dep_decision,)

ステップ 2: Google Cloud の事前構築済みコンポーネントを追加する

このステップでは、パイプラインの残りのコンポーネントを定義し、それらすべてがどのように機能するのかを確認します。

1. 最初に、タイムスタンプを使用してパイプライン実行の表示名を定義します。

import time DISPLAY_NAME = 'automl-beans{}'.format(str(int(time.time()))) print(DISPLAY_NAME)

2. 次に、ノートブックの新しいセルに以下をコピーします。

@kfp.dsl.pipeline(name="automl-tab-beans-training-v2", pipeline_root=PIPELINE_ROOT) def pipeline( bq_source: str = "bq://aju-dev-demos.beans.beans1", display_name: str = DISPLAY_NAME, project: str = PROJECT_ID, gcp_region: str = "{{{ project_0.default_region | Placeholder value. }}}", api_endpoint: str = "{{{ project_0.default_region | Placeholder value. }}}-aiplatform.googleapis.com", thresholds_dict_str: str = '{"auRoc": 0.95}', ): dataset_create_op = gcc_aip.TabularDatasetCreateOp( project=project, display_name=display_name, bq_source=bq_source ) training_op = gcc_aip.AutoMLTabularTrainingJobRunOp( project=project, display_name=display_name, optimization_prediction_type="classification", budget_milli_node_hours=1000, column_transformations=[ {"numeric": {"column_name": "Area"}}, {"numeric": {"column_name": "Perimeter"}}, {"numeric": {"column_name": "MajorAxisLength"}}, {"numeric": {"column_name": "MinorAxisLength"}}, {"numeric": {"column_name": "AspectRation"}}, {"numeric": {"column_name": "Eccentricity"}}, {"numeric": {"column_name": "ConvexArea"}}, {"numeric": {"column_name": "EquivDiameter"}}, {"numeric": {"column_name": "Extent"}}, {"numeric": {"column_name": "Solidity"}}, {"numeric": {"column_name": "roundness"}}, {"numeric": {"column_name": "Compactness"}}, {"numeric": {"column_name": "ShapeFactor1"}}, {"numeric": {"column_name": "ShapeFactor2"}}, {"numeric": {"column_name": "ShapeFactor3"}}, {"numeric": {"column_name": "ShapeFactor4"}}, {"categorical": {"column_name": "Class"}}, ], dataset=dataset_create_op.outputs["dataset"], target_column="Class", ) model_eval_task = classif_model_eval_metrics( project, gcp_region, api_endpoint, thresholds_dict_str, training_op.outputs["model"], ) with dsl.Condition( model_eval_task.outputs["dep_decision"] == "true", name="deploy_decision", ): deploy_op = gcc_aip.ModelDeployOp( # noqa: F841 model=training_op.outputs["model"], project=project, machine_type="e2-standard-4", )

このコードの内容は次のとおりです。

• 最初に、前のパイプラインと同様に、このパイプラインが受け取る入力パラメータを定義します。これらはパイプライン内の他のステップの出力には依存しないため、手動で設定する必要があります。
• パイプラインのその他の部分では、Vertex AI サービスとやり取りするために、いくつかの事前構築済みコンポーネントを使用します。
  • TabularDatasetCreateOp は、Cloud Storage または BigQuery をデータセットのソースとして、Vertex AI に表形式のデータセットを作成します。このパイプラインでは、BigQuery テーブルの URL を介してデータを渡します。
  • AutoMLTabularTrainingJobRunOp は、表形式のデータセットに対する AutoML のトレーニング ジョブを開始します。このコンポーネントには、モデルタイプ（ここでは分類）、列のデータ、トレーニングの実行時間、データセットへのポインタなど、いくつかの構成パラメータを渡します。このコンポーネントにデータセットを渡すために、dataset_create_op.outputs["dataset"] を介して前のコンポーネントの出力を渡していることに注意してください。
  • ModelDeployOp は、指定されたモデルを Vertex AI のエンドポイントにデプロイします。追加の構成オプションも使用できますが、ここではエンドポイントのマシンタイプ、プロジェクト、デプロイするモデルを指定します。パイプラインのトレーニング ステップの出力にアクセスすることで、モデルを渡します。
• このパイプラインでは、条件付きロジックも使用されています。これは、条件を定義し、その条件の結果に基づいて異なるブランチを実行できる Vertex Pipelines の機能です。パイプラインを定義したときに thresholds_dict_str パラメータを渡したことを思い出してください。これは、モデルをエンドポイントにデプロイするかどうかの決定に使用する精度のしきい値です。これを実装するために、KFP SDK の Condition クラスを使用します。渡される条件は、このラボで前に定義したカスタム評価コンポーネントの出力です。この条件が true の場合、パイプラインは引き続き deploy_op コンポーネントを実行します。精度が定義済みのしきい値を満たしていない場合、パイプラインはここで停止し、モデルをデプロイしません。

ステップ 3: エンドツーエンドの ML パイプラインをコンパイルして実行する

1. パイプライン全体が定義されたので、これをコンパイルします。

compiler.Compiler().compile( pipeline_func=pipeline, package_path="tab_classif_pipeline.json" )

2. 次に、パイプラインの実行を開始します。

response = api_client.create_run_from_job_spec( "tab_classif_pipeline.json", pipeline_root=PIPELINE_ROOT, parameter_values={"project": PROJECT_ID,"display_name": DISPLAY_NAME} )

3. 上のセルの実行後に表示されるリンクをクリックして、パイプラインをコンソールに表示します。このパイプラインの実行には 1 時間以上かかります。そのほとんどの時間は AutoML のトレーニング ステップに費やされます。完成したパイプラインは次のようになります。

4. 上部の「アーティファクトを開く」ボタンを使用して、パイプラインで作成された各アーティファクトの詳細を表示できます。たとえば、dataset アーティファクトをクリックすると、作成された Vertex AI データセットの詳細が表示されます。ここからリンクをクリックして、そのデータセットのページに移動できます。

5. 同様に、カスタム評価コンポーネントの結果として得られた指標の可視化を見るには、metricsc というアーティファクトをクリックします。ダッシュボードの右側で、このモデルの混同行列を確認できます。

6. このパイプラインの実行によって作成されたモデルとエンドポイントを表示するには、モデル セクションに移動し、automl-beans という名前のモデルをクリックします。このモデルがエンドポイントにデプロイされているのを確認できます。

7. パイプラインのグラフで endpoint アーティファクトをクリックしても、このページにアクセスできます。

8. コンソールでパイプラインのグラフを確認するほかに、Vertex Pipelines で リネージのトラッキングを使用することもできます。

9. リネージのトラッキングとは、パイプライン全体で作成されたアーティファクトのトラッキングを意味します。これにより、各アーティファクトがどこで作成され、ML ワークフロー全体でどのように使用されているかを理解できます。たとえば、このパイプラインで作成されたデータセットについてリネージのトラッキングを見るには、dataset アーティファクトをクリックしてから、[リネージを表示] をクリックします。

このアーティファクトが使用されているすべての場所が表示されます。

注: パイプライン内のトレーニング ジョブが開始されるまで待ってから、以下の進捗を確認します。トレーニング ジョブにかかる時間は、このラボに割り当てられている時間よりも長くなりますが、トレーニング ジョブを正しく送信すれば、満点が付与されます。 エンドツーエンドの ML パイプラインのトレーニング ジョブが開始されたかどうか確認する

ステップ 4: パイプラインの実行間で指標を比較する（オプション）

• このパイプラインを複数回実行すると、実行間で指標を比較したくなるかもしれません。aiplatform.get_pipeline_df() メソッドを使用して、実行のメタデータにアクセスできます。ここでは、このパイプラインのすべての実行のメタデータを取得し、Pandas DataFrame に読み込みます。

pipeline_df = aiplatform.get_pipeline_df(pipeline="automl-tab-beans-training-v2") small_pipeline_df = pipeline_df.head(2) small_pipeline_df

これで、Vertex Pipelines でエンドツーエンドの ML パイプラインを構築、実行し、メタデータを取得する方法を学習しました。

お疲れさまでした

このラボでは、絵文字のパイプラインを作成して実行しました。また、Vertex Pipelines でエンドツーエンドの ML パイプラインを構築、実行し、メタデータを取得する方法も学習しました。

次のステップと詳細情報

Developer Relations の Codelab を使用して、ご自身の Google Cloud プロジェクトで同じシナリオを試してみてください。

• Vertex Pipelines の概要 | Sara Robinson

Google Cloud トレーニングと認定資格

Google Cloud トレーニングと認定資格を通して、Google Cloud 技術を最大限に活用できるようになります。必要な技術スキルとベスト プラクティスについて取り扱うクラスでは、学習を継続的に進めることができます。トレーニングは基礎レベルから上級レベルまであり、オンデマンド、ライブ、バーチャル参加など、多忙なスケジュールにも対応できるオプションが用意されています。認定資格を取得することで、Google Cloud テクノロジーに関するスキルと知識を証明できます。

マニュアルの最終更新日: 2024 年 6 月 28 日

ラボの最終テスト日: 2024 年 6 月 28 日

Copyright 2024 Google LLC All rights reserved. Google および Google のロゴは Google LLC の商標です。その他すべての企業名および商品名はそれぞれ各社の商標または登録商標です。