GSP965

總覽

Pipeline 可助您自動執行及重複使用機器學習工作流程。Vertex AI 整合了 Google Cloud 機器學習服務，提供流暢的開發體驗。以 AutoML 訓練的模型和自訂模型，先前需透過不同的服務存取。Vertex AI 將這些服務併至單一 API，並加入其他新產品。Vertex AI 也提供 Vertex Pipelines 等各種 MLOps 產品。本實驗室將帶您瞭解如何透過 Vertex Pipelines 建立及執行機器學習 pipeline。

機器學習 pipeline 有什麼用途？

深入探索前，我們先介紹使用 pipeline 的原因。假設您在建構機器學習工作流程，流程中會處理資料、訓練模型、調整超參數、評估及開發模型。每個步驟可能有不同的依附元件，如果光看整個工作流程，可能難以掌握這些步驟。開始擴展機器學習流程時，您或許會想與團隊成員分享機器學習工作流程，方便他們執行流程及貢獻程式碼。這時如果沒有可靠且可重複使用的程序，可能會難以進行。Pipeline 可將機器學習流程的各步驟化為容器，讓您開發個別步驟，並用可重複使用的方法，追蹤各步驟的輸入和輸出內容。您也可以根據 Cloud 環境的其他事件 (例如：有新的訓練資料時)，排定或觸發 pipeline 執行作業。

目標

本實驗室的內容包括：

• 使用 Kubeflow Pipelines SDK 建構可擴充的機器學習 pipeline
• 建立並執行接受文字輸入內容的 3 步驟簡易 pipeline
• 建立並執行 pipeline 來訓練、評估及部署 AutoML 分類模型
• 使用透過「google_cloud_pipeline_components」程式庫的預先建構元件，與 Vertex AI 服務互動
• 使用 Cloud Scheduler 為 pipeline 工作安排時間

設定和需求

瞭解以下事項後，再點選「Start Lab」按鈕

請詳閱以下操作說明。實驗室活動會計時，且中途無法暫停。點選「Start Lab」後就會開始計時，顯示可使用 Google Cloud 資源的時間。

您將在真正的雲端環境完成實作實驗室活動，而不是模擬或示範環境。為此，我們會提供新的暫時憑證，供您在實驗室活動期間登入及存取 Google Cloud。

為了順利完成這個實驗室，請先確認：

• 可以使用標準的網際網路瀏覽器 (Chrome 瀏覽器為佳)。

注意事項：請使用無痕模式 (建議選項) 或私密瀏覽視窗執行此實驗室，這可以防止個人帳戶和學員帳戶之間的衝突，避免個人帳戶產生額外費用。

• 是時候完成實驗室活動了！別忘了，活動一旦開始將無法暫停。

注意事項：務必使用實驗室專用的學員帳戶。如果使用其他 Google Cloud 帳戶，可能會產生額外費用。

如何開始研究室及登入 Google Cloud 控制台

1. 點選「Start Lab」按鈕。如果實驗室會產生費用，畫面上會出現選擇付款方式的對話方塊。左側的「Lab Details」窗格會顯示下列項目：

   • 「Open Google Cloud console」按鈕
   • 剩餘時間
   • 必須在這個研究室中使用的臨時憑證
   • 完成這個實驗室所需的其他資訊 (如有)

2. 點選「Open Google Cloud console」；如果使用 Chrome 瀏覽器，也能按一下滑鼠右鍵，選取「在無痕視窗中開啟連結」。

   接著，實驗室會啟動相關資源，並開啟另一個分頁，顯示「登入」頁面。

   提示：您可以在不同的視窗中並排開啟分頁。

   注意：如果頁面中顯示「選擇帳戶」對話方塊，請點選「使用其他帳戶」。

3. 如有必要，請將下方的 Username 貼到「登入」對話方塊。

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

   您也可以在「Lab Details」窗格找到 Username。

4. 點選「下一步」。

5. 複製下方的 Password，並貼到「歡迎使用」對話方塊。

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

   您也可以在「Lab Details」窗格找到 Password。

6. 點選「下一步」。

   重要事項：請務必使用實驗室提供的憑證，而非自己的 Google Cloud 帳戶憑證。 注意：如果使用自己的 Google Cloud 帳戶來進行這個實驗室，可能會產生額外費用。

