在 Vertex Pipelines 上运行自定义模型训练

1. 概览

在本实验中，您将学习如何在 Vertex Pipelines 上使用 Kubeflow Pipelines SDK 运行自定义模型训练作业。

学习内容

您将了解如何：

  • 使用 Kubeflow Pipelines SDK 构建可扩缩的机器学习流水线
  • 创建使用 Vertex AI 代管式数据集并将在流水线中的 Vertex AI Training 上运行的自定义 Scikit-learn 模型训练作业并对其进行容器化
  • 在 Vertex Pipelines 中运行批量预测作业
  • 使用预构建的组件与通过 google_cloud_pipeline_components 库提供的 Vertex AI 服务进行交互

在 Google Cloud 上运行本实验的总费用约为 $5。

2. Vertex AI 简介

本实验使用 Vertex AI，这是 Google Cloud 上的端到端代管式机器学习平台。Vertex AI 将整个 Google Cloud 中的 Google 机器学习产品集成到无缝的开发体验中。除了模型训练和部署服务之外，Vertex AI 还包括各种 MLOps 产品，包括 Vertex Pipelines（本实验的重点）、模型监控、Feature Store 等。您可以在下图中查看所有 Vertex AI 产品。

Vertex 产品概览

如果您有任何反馈，请参阅支持页面。

机器学习流水线有何作用？

在深入了解之前，我们先来了解一下为什么要使用流水线。假设您要构建一个机器学习工作流，其中包含处理数据、训练模型、超参数调节、评估和模型部署。其中每个步骤都可能具有不同的依赖项，如果您将整个工作流视为单体式应用，这些依赖项可能会很难处理。开始扩展机器学习流程时，您可能希望与团队中的其他人共享机器学习工作流，以便他们运行该工作流并贡献代码。如果没有可靠、可重现的过程，这可能会变得非常困难。借助流水线，机器学习流程中的每个步骤都有自己的容器。这样，您就可以独立开发步骤，并以可重现的方式跟踪每个步骤的输入和输出。您还可以根据云环境中的其他事件安排或触发流水线的运行，例如在有新的训练数据时启动流水线运行。

要点：流水线可帮助您简化和重现机器学习工作流。

3. Cloud 环境设置

您需要一个启用了结算功能的 Google Cloud Platform 项目才能运行此 Codelab。如需创建项目，请按照此处的说明操作。

第 1 步：启动 Cloud Shell

在本实验中，您将使用 Cloud Shell 会话，这是一个由在 Google 云中运行的虚拟机托管的命令解释器。您可以在自己的计算机本地轻松运行此部分，但借助 Cloud Shell，每个人都可以在一致的环境中获得可重现的体验。完成实验后，您可以在自己的计算机上重试本部分。

为 Cloud Shell 授权

激活 Cloud Shell

在 Cloud 控制台的右上角，点击下方按钮以激活 Cloud Shell：

激活 Cloud Shell

如果您以前从未启动过 Cloud Shell，系统会显示一个中间屏幕（非首屏）来介绍 Cloud Shell。如果是这种情况，请点击继续（此后您将不会再看到此通知）。一次性屏幕如下所示：

Cloud Shell 设置

预配和连接到 Cloud Shell 只需花几分钟时间。

Cloud Shell 初始化

这个虚拟机装有您需要的所有开发工具。它提供了一个持久的 5GB 主目录，并且在 Google Cloud 中运行，大大增强了网络性能和身份验证。只需使用一个浏览器或 Google Chromebook 即可完成本 Codelab 中的大部分（甚至全部）工作。

在连接到 Cloud Shell 后，您应该会看到自己已通过身份验证，并且相关项目已设置为您的项目 ID：

在 Cloud Shell 中运行以下命令以确认您已通过身份验证：

gcloud auth list

您应该会在命令输出中看到如下内容：

Cloud Shell 输出

在 Cloud Shell 中运行以下命令，以确认 gcloud 命令了解您的项目：

gcloud config list project

命令输出

[core]
project = <PROJECT_ID>

如果不是上述结果，您可以使用以下命令进行设置：

gcloud config set project <PROJECT_ID>

命令输出

Updated property [core/project].

Cloud Shell 有一些环境变量，其中 GOOGLE_CLOUD_PROJECT 包含我们当前 Cloud 项目的名称。在本实验的各个位置，都会用到这个 ID。您可以通过运行以下命令来查看此信息：

echo $GOOGLE_CLOUD_PROJECT

第 2 步：启用 API

在后续步骤中，您将看到需要这些服务的位置（以及原因），但现在，请运行此命令，为您的项目授予对 Compute Engine、Container Registry 和 Vertex AI 服务的访问权限：

gcloud services enable compute.googleapis.com         \
                       containerregistry.googleapis.com  \
                       aiplatform.googleapis.com

这将生成类似于以下内容的成功消息：

Operation "operations/acf.cc11852d-40af-47ad-9d59-477a12847c9e" finished successfully.

第 3 步：创建 Cloud Storage 存储分区

如需在 Vertex AI 上运行训练作业，我们需要一个存储分区来存储我们保存的模型资源。存储分区必须是区域级的。我们在这里使用的是 us-central，但欢迎您使用其他区域（只需在整个实验中替换它即可）。如果您已有存储分区，可以跳过此步骤。

在 Cloud Shell 终端运行以下命令以创建存储分区：

BUCKET_NAME=gs://$GOOGLE_CLOUD_PROJECT-bucket
gsutil mb -l us-central1 $BUCKET_NAME

接下来，我们将为计算服务账号授予对此存储分区的访问权限。这将确保 Vertex Pipelines 具有将文件写入此存储分区所需的权限。运行以下命令以添加此权限：

gcloud projects describe $GOOGLE_CLOUD_PROJECT > project-info.txt
PROJECT_NUM=$(cat project-info.txt | sed -nre 's:.*projectNumber\: (.*):\1:p')
SVC_ACCOUNT="${PROJECT_NUM//\'/}-compute@developer.gserviceaccount.com"
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT --member serviceAccount:$SVC_ACCOUNT --role roles/storage.objectAdmin

第 4 步：创建 Vertex AI Workbench 实例

在 Cloud Console 的 Vertex AI 部分中，点击“Workbench”：

Vertex AI 菜单

然后，在用户管理的笔记本中，点击新建笔记本：

创建新的笔记本

然后选择 TensorFlow 企业版 2.3（带 LTS）实例类型，不带 GPU：

TFE 实例

使用默认选项，然后点击创建。

第 5 步：打开笔记本

创建实例后，选择打开 JupyterLab：

打开笔记本

4. Vertex Pipelines 设置

我们还需要安装几个额外的库，才能使用 Vertex Pipelines：

  • Kubeflow 流水线：这是我们将用于构建流水线的 SDK。Vertex Pipelines 支持运行使用 Kubeflow 流水线或 TFX 构建的流水线。
  • Google Cloud 流水线组件：此库提供预构建的组件，让您可以轻松地在流水线步骤中与 Vertex AI 服务进行交互。

第 1 步：创建 Python 笔记本并安装库

首先，在笔记本实例（可通过点击笔记本左上角的 + 图标访问）的“启动器”菜单中，选择 Python 3 创建笔记本：

创建 Python3 笔记本

您可以通过点击笔记本实例左上角的 + 号来访问启动器菜单。

如需安装我们将在本实验中使用的两项服务，请先在笔记本单元中设置用户标志：

USER_FLAG = "--user"

然后从笔记本运行以下命令：

!pip3 install {USER_FLAG} google-cloud-aiplatform==1.7.0 --upgrade
!pip3 install {USER_FLAG} kfp==1.8.9 google-cloud-pipeline-components==0.2.0

安装这些软件包后，您需要重启内核：

import os

if not os.getenv("IS_TESTING"):
    # Automatically restart kernel after installs
    import IPython

    app = IPython.Application.instance()
    app.kernel.do_shutdown(True)

最后，检查您是否已正确安装软件包。KFP SDK 版本应不低于 1.8：

!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 和之前创建的存储分区。接下来，我们将为每个元素创建变量。

如果您不知道项目 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)