7. 按過後續的所有頁面：

   • 接受條款及細則。
   • 由於這是臨時帳戶，請勿新增救援選項或雙重驗證機制。
   • 請勿申請免費試用。

Google Cloud 控制台稍後會在這個分頁開啟。

注意：如要使用 Google Cloud 產品和服務，請點選「導覽選單」，或在「搜尋」欄位輸入服務或產品名稱。

工作 1：建立 Vertex 筆記本執行個體

1. 按一下「導覽選單」。

2. 前往「Vertex AI」，然後點選「Workbench」。

3. 在筆記本執行個體頁面，前往「使用者自行管理的筆記本」分頁，然後等待 ai-notebook 建立完畢。

注意：筆記本需要幾分鐘才能建立完畢。

4. 建立執行個體後，請選取「Open JupyterLab」：

確認筆記本已建立完畢

工作 2：設定 Vertex Pipelines

使用 Vertex Pipelines 前，需先安裝幾個額外的程式庫：

• Kubeflow Pipelines 是用於建構 pipeline 的 SDK。Vertex Pipelines 可執行透過 Kubeflow Pipelines 和/或 TFX 建構的 pipeline。
• Google Cloud Pipeline Components 提供預先建立的元件，可讓您在建構 pipeline 時，輕鬆操作 Vertex AI 服務。

步驟 1：建立 Python 筆記本及安裝程式庫

1. 從筆記本執行個體的啟動器選單，選取「Python 3」建立筆記本：

2. 點選筆記本執行個體左上方的「+」符號，即可打開啟動器選單。

3. 如要安裝本實驗室所需的兩項服務，請先在筆記本儲存格中設定使用者旗標：

USER_FLAG = "--user"

4. 接著，從筆記本執行下列指令：

!pip3 install {USER_FLAG} google-cloud-aiplatform==1.59.0 !pip3 install {USER_FLAG} kfp google-cloud-pipeline-components==0.1.1 --upgrade !pip3 uninstall -y shapely pygeos geopandas !pip3 install shapely==1.8.5.post1 pygeos==0.12.0 geopandas>=0.12.2 !pip3 install google-cloud-pipeline-components 注意：畫面會顯示與版本相關的警告與錯誤訊息，您可以放心忽略。

5. 安裝這些套件後，需重新啟動核心：

import os if not os.getenv("IS_TESTING"): # Automatically restart kernel after installs 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 和 bucket

您會在本實驗室用到自己的 Cloud 專案 ID，以及先前建立的 bucket。接下來，請為 ID 和 bucket 建立變數。

1. 如果不知道自己的專案 ID，執行以下指令或許就能取得：

import os PROJECT_ID = "" # Get your Google Cloud project ID from gcloud 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 名稱的變數。

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 前，最後還需定義幾項常數變數。PIPELINE_ROOT 是 Cloud Storage 路徑，pipeline 建立的構件會寫入到這個位置。此程式碼使用 做為區域，但如果您在建立 bucket 時使用不同的 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

執行上方的程式碼後，您應該會看到 pipeline 的根目錄。這是要寫入 pipeline 構件的 Cloud Storage 位置。格式為 gs://<bucket_name>/pipeline_root/。

工作 3：建立第一個 pipeline

• 使用 KFP SDK 建立短 pipeline。這個 pipeline 不會進行任何與機器學習相關的作業 (請放心，日後還有機會學)，本練習涵蓋以下主題：

  • 如何在 KFP SDK 建立自訂元件
  • 如何在 Vertex Pipelines 執行及監控 pipeline

接著建立輸出句子的 pipeline，句中有以下兩項元件：產品名稱和表情符號說明。這個 pipeline 將由三項元件組成：

• product_name 會使用產品名稱做為輸入內容，並傳回該字串做為輸出內容。
• emoji 會將表情符號的文字說明轉換為表情符號，像是把文字程式碼「sparkles」轉為 ✨。這個元件使用表情符號程式庫，顯示如何在 pipeline 管理外部依附元件。
• build_sentence 會使用先前兩個項目的輸出內容，建構含有表情符號的句子。舉例來說，輸出結果可能會是「Vertex Pipelines 很 ✨」。