否则，请在此处进行设置：

if PROJECT_ID == "" or PROJECT_ID is None:
    PROJECT_ID = "your-project-id"  # @param {type:"string"}

然后创建一个变量来存储存储分区名称。如果您在本实验中创建了该 API，则以下配置将会起作用。否则，您需要手动进行设置：

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

第 3 步：导入库

添加以下代码，以导入我们将在此 Codelab 中使用的库：

from kfp.v2 import compiler, dsl
from kfp.v2.dsl import pipeline

from google.cloud import aiplatform
from google_cloud_pipeline_components import aiplatform as gcc_aip

第 4 步：定义常量

在构建流水线之前，我们需要做的最后一件事是定义一些常量变量。PIPELINE_ROOT 是 Cloud Storage 路径，我们的流水线创建的制品将写入其中。我们在此处使用 us-central1 作为区域，但如果您在创建存储分区时使用了其他区域，请更新以下代码中的 REGION 变量：

PATH=%env PATH
%env PATH={PATH}:/home/jupyter/.local/bin
REGION="us-central1"

PIPELINE_ROOT = f"{BUCKET_NAME}/pipeline_root/"
PIPELINE_ROOT

运行上述代码后，您应该会看到输出的流水线根目录。这是将写入流水线中的制品的 Cloud Storage 位置。其格式为 gs://YOUR-BUCKET-NAME/pipeline_root/

5. 配置自定义模型训练作业

在设置流水线之前，我们需要为自定义模型训练作业编写代码。为了训练模型，我们将使用 UCI 机器学习干豆数据集，数据集为：KOKLU, M. 和 OZKAN, I.A.,(2020)，“Multiclass Classification of Dry Beans Using Computer Vision and Machine Learning Techniques”。”。DOI。

我们的第一个流水线步骤是使用包含此 Bean 数据版本的 BigQuery 表，在 Vertex AI 中创建代管式数据集。数据集将作为输入传递给训练作业。在我们的训练代码中，我们有权访问环境变量来访问此代管式数据集。

以下是设置自定义训练作业的方法：

  • 编写一个 Scikit-learn DecisionTreeClassifier 模型来对数据中的 Bean 类型进行分类
  • 将训练代码封装在 Docker 容器中，并将其推送到 Container Registry

之后，我们将能够直接从流水线启动 Vertex AI Training 作业。让我们开始吧！

第 1 步：在 Docker 容器中定义训练代码

在您的 Notebooks 实例中，打开启动器并选择“Terminal”（终端）：

打开终端

然后运行以下命令，以设置用于添加容器化代码的目录：

mkdir traincontainer
cd traincontainer
touch Dockerfile

mkdir trainer
touch trainer/train.py

运行这些命令后，您应该会看到在左侧创建一个名为 traincontainer/ 的目录（您可能需要点击刷新图标才能看到该目录）。您将在 traincontainer/ 目录中看到以下内容：

+ Dockerfile
+ trainer/
    + train.py

将代码容器化的第一步是创建一个 Dockerfile。Dockerfile 中将包含运行映像所需的所有命令。它将安装我们使用的所有库，并为训练代码设置入口点。打开您刚刚创建的 Dockerfile，并添加以下内容：

FROM gcr.io/deeplearning-platform-release/sklearn-cpu.0-23
WORKDIR /

# Copies the trainer code to the docker image.
COPY trainer /trainer

RUN pip install sklearn google-cloud-bigquery joblib pandas google-cloud-storage

# Sets up the entry point to invoke the trainer.
ENTRYPOINT ["python", "-m", "trainer.train"]

如需在笔记本实例中修改文件时保存文件，您可以使用 ctrl+s。

接下来，打开 train.py 文件。我们将在此处添加训练代码。将以下内容复制到 train.py。这会从我们的代管式数据集中检索数据，将其放入 Pandas DataFrame 中，训练 Scikit-learn 模型，然后将经过训练的模型上传到 Cloud Storage：

from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import roc_curve
from sklearn.model_selection import train_test_split
from google.cloud import bigquery
from google.cloud import storage
from joblib import dump

import os
import pandas as pd