步驟 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 修飾符會在 pipeline 執行時，將此函式編譯至元件。編寫自訂元件時，可使用此修飾符。
• base_image 參數會指定這個元件要用的容器映像檔。
• output_component_file 為選用參數，可指定要寫入已編譯元件的 yaml 檔案。執行該儲存格後，應該會看到該檔案寫入筆記本執行個體。如果您想與其他人分享這個元件，可以傳送生成的 yaml 檔案，並請他們使用下列程式碼載入檔案：

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

「函式定義」後方的 -> str 指定這個元件的輸出內容類型。

步驟 2：再建立兩個元件

1. 為完成 pipeline，請再建立兩個元件。第一個元件會將字串做為輸入內容，並轉換為對應的表情符號 (如有)。此元件會傳回元組 (內含傳遞的輸入文字)，以及相應的表情符號：

@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. 這個 pipeline 的最後一個元件會使用前兩個元件的輸出內容，並將其合併來傳回字串：

@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：將元件組合成 pipeline

上方元件定義建立了工廠函式，您可在 pipeline 定義中使用這些函式建立步驟。

1. 請設定 pipeline：使用 @dsl.pipeline 修飾符，為 pipeline 加入名稱和說明，然後提供要寫入 pipeline 構件的根路徑。構件是指所有由 pipeline 生成的輸出檔案。這個簡單的 pipeline 不會生成任何檔案，但下一個 pipeline 會。

2. 您會在下一個程式碼區塊，定義 intro_pipeline 函式，指定初始 pipeline 步驟的輸入內容，以及步驟的銜接方式：

• product_task 會使用產品名稱做為輸入內容。在本實驗室，此元件會傳遞「Vertex Pipelines」，但您可以視需要變更。
• emoji_task 會使用表情符號的文字程式碼做為輸入內容，您也可以視需要變更。舉例來說，「party_face」是 🥳 表情符號。請注意，由於沒有任何步驟會將輸入內容傳遞至這個元件和 product_task 元件，因此定義 pipeline 時，需要手動指定這些元件的輸入內容。
• 在 pipeline 最後一個步驟，consumer_task 有三個輸入參數：
  • product_task 的輸出內容。由於這個步驟只會產生一項輸出內容，您可以用 product_task.output 參照。
  • emoji_task 步驟的 emoji 輸出內容。請查看您在上方定義的 emoji 元件，您在其中命名了輸入內容參數。
  • 同樣地，emoji 元件中名為 emoji_text 的輸出內容。如果 pipeline 傳遞的文字沒有對應的表情符號，系統會使用該文字來建構句子。

@dsl.pipeline( name="hello-world", description="An intro pipeline", pipeline_root=PIPELINE_ROOT, ) # You can change the `text` and `emoji_str` parameters here to update the pipeline output 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：編譯及執行 pipeline

1. 定義 pipeline 後，即可編譯 pipeline。以下程式碼會產生用於執行 pipeline 的 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. 最後，執行 pipeline：

response = api_client.create_run_from_job_spec( job_spec_path="intro_pipeline_job.json", # pipeline_root=PIPELINE_ROOT # this argument is necessary if you did not specify PIPELINE_ROOT as part of the pipeline definition. )

執行 pipeline 應該會產生連結，點按就能在控制台查看 pipeline 執行。完成後應如下所示：

4. 這個 pipeline 的執行時間為 5 至 6 分鐘。完成後，點選 build-sentence 元件即可查看最終輸出內容：

您已熟悉 KFP SDK 和 Vertex Pipelines 的運作方式，現在可以建構 pipeline，運用其他 Vertex AI 服務建立及部署機器學習模型。

確認已完成表情符號 pipeline

工作 4：建立端對端機器學習 pipeline

接著就來建構第一個機器學習 pipeline 吧！您將在這個 pipeline，使用 UCI Machine Learning Dry Beans 資料集 (來源：KOKLU, M. and OZKAN, I.A., (2020), "Multiclass Classification of Dry Beans Using Computer Vision and Machine Learning Techniques."In Computers and Electronics in Agriculture, 174, 105507. DOI)。

注意：這個 pipeline 需要超過 2 小時才能建構及執行完畢。因此，您可以先完成本實驗室，不必等待 pipeline 執行完畢。請按照步驟操作，直到 pipeline 工作開始運作即可。

此為表格資料集，您將在 pipeline 使用此資料集訓練、評估及部署 AutoML 模型，並根據特徵將豆子分為 7 類。

這個 pipeline 將進行以下操作：

• 在 Vertex AI 建立資料集
• 運用 AutoML 訓練表格分類模型
• 取得這個模型的評估指標
• 根據評估指標，決定是否使用 Vertex Pipelines 中的條件邏輯來部署模型
• 使用 Vertex Prediction 將模型部署至端點

接下來的各個步驟都是一個元件。大部分的 pipeline 步驟會使用您先前在這個實驗室匯入的 google_cloud_pipeline_components程式庫，透過預先建立的元件操作 Vertex AI 服務。

在這裡，我們會先定義一個自訂元件，再用預先建立的元件，定義其餘的 pipeline 步驟。透過預先建立的元件，您將能輕鬆使用 Vertex AI 服務，例如訓練和部署模型的服務。

此步驟約需 1 小時，大部分的時間會用於 pipeline 的 AutoML 訓練作業上。

步驟 1：用於評估模型的自訂元件

接下來請定義自訂元件，在模型訓練完成後、pipeline 作業結尾時使用。此元件將進行以下操作：

• 從訓練後的 AutoML 分類模型取得評估指標
• 剖析指標，並在 Vertex Pipelines UI 轉譯
• 比較指標數據與門檻的差距，決定是否要部署模型

定義元件前，需先瞭解元件的輸入和輸出參數。這個管道會使用您 Cloud 專案的部分中繼資料、訓練後的模型 (之後會定義這個元件)、模型的評估指標，以及 thresholds_dict_str，做為輸入內容。

您會在執行 pipeline 時定義 thresholds_dict_str。在本例的分類模型中，此為應部署模型的 ROC 曲線下面積值。舉例來說，如果傳入 0.95，表示您希望 pipeline 只在這項指標超過 95% 時部署模型。

評估元件會傳回字串，指出是否要部署模型。

• 將以下程式碼加入筆記本儲存格，建立這個自訂元件：

@component( base_image="gcr.io/deeplearning-platform-release/tf2-cpu.2-3:latest", output_component_file="tables_eval_component.yaml", # Optional: you can use this to load the component later 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)]): # Return parameter. """This function renders evaluation metrics for an AutoML Tabular classification model. It retrieves the classification model evaluation generated by the AutoML Tabular training process, does some parsing, and uses that info to render the ROC curve and confusion matrix for the model. It also uses given metrics threshold information and compares that to the evaluation results to determine whether the model is sufficiently accurate to deploy. """ import json import logging from google.cloud import aiplatform # Fetch model eval info 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, ) # Use the given metrics threshold(s) to determine whether the model is # accurate enough to deploy. 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"]: # higher is better if metrics_dict[k] < v: # if under threshold, don't deploy 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"]) # log the ROC curve 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) # log the confusion matrix 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"], ) # log textual metrics info as well 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) # extract the model resource name from the input Model Artifact model_resource_path = model.uri.replace("aiplatform://v1/", "") logging.info("model path: %s", model_resource_path) client_options = {"api_endpoint": api_endpoint} # Initialize client that will be used to create and send requests. 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 預先建立元件

您會在本步驟定義其餘 pipeline 元件，並瞭解這些元件的搭配方式。

1. 首先，請使用時間戳記為 pipeline 執行作業定義顯示名稱：

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", )

此程式碼會有以下結果：

• 首先，如同前一個 pipeline，這個 pipeline 要用的輸入參數由您定義。您需要手動設定參數，因為這些參數不會採用 pipeline 中其他步驟的輸出內容。
• Pipeline 的其他步驟會使用預先建立的元件來操作 Vertex AI 服務：
  • TabularDatasetCreateOp 會根據 Cloud Storage 或 BigQuery 的資料集來源，在 Vertex AI 建立表格資料集。在這個 pipeline，您會透過 BigQuery 資料表網址傳遞資料。
  • AutoMLTabularTrainingJobRunOp 會啟動表格資料集的 AutoML 訓練工作。請傳遞一些設定參數至這個元件，包括模型類型 (本例為分類模型)、資料欄的一些資料、訓練的執行時間，以及資料集的相關資訊。請注意，如要將資料集傳入此元件，需使用 dataset_create_op.outputs["dataset"] 提供前一個元件的輸出內容。
  • ModelDeployOp 會將指定模型部署到 Vertex AI 中的端點。有其他設定選項，但在這裡，請提供端點機型、專案及要部署的模型，再存取 pipeline 中訓練步驟的輸出內容，將資料傳入模型。
• 這個 pipeline 也會使用 Vertex Pipelines 的條件邏輯功能，讓您定義條件，以及依據該條件結果產生的不同分支版本。請注意，定義 pipeline 時，需傳遞 thresholds_dict_str 參數。此為準確率門檻，用於決定是否要將模型部署至端點。請使用 KFP SDK 的 Condition 類別來導入這項機制。傳入的條件，是您先前在本實驗室定義的自訂評估元件的輸出結果。如果此條件為「true」，pipeline 將繼續執行 deploy_op 元件。如果準確率未達預先定義的門檻，pipeline 會停止且不會部署模型。

步驟 3：編譯及執行端對端機器學習 pipeline

1. 定義完整 pipeline 後，就能編譯 pipeline：

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

2. 接著，開始執行 pipeline：

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. 按一下執行上方儲存格後顯示的連結，在控制台查看 pipeline。此 pipeline 的執行時間會稍微超過 1 小時，大部分的時間會用於 AutoML 訓練步驟。執行完畢的 pipeline 應如下所示：

4. 切換頂端的「Expand artifacts」按鈕，就能深入瞭解以 pipeline 建立的各個構件。舉例來說，點選 dataset 構件，便會看到 Vertex AI 資料集的詳細資料。點選下方連結，即可前往該資料集的頁面：

5. 同樣地，如要查看自訂評估元件產生的指標圖表，請點選「metricsc」構件。您可以在資訊主頁右側，看到這個模型的混淆矩陣：

6. 如要查看執行這個 pipeline 時，系統建立的模型和端點，請前往模型區塊，然後按一下「automl-beans」模型。您應該會看到該模型部署至某個端點：

7. 您也可以點選 pipeline 圖表的「endpoint」構件，前往該頁面。

8. 除了在控制台查看 pipeline 圖表，您也可以使用 Vertex Pipelines 的歷程追蹤功能。

9. 歷程追蹤是指追蹤整個 pipeline 中建立的構件，可協助您瞭解構件的建立位置，以及構件在整個機器學習工作流程的用途。舉例來說，此 pipeline 建立了一個資料集，如要查看資料集的歷程追蹤，請依序點選資料集構件和「查看歷程」：

下圖顯示所有使用此構件的位置：

注意：請先等待 pipeline 的訓練工作開始，再於下方檢查進度。這個訓練工作的執行時間會比本實驗室的學習時間長，但只要成功提交訓練工作，就能獲得所有分數。 確認端對端機器學習 pipeline 訓練工作已啟動

步驟 4：比較不同 pipeline 執行作業的指標 (非必要)

• 如果您多次執行這個 pipeline，可能會想比較各執行作業的指標。您可以使用 aiplatform.get_pipeline_df() 方法存取執行作業中繼資料。我們會在此步驟，取得這個 pipeline 所有執行作業的中繼資料，並將這些資料載入 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，建構、執行端對端機器學習 pipeline，並取得 pipeline 中繼資料。

恭喜！

您在本實驗室建立並執行了表情符號 pipeline，也瞭解如何在 Vertex Pipelines，建構、執行端對端機器學習 pipeline，並取得 pipeline 中繼資料。

後續步驟/瞭解詳情

使用 Developer Relations 的 Codelab，在自己的 Google Cloud 專案嘗試類似的用法！

• Sara Robinson 的 Vertex Pipelines 簡介

Google Cloud 教育訓練與認證

協助您瞭解如何充分運用 Google Cloud 的技術。我們的課程會介紹專業技能和最佳做法，讓您可以快速掌握要領並持續進修。我們提供從基本到進階等級的訓練課程，並有隨選、線上和虛擬課程等選項，方便您抽空參加。認證可協助您驗證及證明自己在 Google Cloud 技術方面的技能和專業知識。

使用手冊上次更新日期：2024 年 10 月 7 日

實驗室上次測試日期：2024 年 10 月 7 日

Copyright 2025 Google LLC 保留所有權利。Google 和 Google 標誌是 Google LLC 的商標，其他公司和產品名稱則有可能是其關聯公司的商標。