bqclient = bigquery.Client()
storage_client = storage.Client()

def download_table(bq_table_uri: str):
    prefix = "bq://"
    if bq_table_uri.startswith(prefix):
        bq_table_uri = bq_table_uri[len(prefix):]

    table = bigquery.TableReference.from_string(bq_table_uri)
    rows = bqclient.list_rows(
        table,
    )
    return rows.to_dataframe(create_bqstorage_client=False)

# These environment variables are from Vertex AI managed datasets
training_data_uri = os.environ["AIP_TRAINING_DATA_URI"]
test_data_uri = os.environ["AIP_TEST_DATA_URI"]

# Download data into Pandas DataFrames, split into train / test
df = download_table(training_data_uri)
test_df = download_table(test_data_uri)
labels = df.pop("Class").tolist()
data = df.values.tolist()
test_labels = test_df.pop("Class").tolist()
test_data = test_df.values.tolist()

# Define and train the Scikit model
skmodel = DecisionTreeClassifier()
skmodel.fit(data, labels)
score = skmodel.score(test_data, test_labels)
print('accuracy is:',score)

# Save the model to a local file
dump(skmodel, "model.joblib")

# Upload the saved model file to GCS
bucket = storage_client.get_bucket("YOUR_GCS_BUCKET")
model_directory = os.environ["AIP_MODEL_DIR"]
storage_path = os.path.join(model_directory, "model.joblib")
blob = storage.blob.Blob.from_string(storage_path, client=storage_client)
blob.upload_from_filename("model.joblib")

然后，导航回笔记本并运行以下命令，将上述脚本中的 YOUR_GCS_BUCKET 替换为您的 Cloud Storage 存储分区的名称：

BUCKET = BUCKET_NAME[5:] # Trim the 'gs://' before adding to train script
!sed -i -r 's@YOUR_GCS_BUCKET@'"$BUCKET"'@' traincontainer/trainer/train.py

如果您愿意，也可以手动执行此操作。如果这样做，在更新脚本时，请确保不要在存储分区名称中包含 gs://。

现在，我们的训练代码位于 Docker 容器中，我们可以在云端运行训练了。

第 2 步：将容器推送到 Container Registry

训练代码完成后，我们可以将其推送到 Google Container Registry。稍后，在配置流水线的训练组件时，我们会将 Vertex Pipelines 指向此容器。

返回到终端，从 traincontainer/ 目录的根目录，使用 Container Registry 上容器映像的 URI 定义一个变量。

PROJECT_ID=$(gcloud config get-value project)
IMAGE_URI="gcr.io/$PROJECT_ID/scikit:v1"

然后，运行以下命令来构建容器：

docker build ./ -t $IMAGE_URI

最后，将容器推送到 Container Registry：

docker push $IMAGE_URI

前往 Cloud 控制台的 Container Registry 部分，验证您的容器是否存在。请参见以下示例：

Container Registry

6. 配置批量预测作业

流水线的最后一步将运行批量预测作业。为此，我们需要在 Cloud Storage 中提供一个 CSV 文件，其中包含我们想要进行预测的样本。我们将在笔记本中创建此 CSV 文件，并使用 gsutil 命令行工具将其复制到 Cloud Storage。

将批量预测示例复制到 Cloud Storage

以下文件包含来自 Bean 数据集中每个类的 3 个示例。以下示例不包括 Class 列，因为这是我们的模型要预测的内容。运行以下命令，在您的笔记本中本地创建此 CSV 文件：

%%writefile batch_examples.csv
Area,Perimeter,MajorAxisLength,MinorAxisLength,AspectRation,Eccentricity,ConvexArea,EquivDiameter,Extent,Solidity,roundness,Compactness,ShapeFactor1,ShapeFactor2,ShapeFactor3,ShapeFactor4
23288,558.113,207.567738,143.085693,1.450653336,0.7244336162,23545,172.1952453,0.8045881703,0.9890847314,0.9395021523,0.8295857874,0.008913077034,0.002604069884,0.6882125787,0.9983578734
23689,575.638,205.9678003,146.7475015,1.403552348,0.7016945718,24018,173.6714472,0.7652721693,0.9863019402,0.8983750474,0.8431970773,0.00869465998,0.002711119968,0.7109813112,0.9978994889
23727,559.503,189.7993849,159.3717704,1.190922235,0.5430731512,24021,173.8106863,0.8037601626,0.9877607094,0.952462433,0.9157600082,0.007999299741,0.003470231343,0.8386163926,0.9987269085
31158,641.105,212.0669751,187.1929601,1.132879009,0.4699241567,31474,199.1773023,0.7813134733,0.989959967,0.9526231013,0.9392188582,0.0068061806,0.003267009878,0.8821320637,0.9993488983
32514,649.012,221.4454899,187.1344232,1.183349841,0.5346736437,32843,203.4652564,0.7849831,0.9899826447,0.9700068737,0.9188051492,0.00681077351,0.002994124691,0.8442029022,0.9989873701
33078,659.456,235.5600775,178.9312328,1.316483846,0.6503915309,33333,205.2223615,0.7877214708,0.9923499235,0.9558229607,0.8712102818,0.007121351881,0.002530662194,0.7590073551,0.9992209221
33680,683.09,256.203255,167.9334938,1.525623324,0.7552213942,34019,207.081404,0.80680321,0.9900349805,0.9070392732,0.8082699962,0.007606985006,0.002002710402,0.6533003868,0.9966903078
33954,716.75,277.3684803,156.3563259,1.773951126,0.825970469,34420,207.9220419,0.7994819873,0.9864613597,0.8305492781,0.7496238998,0.008168948587,0.001591181142,0.5619359911,0.996846984
36322,719.437,272.0582306,170.8914975,1.591993952,0.7780978465,36717,215.0502424,0.7718560075,0.9892420405,0.8818487005,0.7904566678,0.007490177594,0.001803782407,0.6248217437,0.9947124371
36675,742.917,285.8908964,166.8819538,1.713132487,0.8119506999,37613,216.0927123,0.7788277766,0.9750618137,0.8350248381,0.7558572692,0.0077952528,0.001569528272,0.5713202115,0.9787472145
37454,772.679,297.6274753,162.1493177,1.835514817,0.8385619338,38113,218.3756257,0.8016695205,0.9827093118,0.7883332637,0.7337213257,0.007946480356,0.001420623993,0.5383469838,0.9881438654
37789,766.378,313.5680678,154.3409867,2.031657789,0.8704771226,38251,219.3500608,0.7805870567,0.9879218844,0.8085170916,0.6995293312,0.008297866252,0.001225659709,0.4893412853,0.9941740339
47883,873.536,327.9986493,186.5201272,1.758516115,0.822571799,48753,246.9140116,0.7584464543,0.9821549443,0.7885506623,0.7527897207,0.006850002074,0.00135695419,0.5666923636,0.9965376533
49777,861.277,300.7570338,211.6168613,1.42123379,0.7105823885,50590,251.7499649,0.8019106536,0.9839296304,0.843243269,0.8370542883,0.00604208839,0.001829706116,0.7006598815,0.9958014989
49882,891.505,357.1890036,179.8346914,1.986207449,0.8640114945,51042,252.0153467,0.7260210171,0.9772736178,0.7886896753,0.7055518063,0.007160679276,0.001094585314,0.4978033513,0.9887407248
53249,919.923,325.3866286,208.9174205,1.557489212,0.7666552108,54195,260.3818974,0.6966846347,0.9825445152,0.7907120655,0.8002231025,0.00611066177,0.001545654241,0.6403570138,0.9973491406
61129,964.969,369.3481688,210.9473449,1.750902193,0.8208567513,61796,278.9836198,0.7501135067,0.9892064211,0.8249553283,0.7553404711,0.006042110436,0.001213219664,0.5705392272,0.9989583843
61918,960.372,353.1381442,224.0962377,1.575832543,0.7728529173,62627,280.7782864,0.7539207091,0.9886790043,0.8436218213,0.7950947556,0.005703319619,0.00140599258,0.6321756704,0.9962029945
141953,1402.05,524.2311633,346.3974998,1.513380332,0.7505863011,143704,425.1354762,0.7147107987,0.9878152313,0.9074598849,0.8109694843,0.003692991084,0.0009853172185,0.6576715044,0.9953071199
145285,1440.991,524.9567463,353.0769977,1.486805285,0.7400216694,146709,430.0960442,0.7860466375,0.9902937107,0.8792413513,0.8192980608,0.003613289371,0.001004269363,0.6712493125,0.9980170255
146153,1476.383,526.1933264,356.528288,1.475881001,0.7354662103,149267,431.3789276,0.7319360978,0.9791380546,0.8425962592,0.8198107159,0.003600290972,0.001003163512,0.6720896099,0.991924286

然后，将该文件复制到您的 Cloud Storage 存储分区：

!gsutil cp batch_examples.csv $BUCKET_NAME

我们将在下一步定义流水线时引用此文件。

7. 使用预构建组件构建流水线

现在，我们的训练代码已在云端，接下来可以从流水线调用它。我们要定义的流水线将使用我们之前安装的 google_cloud_pipeline_components 库中的三个预构建组件。这些预定义组件简化了我们为设置流水线而编写的代码，并使我们能够使用模型训练和批量预测等 Vertex AI 服务。

以下是我们的三步流水线将执行的操作：

  • 在 Vertex AI 中创建代管式数据集
  • 使用我们设置的自定义容器在 Vertx AI 上运行训练作业
  • 使用我们经过训练的 Scikit-learn 分类模型运行批量预测作业

第 1 步：定义流水线

由于我们使用的是预构建的组件，因此可以在流水线定义中设置整个流水线。将以下内容添加到笔记本单元：

@pipeline(name="automl-beans-custom",
                  pipeline_root=PIPELINE_ROOT)
def pipeline(
    bq_source: str = "bq://sara-vertex-demos.beans_demo.large_dataset",
    bucket: str = BUCKET_NAME,
    project: str = PROJECT_ID,
    gcp_region: str = REGION,
    bq_dest: str = "",
    container_uri: str = "",
    batch_destination: str = ""
):
    dataset_create_op = gcc_aip.TabularDatasetCreateOp(
        display_name="tabular-beans-dataset",
        bq_source=bq_source,
        project=project,
        location=gcp_region
    )

    training_op = gcc_aip.CustomContainerTrainingJobRunOp(
        display_name="pipeline-beans-custom-train",
        container_uri=container_uri,
        project=project,
        location=gcp_region,
        dataset=dataset_create_op.outputs["dataset"],
        staging_bucket=bucket,
        training_fraction_split=0.8,
        validation_fraction_split=0.1,
        test_fraction_split=0.1,
        bigquery_destination=bq_dest,
        model_serving_container_image_uri="us-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.0-24:latest",
        model_display_name="scikit-beans-model-pipeline",
        machine_type="n1-standard-4",
    )
    batch_predict_op = gcc_aip.ModelBatchPredictOp(
        project=project,
        location=gcp_region,
        job_display_name="beans-batch-predict",
        model=training_op.outputs["model"],
        gcs_source_uris=["{0}/batch_examples.csv".format(BUCKET_NAME)],
        instances_format="csv",
        gcs_destination_output_uri_prefix=batch_destination,
        machine_type="n1-standard-4"
    )

第 2 步：编译并运行流水线

定义流水线后，您就可以对其进行编译了。以下命令将生成一个用于运行流水线的 JSON 文件：

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

接下来，创建一个 TIMESTAMP 变量。我们将在作业 ID 中使用此名称：

from datetime import datetime

TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")

然后定义流水线作业，传入一些特定于项目的参数：

pipeline_job = aiplatform.PipelineJob(
    display_name="custom-train-pipeline",
    template_path="custom_train_pipeline.json",
    job_id="custom-train-pipeline-{0}".format(TIMESTAMP),
    parameter_values={
        "project": PROJECT_ID,
        "bucket": BUCKET_NAME,
        "bq_dest": "bq://{0}".format(PROJECT_ID),
        "container_uri": "gcr.io/{0}/scikit:v1".format(PROJECT_ID),
        "batch_destination": "{0}/batchpredresults".format(BUCKET_NAME)
    },
    enable_caching=True,
)

最后，运行该作业以创建新的流水线执行作业：

pipeline_job.submit()

运行此单元后，您应该会看到日志，其中包含可在控制台中查看流水线运行的链接：

流水线作业日志

前往该链接。您也可以打开流水线信息中心来访问该信息中心。完成后，您的流水线应如下所示：

已完成的简介流水线

此流水线将运行 5-10 分钟，但在其完成之前您可以继续执行下一步。接下来，您将详细了解每个流水线步骤中会发生什么。

8. 了解流水线执行情况

我们来深入了解一下每个流水线步骤。

流水线第 1 步：创建代管式数据集

流水线的第一步是在 Vertex AI 中创建代管式数据集。如果您在控制台的“流水线”部分中点击以下数据集链接：

指向流水线中的数据集的链接

您会在 Vertex AI 中看到您的数据集，其中包括指向 BigQuery 中数据源的链接以及数据集中不同列的相关信息。将代管式数据集上传到 Vertex AI 后，该数据集可用于训练 AutoML 或自定义模型。

对于使用代管式数据集的自定义模型作业，Vertex AI 会将特殊的环境变量传递给训练作业，并负责将数据拆分为训练集和测试集。我们将在下一个流水线步骤中用到它。

流水线第 2 步：在 Vertex AI Training 中训练模型

在自定义训练作业运行时，您可以直接在 Vertex Pipelines 控制台中点击查看日志：

自定义训练日志

您还可以在 Vertex AI Training 信息中心中查看自定义训练作业的详细信息。训练作业完成后，系统会在 Vertex AI 中创建模型资源。然后，我们可以将此模型部署到端点以进行在线预测，或创建批量预测作业，我们将在下一个流水线步骤中执行此操作。

流水线第 3 步：对模型运行批量预测作业

最后，我们的管道将对我们通过 CSV 文件传入的样本进行预测。批量预测作业完成后，Vertex AI 会将 CSV 文件写入我们在 Cloud Storage 中指定的位置。当此流水线步骤开始运行时，您可以导航到 Vertex AI 控制台的“批量预测”部分，以查看创建的作业。

作业完成后，点击作业以查看批量预测的 Cloud Storage 网址：

批量预测作业

点击该链接可前往 Cloud Storage 目录（您可以在其中找到预测结果），然后点击以下载其中一个 prediction.results 文件。在该文件中，您应该会看到如下所示的行：

{"instance": [33954.0, 716.75, 277.3684803, 156.3563259, 1.773951126, 0.825970469, 34420.0, 207.9220419, 0.7994819873, 0.9864613597, 0.8305492781, 0.7496238998, 0.008168948587, 0.001591181142, 0.5619359911, 0.996846984], "prediction": "HOROZ"}

这包括特定实例的特征值，以及我们的模型预测的类别。在本示例中，我们的模型认为这是一个“HOROZ”bean。

至此，您已完成本实验！

🎉 恭喜！🎉

您学习了如何使用 Vertex AI 执行以下操作：

  • 使用 Kubeflow Pipelines SDK 构建具有自定义组件的端到端流水线
  • 在 Vertex Pipelines 上运行流水线并使用 SDK 启动流水线运行
  • 在控制台中查看和分析 Vertex Pipelines 图
  • 使用预构建的流水线组件将 Vertex AI 服务添加到流水线
  • 安排周期性流水线作业

如需详细了解 Vertex 的不同部分，请参阅相关文档。

9. 清理

为了不向您收费，因此建议您删除在整个实验过程中创建的资源。

第 1 步：停止或删除 Notebooks 实例

如果您想继续使用在本实验中创建的笔记本，建议您在不使用时将其关闭。在 Cloud 控制台的 Notebooks 界面中，选择相应笔记本，然后选择停止。如果您想完全删除实例，请选择删除：

可停止实例

第 2 步：删除 Cloud Storage 存储分区

如需删除存储桶，请使用 Cloud Console 中的导航菜单，浏览到“存储空间”，选择您的存储桶，然后点击“删除”：

删除存储空间