创建训练流水线

借助训练流水线，您可以执行自定义机器学习 (ML) 训练，并根据训练输出自动创建 Model 资源。

创建流水线之前

在 Vertex AI 上创建训练流水线之前，您需要创建一个 Python 训练应用或自定义容器，以定义要在 Vertex AI 上运行的训练代码和依赖项。如果您使用 TensorFlow、scikit-learn 或 XGBoost 创建 Python 训练应用，则可以使用我们的预构建容器来运行您的代码。如果您不确定选择哪个选项，请参阅训练代码要求了解详情。

训练流水线选项

训练流水线封装了训练作业和额外步骤。本指南介绍两种不同的训练流水线：

启动 CustomJob 并将生成的模型上传到 Vertex AI
启动超参数调节作业并将生成的模型上传到 Vertex AI

此外，您还可以在训练流水线中使用代管式数据集。详细了解如何配置训练流水线配置以使用代管式数据集。

`CustomJob` 包含的内容

创建自定义作业时，您需要指定 Vertex AI 运行训练代码所需的设置，包括：

用于单节点训练的一个工作器池 (WorkerPoolSpec)，或用于分布式训练的多个工作器池
用于配置作业调度的可选设置 (Scheduling)、为训练代码设置特定的环境变量、使用自定义服务账号和使用 VPC 网络对等互连

在工作器池中，您可以指定以下设置：

机器类型和加速器
工作器池运行的训练代码的配置：Python 训练应用 (PythonPackageSpec) 或自定义容器 (ContainerSpec)

如果您要在 Vertex AI 训练流水线之外创建独立的自定义作业，请参阅自定义作业指南。

配置流水线以使用代管式数据集

在训练流水线中，您可以将自定义训练作业或超参数调节作业配置为使用代管式数据集。通过代管式数据集，您可以使用训练应用和模型管理数据集。

如需在训练流水线中使用代管式数据集，请执行以下操作：

创建数据集。
更新训练应用以使用代管式数据集。如需了解详情，请参阅 Vertex AI 如何将数据集传递给训练应用。
在创建训练流水线时指定代管式数据集。例如，如果您使用 REST API 创建训练流水线，请在 inputDataConfig 部分中指定数据集设置。

您必须在创建数据集的区域中创建训练流水线。

如需了解详情，请参阅 TrainingPipeline 的API 参考文档。

配置分布式训练

在训练流水线中，您可以通过指定多个工作器池来配置用于分布式训练的自定义训练作业或超参数调节作业。

此页面上的所有示例均为使用一个工作器池的单副本训练作业。要修改它们以进行分布式训练，请执行以下操作：

使用第一个工作器池配置主副本，并将副本计数设置为 1。
添加更多工作器池来配置工作器副本、参数服务器副本或评估器副本（如果您的机器学习框架支持分布式训练的这些额外集群任务）。

详细了解如何使用分布式训练。

自定义作业和模型上传

此训练流水线封装了一个自定义的作业以及额外的便捷步骤，可以更轻松地在训练后将模型部署到 Vertex AI。此训练流水线执行两个主要任务：

训练流水线会创建一个 CustomJob 资源。自定义作业使用您指定的计算资源运行训练应用。
自定义作业完成后，训练流水线会在您为 Cloud Storage 存储桶指定的输出目录中查找训练应用创建的模型工件。训练流水线使用这些工件来创建模型资源，帮助您为模型部署做好准备。

您可以通过两种方式设置模型工件的位置：

如果您为训练作业设置了 baseOutputDirectory，请使用 Vertex AI 设置的 $AIP_MODEL_DIR 环境变量确保训练代码将模型工件保存到该位置。训练作业完成后，Vertex AI 会在 gs://BASE_OUTPUT_DIRECTORY/model 中搜索生成的模型工件。
注意：如果您使用的是 Python 版 Vertex AI SDK，则可以省略 base_output_dir 特性。在这种情况下，Vertex AI 会将模型工件输出到暂存目录中带有时间戳的目录。如需了解详情，请参阅：
如果您设置了 modelToUpload.artifactUri 字段，训练流水线会从该 URI 上传模型工件。如果您未设置 baseOutputDirectory，则必须设置此字段。

如果您同时指定 baseOutputDirectory 和 modelToUpload.artifactUri，则 Vertex AI 会使用 modelToUpload.artifactUri。

要创建此类训练流水线，请执行以下操作：

控制台

在 Google Cloud 控制台的 Vertex AI 部分中，转到训练流水线页面。

转到“训练流水线”
点击创建以打开训练新模型窗格。

注意：您可以在浏览器中输入 model.new，以直接转到模型创建页面。
在训练方法步骤中，指定以下设置：
1. 如果您要使用代管式数据集进行训练，请指定数据集和注释集。
  
  否则，在数据集下拉列表中，选择无代管式数据集。
2. 选择自定义训练（高级）。
点击继续。
在模型详情步骤中，选择训练新模型或训练新版本。如果您选择训练新模型，请为模型输入一个名称 MODEL_NAME。点击继续。
在训练容器步骤中，指定以下设置：
1. 选择要使用预构建容器还是自定义容器进行训练。
2. 根据您的选择，执行以下相应操作：
  - 如果要使用预构建容器进行训练，请向 Vertex AI 提供使用您上传到 Cloud Storage 的训练软件包所需的信息：
    1. 使用模型框架和模型框架版本下拉列表指定要使用的预构建容器。
    2. 在软件包位置字段中，指定您创建并上传的 Python 训练应用的 Cloud Storage URI。此文件通常以 .tar.gz 结尾。
    3. 在 Python 模块字段中，输入训练应用的入口点的模块名称。
  - 如果要使用自定义容器进行训练，请在容器映像字段中指定容器映像的 Artifact Registry 或 Docker Hub URI。
3. 在 Model output directory 字段中，指定您有权访问的存储桶中的目录的 Cloud Storage URI。目录不需要已经存在。
  
  此值会在 baseOutputDirectory API 字段中传递给 Vertex AI，该字段设置训练应用在其运行时可以访问的几个环境变量。
  
  在训练结束时，Vertex AI 会在此 URI 的子目录中查找模型工件，用于创建 Model。（此子目录可作为 AIP_MODEL_DIR 环境变量提供给训练代码。）
  
  如果您不使用超参数调节，Vertex AI 会在 BASE_OUTPUT_DIRECTORY/model/ 中查找模型工件。
4. 可选：在参数字段中，您可以指定 Vertex AI 开始运行训练代码时使用的参数。所有参数的总长度上限为 10 万个字符。这些参数的行为因您使用的容器类型而异：
  - 如果您使用的是预构建容器，Vertex AI 会将参数作为命令行标志传递给 Python 模块。
  - 如果您使用的是自定义容器，则 Vertex AI 会使用参数替换容器的 CMD 指令。
点击继续。
在超参数调节步骤中，确保未选中启用超参数调节复选框。点击继续。
在计算和价格步骤中，指定以下设置：
1. 在区域下拉列表中，选择一个支持自定义训练的区域。
2. 在工作器池 0 部分中，指定用于训练的计算资源。
  
  如果指定加速器，请确保所选加速器类型在所选区域可用。
  
  如果您要执行分布式训练，则点击 Add more worker pools，并为您需要的每个额外工作器池指定一组额外的计算资源。
点击继续。
在预测容器步骤中，指定以下设置：
1. 选择要使用预构建容器还是自定义容器通过经过训练的模型进行预测。
2. 根据您的选择，执行以下相应操作：
  - 如果您要使用预构建容器来进行预测，则使用模型框架、模型框架版本和加速器类型字段来选择要用来进行预测的预构建预测容器。
    
    将模型框架和模型框架版本与用于训练的机器学习框架匹配。如果您希望稍后使用 GPU 进行在线预测或批量预测，请仅指定加速器类型。
  - 如果您要使用自定义容器来执行预测，请执行以下操作：
    1. 在容器映像字段中，指定您的容器映像的 Artifact Registry URI。
    2. 或者，您可以指定命令来替换容器的 ENTRYPOINT 指令。
3. 模型目录 (Model directory) 字段包含您之前在训练容器步骤的 Model output directory 字段中设置的值。更改这两个字段具有相同的效果。如需详细了解此字段，请参阅之前的说明。
4. 将预测架构部分中的字段留空。
点击开始训练以启动自定义训练流水线。

REST

通过以下代码示例使用 trainingPipeline 资源的 create 方法创建训练流水线。

注意：如果要设置此流水线以创建新模型版本，则可以选择在 trainingPipeline 字段中添加 PARENT_MODEL。

如需了解详情，请参阅使用 Vertex AI Model Registry 进行模型版本控制。

在使用任何请求数据之前，请先进行以下替换：

LOCATION_ID：运行训练代码并存储 Model 的区域。
PROJECT_ID：您的项目 ID。
TRAINING_PIPELINE_NAME：必填。trainingPipeline 的显示名。
如果训练应用使用 Vertex AI 数据集，请指定以下内容：
- DATASET_ID：数据集的 ID。
- ANNOTATIONS_FILTER：根据您指定的注释过滤数据集。
- ANNOTATION_SCHEMA_URI：根据指定的注释架构 URI 过滤数据集。
- 使用以下任一选项来指定数据项拆分为训练集、验证集和测试集的方式。
  - 如需根据定义每个集合大小的比例来拆分数据集，请指定以下内容：
    - TRAINING_FRACTION：用于训练模型的数据集比例。
    - VALIDATION_FRACTION：用于验证模型的数据集比例。
    - TEST_FRACTION：用于评估模型的数据集比例。
  - 如需根据过滤条件拆分数据集，请指定以下内容：
    - TRAINING_FILTER：将数据集过滤为用于训练模型的数据项。
    - VALIDATION_FILTER：将数据集过滤为用于验证模型的数据项。
    - TEST_FILTER：将数据集过滤为用于评估模型的数据项。
  - 如需使用预定义的拆分，请指定以下内容：
    - PREDEFINED_SPLIT_KEY：用于拆分数据集的列的名称。此列中可接受的值包括“training”“validation”和“test”。
  - 如需根据数据项的时间戳拆分数据集，请指定以下内容：
    - TIMESTAMP_TRAINING_FRACTION：用于训练模型的数据集比例。
    - TIMESTAMP_VALIDATION_FRACTION：用于验证模型的数据集比例。
    - TIMESTAMP_TEST_FRACTION：用于评估模型的数据集比例。
    - TIMESTAMP_SPLIT_KEY：用于拆分数据集的时间戳列的名称。
- OUTPUT_URI_PREFIX：Vertex AI 在训练数据集被拆分为训练集、验证集和测试集后将其导出的目标 Cloud Storage 位置。
定义自定义训练作业：
- MACHINE_TYPE：机器类型。请参阅可用于训练的机器类型。
- ACCELERATOR_TYPE：（可选）要附加到每个试验的加速器类型。
- ACCELERATOR_COUNT：（可选）要附加到每个试验的加速器数量。
- REPLICA_COUNT：每次试验要使用的工作器副本数。
- 如果训练应用在自定义容器中运行，请指定以下内容：
  - CUSTOM_CONTAINER_IMAGE_URI：要在每个工作器副本上运行的容器映像的 Artifact Registry、Container Registry 或 Docker Hub URI。
  - CUSTOM_CONTAINER_COMMAND：（可选）启动容器时要调用的命令。此命令会替换容器的默认入口点。
  - CUSTOM_CONTAINER_ARGS：（可选）启动容器时要传递的参数。所有参数的总长度上限为 10 万个字符。
- 如果训练应用是在预构建容器中运行的 Python 软件包，请指定以下内容：
  - PYTHON_PACKAGE_EXECUTOR_IMAGE_URI：运行提供的 Python 软件包的容器映像的 URI。请参阅可用于训练的预构建容器。
  - PYTHON_PACKAGE_URIS：Python 软件包文件的 Cloud Storage 位置，这些文件是训练程序及其从属软件包。软件包 URI 的数量上限为 100。
  - PYTHON_MODULE：安装软件包后要运行的 Python 模块名称。
  - PYTHON_PACKAGE_ARGS：（可选）要传递给 Python 模块的命令行参数。所有参数的总长度上限为 10 万个字符。
- TIMEOUT：（可选）作业的最长运行时间。
MODEL_NAME：TrainingPipeline 上传（创建）的模型的显示名。
MODEL_DESCRIPTION：模型的说明。
IMAGE_URI：用于运行预测的容器映像的 URI。例如us-docker.pkg.dev/vertex-ai/prediction/tf2-cpu.2-1:latest。使用预构建容器或自定义容器。
modelToUpload.labels：用于组织模型的任何键值对。例如：
- "env": "prod"
- "tier": "backend"
指定要应用于此训练流水线的任何标签的 LABEL_NAME 和 LABEL_VALUE。

HTTP 方法和网址：

POST https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/trainingPipelines

请求 JSON 正文：

{
  "displayName": "TRAINING_PIPELINE_NAME",
  "inputDataConfig": {
    "datasetId": DATASET_ID,
    "annotationsFilter": ANNOTATIONS_FILTER,
    "annotationSchemaUri": ANNOTATION_SCHEMA_URI,

    // Union field split can be only one of the following:
    "fractionSplit": {
      "trainingFraction": TRAINING_FRACTION,
      "validationFraction": VALIDATION_FRACTION,
      "testFraction": TEST_FRACTION
    },
    "filterSplit": {
      "trainingFilter": TRAINING_FILTER,
      "validationFilter": VALIDATION_FILTER,
      "testFilter": TEST_FILTER
    },
    "predefinedSplit": {
      "key": PREDEFINED_SPLIT_KEY
    },
    "timestampSplit": {
      "trainingFraction": TIMESTAMP_TRAINING_FRACTION,
      "validationFraction": TIMESTAMP_VALIDATION_FRACTION,
      "testFraction": TIMESTAMP_TEST_FRACTION,
      "key": TIMESTAMP_SPLIT_KEY
    }
    // End of list of possible types for union field split.
    "gcsDestination": {
      "outputUriPrefix": OUTPUT_URI_PREFIX
    }
  },
  "trainingTaskDefinition": "gs://google-cloud-aiplatform/schema/trainingjob/definition/custom_task_1.0.0.yaml",
  "trainingTaskInputs": {
    "workerPoolSpecs": [
        {
          "machineSpec": {
            "machineType": MACHINE_TYPE,
            "acceleratorType": ACCELERATOR_TYPE,
            "acceleratorCount": ACCELERATOR_COUNT
          },
          "replicaCount": REPLICA_COUNT,

          // Union field task can be only one of the following:
          "containerSpec": {
            "imageUri": CUSTOM_CONTAINER_IMAGE_URI,
            "command": [
              CUSTOM_CONTAINER_COMMAND
            ],
            "args": [
              CUSTOM_CONTAINER_ARGS
            ]
          },
          "pythonPackageSpec": {
            "executorImageUri": PYTHON_PACKAGE_EXECUTOR_IMAGE_URI,
            "packageUris": [
              PYTHON_PACKAGE_URIS
            ],
            "pythonModule": PYTHON_MODULE,
            "args": [
              PYTHON_PACKAGE_ARGS
            ]
          }
          // End of list of possible types for union field task.
        }
      ],
      "scheduling": {
        "TIMEOUT": TIMEOUT
      }
    }
  },
  "modelToUpload": {
    "displayName": "MODEL_NAME",
    "predictSchemata": {},
    "containerSpec": {
      "imageUri": "IMAGE_URI"
    }
  },
  "labels": {
    LABEL_NAME_1": LABEL_VALUE_1,
    LABEL_NAME_2": LABEL_VALUE_2
  }
}

如需发送请求，请选择以下方式之一：

curl

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI，或者使用了 Cloud Shell，这会使您自动登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/trainingPipelines"

PowerShell

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/trainingPipelines" | Select-Object -Expand Content

响应包含有关规范的信息以及 TRAININGPIPELINE_ID。

响应


{
  "name": "projects/PROJECT_ID/locations/LOCATION_ID/trainingPipelines/TRAININGPIPELINE_ID",
  "displayName": "TRAINING_PIPELINE_NAME",
  "input_data_config" {
    "dataset_id": "1234567891011121314"
    "gcs_destination" {
      "output_uri_prefix": "gs://BUCKET_NAME/data/20200915191342"
    }
    "annotation_schema_uri": "gs://google-cloud-aiplatform/schema/dataset/annotation/image_classification_1.0.0.yaml"
  },
  "trainingTaskDefinition": "gs://google-cloud-aiplatform/schema/trainingjob/definition/custom_task_1.0.0.yaml",
  "trainingTaskInputs": {
    "workerPoolSpecs": [
      {
        "machineSpec": {
          "machineType": "n1-standard-4"
        },
        "replicaCount": "1",
        "pythonPackageSpec": {
          "executorImageUri": "us-docker.pkg.dev/vertex-ai/training/training-tf-cpu.2-1:latest",
          "packageUris": [
            "gs://BUCKET_NAME/training/hello-custom-training-1.0.tar.gz"
          ],
          "pythonModule": "trainer.task",
          "args": [
            "--model-dir=gs://BUCKET_NAME/output/"
          ]
        }
      }
    ]
  },
  "trainingTaskMetadata": {
    "backingCustomJob": "projects/PROJECT_ID/locations/LOCATION_ID/customJobs/CUSTOM_JOB_ID"
  },
  "modelToUpload": {
    "displayName": "MODEL_NAME",
    "predictSchemata": {},
    "containerSpec": {
      "imageUri": "us-docker.pkg.dev/vertex-ai/prediction/tf2-cpu.2-1:latest"
    }
  },
  "state": "PIPELINE_STATE_PENDING",
  "createTime": "2020-09-15T19:09:54.342080Z",
  "startTime": "2020-09-15T19:13:42.991045Z",
}

Java

在尝试此示例之前，请按照《Vertex AI 快速入门：使用客户端库》中的 Java 设置说明执行操作。如需了解详情，请参阅 Vertex AI Java API 参考文档。

如需向 Vertex AI 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置身份验证。

import com.google.cloud.aiplatform.v1.LocationName;
import com.google.cloud.aiplatform.v1.Model;
import com.google.cloud.aiplatform.v1.ModelContainerSpec;
import com.google.cloud.aiplatform.v1.PipelineServiceClient;
import com.google.cloud.aiplatform.v1.PipelineServiceSettings;
import com.google.cloud.aiplatform.v1.TrainingPipeline;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.google.protobuf.Value;
import com.google.protobuf.util.JsonFormat;
import java.io.IOException;

public class CreateTrainingPipelineCustomJobSample {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String project = "PROJECT";
    String displayName = "DISPLAY_NAME";
    String modelDisplayName = "MODEL_DISPLAY_NAME";
    String containerImageUri = "CONTAINER_IMAGE_URI";
    String baseOutputDirectoryPrefix = "BASE_OUTPUT_DIRECTORY_PREFIX";
    createTrainingPipelineCustomJobSample(
        project, displayName, modelDisplayName, containerImageUri, baseOutputDirectoryPrefix);
  }

  static void createTrainingPipelineCustomJobSample(
      String project,
      String displayName,
      String modelDisplayName,
      String containerImageUri,
      String baseOutputDirectoryPrefix)
      throws IOException {
    PipelineServiceSettings settings =
        PipelineServiceSettings.newBuilder()
            .setEndpoint("us-central1-aiplatform.googleapis.com:443")
            .build();
    String location = "us-central1";

    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests. After completing all of your requests, call
    // the "close" method on the client to safely clean up any remaining background resources.
    try (PipelineServiceClient client = PipelineServiceClient.create(settings)) {
      JsonObject jsonMachineSpec = new JsonObject();
      jsonMachineSpec.addProperty("machineType", "n1-standard-4");

      // A working docker image can be found at
      // gs://cloud-samples-data/ai-platform/mnist_tfrecord/custom_job
      // This sample image accepts a set of arguments including model_dir.
      JsonObject jsonContainerSpec = new JsonObject();
      jsonContainerSpec.addProperty("imageUri", containerImageUri);
      JsonArray jsonArgs = new JsonArray();
      jsonArgs.add("--model_dir=$(AIP_MODEL_DIR)");
      jsonContainerSpec.add("args", jsonArgs);

      JsonObject jsonJsonWorkerPoolSpec0 = new JsonObject();
      jsonJsonWorkerPoolSpec0.addProperty("replicaCount", 1);
      jsonJsonWorkerPoolSpec0.add("machineSpec", jsonMachineSpec);
      jsonJsonWorkerPoolSpec0.add("containerSpec", jsonContainerSpec);

      JsonArray jsonWorkerPoolSpecs = new JsonArray();
      jsonWorkerPoolSpecs.add(jsonJsonWorkerPoolSpec0);

      JsonObject jsonBaseOutputDirectory = new JsonObject();
      // The GCS location for outputs must be accessible by the project's AI Platform
      // service account.
      jsonBaseOutputDirectory.addProperty("output_uri_prefix", baseOutputDirectoryPrefix);

      JsonObject jsonTrainingTaskInputs = new JsonObject();
      jsonTrainingTaskInputs.add("workerPoolSpecs", jsonWorkerPoolSpecs);
      jsonTrainingTaskInputs.add("baseOutputDirectory", jsonBaseOutputDirectory);

      Value.Builder trainingTaskInputsBuilder = Value.newBuilder();
      JsonFormat.parser().merge(jsonTrainingTaskInputs.toString(), trainingTaskInputsBuilder);
      Value trainingTaskInputs = trainingTaskInputsBuilder.build();
      String trainingTaskDefinition =
          "gs://google-cloud-aiplatform/schema/trainingjob/definition/custom_task_1.0.0.yaml";
      String imageUri = "gcr.io/cloud-aiplatform/prediction/tf-cpu.1-15:latest";
      ModelContainerSpec containerSpec =
          ModelContainerSpec.newBuilder().setImageUri(imageUri).build();
      Model modelToUpload =
          Model.newBuilder()
              .setDisplayName(modelDisplayName)
              .setContainerSpec(containerSpec)
              .build();
      TrainingPipeline trainingPipeline =
          TrainingPipeline.newBuilder()
              .setDisplayName(displayName)
              .setTrainingTaskDefinition(trainingTaskDefinition)
              .setTrainingTaskInputs(trainingTaskInputs)
              .setModelToUpload(modelToUpload)
              .build();
      LocationName parent = LocationName.of(project, location);
      TrainingPipeline response = client.createTrainingPipeline(parent, trainingPipeline);
      System.out.format("response: %s\n", response);
      System.out.format("Name: %s\n", response.getName());
    }
  }
}

Python

如需了解如何安装或更新 Python，请参阅安装 Python 版 Vertex AI SDK。如需了解详情，请参阅 Python API 参考文档。

以下示例展示了如何使用 Python 版 Vertex AI SDK 创建自定义训练流水线。选择要使用自定义容器还是预构建容器进行训练：

预构建容器

使用 Python 版 Vertex AI SDK 创建在预构建容器中运行 Python 代码的训练流水线时，可以通过以下任一方式提供训练代码：

指定 Cloud Storage 中 Python 源分发软件包的 URI。

（如果不使用 Python 版 Vertex AI SDK 创建训练流水线时，此选项也可用。）
指定本地机器上 Python 脚本的路径。在创建训练流水线之前，Python 版 Vertex AI SDK 会将脚本打包为源分发并上传到您选择的 Cloud Storage 存储桶。

（只有使用 Python 版 Vertex AI SDK 时此选项才可用。）

要查看每个选项的代码示例，请选择相应的标签页：

套餐

以下示例使用 CustomPythonPackageTrainingJob 类。

def create_training_pipeline_custom_package_job_sample(
    project: str,
    location: str,
    staging_bucket: str,
    display_name: str,
    python_package_gcs_uri: str,
    python_module_name: str,
    container_uri: str,
    model_serving_container_image_uri: str,
    dataset_id: Optional[str] = None,
    model_display_name: Optional[str] = None,
    args: Optional[List[Union[str, float, int]]] = None,
    replica_count: int = 1,
    machine_type: str = "n1-standard-4",
    accelerator_type: str = "ACCELERATOR_TYPE_UNSPECIFIED",
    accelerator_count: int = 0,
    training_fraction_split: float = 0.8,
    validation_fraction_split: float = 0.1,
    test_fraction_split: float = 0.1,
    sync: bool = True,
    tensorboard_resource_name: Optional[str] = None,
    service_account: Optional[str] = None,
):
    aiplatform.init(project=project, location=location, staging_bucket=staging_bucket)

    job = aiplatform.CustomPythonPackageTrainingJob(
        display_name=display_name,
        python_package_gcs_uri=python_package_gcs_uri,
        python_module_name=python_module_name,
        container_uri=container_uri,
        model_serving_container_image_uri=model_serving_container_image_uri,
    )

    # This example uses an ImageDataset, but you can use another type
    dataset = aiplatform.ImageDataset(dataset_id) if dataset_id else None

    model = job.run(
        dataset=dataset,
        model_display_name=model_display_name,
        args=args,
        replica_count=replica_count,
        machine_type=machine_type,
        accelerator_type=accelerator_type,
        accelerator_count=accelerator_count,
        training_fraction_split=training_fraction_split,
        validation_fraction_split=validation_fraction_split,
        test_fraction_split=test_fraction_split,
        sync=sync,
        tensorboard=tensorboard_resource_name,
        service_account=service_account,
    )

    model.wait()

    print(model.display_name)
    print(model.resource_name)
    print(model.uri)
    return model

Script

以下示例使用 CustomTrainingJob 类。

def create_training_pipeline_custom_job_sample(
    project: str,
    location: str,
    staging_bucket: str,
    display_name: str,
    script_path: str,
    container_uri: str,
    model_serving_container_image_uri: str,
    dataset_id: Optional[str] = None,
    model_display_name: Optional[str] = None,
    args: Optional[List[Union[str, float, int]]] = None,
    replica_count: int = 0,
    machine_type: str = "n1-standard-4",
    accelerator_type: str = "ACCELERATOR_TYPE_UNSPECIFIED",
    accelerator_count: int = 0,
    training_fraction_split: float = 0.8,
    validation_fraction_split: float = 0.1,
    test_fraction_split: float = 0.1,
    sync: bool = True,
    tensorboard_resource_name: Optional[str] = None,
    service_account: Optional[str] = None,
):
    aiplatform.init(project=project, location=location, staging_bucket=staging_bucket)

    job = aiplatform.CustomTrainingJob(
        display_name=display_name,
        script_path=script_path,
        container_uri=container_uri,
        model_serving_container_image_uri=model_serving_container_image_uri,
    )

    # This example uses an ImageDataset, but you can use another type
    dataset = aiplatform.ImageDataset(dataset_id) if dataset_id else None

    model = job.run(
        dataset=dataset,
        model_display_name=model_display_name,
        args=args,
        replica_count=replica_count,
        machine_type=machine_type,
        accelerator_type=accelerator_type,
        accelerator_count=accelerator_count,
        training_fraction_split=training_fraction_split,
        validation_fraction_split=validation_fraction_split,
        test_fraction_split=test_fraction_split,
        sync=sync,
        tensorboard=tensorboard_resource_name,
        service_account=service_account,
    )

    model.wait()

    print(model.display_name)
    print(model.resource_name)
    print(model.uri)
    return model

自定义容器

以下示例使用 CustomContainerTrainingJob 类。

def create_training_pipeline_custom_container_job_sample(
    project: str,
    location: str,
    staging_bucket: str,
    display_name: str,
    container_uri: str,
    model_serving_container_image_uri: str,
    dataset_id: Optional[str] = None,
    model_display_name: Optional[str] = None,
    args: Optional[List[Union[str, float, int]]] = None,
    replica_count: int = 1,
    machine_type: str = "n1-standard-4",
    accelerator_type: str = "ACCELERATOR_TYPE_UNSPECIFIED",
    accelerator_count: int = 0,
    training_fraction_split: float = 0.8,
    validation_fraction_split: float = 0.1,
    test_fraction_split: float = 0.1,
    sync: bool = True,
    tensorboard_resource_name: Optional[str] = None,
    service_account: Optional[str] = None,
):
    aiplatform.init(project=project, location=location, staging_bucket=staging_bucket)

    job = aiplatform.CustomContainerTrainingJob(
        display_name=display_name,
        container_uri=container_uri,
        model_serving_container_image_uri=model_serving_container_image_uri,
    )

    # This example uses an ImageDataset, but you can use another type
    dataset = aiplatform.ImageDataset(dataset_id) if dataset_id else None

    model = job.run(
        dataset=dataset,
        model_display_name=model_display_name,
        args=args,
        replica_count=replica_count,
        machine_type=machine_type,
        accelerator_type=accelerator_type,
        accelerator_count=accelerator_count,
        training_fraction_split=training_fraction_split,
        validation_fraction_split=validation_fraction_split,
        test_fraction_split=test_fraction_split,
        sync=sync,
        tensorboard=tensorboard_resource_name,
        service_account=service_account,
    )

    model.wait()

    print(model.display_name)
    print(model.resource_name)
    print(model.uri)
    return model

超参数调节作业和模型上传

此训练流水线封装了一个超参数调节作业以及额外的便捷步骤，可以更轻松地在训练后将模型部署到 Vertex AI。此训练流水线执行两个主要任务：

训练流水线会创建一个超参数调节作业资源。超参数调节作业会创建多个试验。每次试验时，自定义作业都会使用您指定的计算资源和超参数运行训练应用。
超参数调节作业完成后，训练流水线会在您为 Cloud Storage 存储桶指定的输出目录 (baseOutputDirectory) 中查找最佳试验中的模型工件。训练流水线使用这些工件来创建模型资源，帮助您为模型部署做好准备。

对于此训练流水线，您必须指定 baseOutputDirectory，Vertex AI 会在其中搜索最佳试验中的模型工件。

超参数调节作业需要配置额外的设置。详细了解 HyperparameterTuningJob 的设置。

REST

通过以下代码示例使用 trainingPipeline 资源的 create 方法创建训练流水线。

在使用任何请求数据之前，请先进行以下替换：

LOCATION_ID：您的项目的区域。
PROJECT_ID：您的项目 ID。
TRAINING_PIPELINE_NAME：必填。trainingPipeline 的显示名。
如果训练应用使用 Vertex AI 数据集，请指定以下内容：
- DATASET_ID：数据集的 ID。
- ANNOTATIONS_FILTER：根据您指定的注释过滤数据集。
- ANNOTATION_SCHEMA_URI：根据指定的注释架构 URI 过滤数据集。
- 使用以下任一选项来指定数据项拆分为训练集、验证集和测试集的方式。
  - 如需根据定义每个集合大小的比例来拆分数据集，请指定以下内容：
    - TRAINING_FRACTION：用于训练模型的数据集比例。
    - VALIDATION_FRACTION：用于验证模型的数据集比例。
    - TEST_FRACTION：用于评估模型的数据集比例。
  - 如需根据过滤条件拆分数据集，请指定以下内容：
    - TRAINING_FILTER：将数据集过滤为用于训练模型的数据项。
    - VALIDATION_FILTER：将数据集过滤为用于验证模型的数据项。
    - TEST_FILTER：将数据集过滤为用于评估模型的数据项。
  - 如需使用预定义的拆分，请指定以下内容：
    - PREDEFINED_SPLIT_KEY：用于拆分数据集的列的名称。此列中可接受的值包括“training”“validation”和“test”。
  - 如需根据数据项的时间戳拆分数据集，请指定以下内容：
    - TIMESTAMP_TRAINING_FRACTION：用于训练模型的数据集比例。
    - TIMESTAMP_VALIDATION_FRACTION：用于验证模型的数据集比例。
    - TIMESTAMP_TEST_FRACTION：用于评估模型的数据集比例。
    - TIMESTAMP_SPLIT_KEY：用于拆分数据集的时间戳列的名称。
- OUTPUT_URI_PREFIX：Vertex AI 在训练数据集被拆分为训练集、验证集和测试集后将其导出的目标 Cloud Storage 位置。
指定超参数调节作业：
- 指定指标：
  - METRIC_ID：此指标的名称。
  - METRIC_GOAL：此指标的目标。可以是 MAXIMIZE 或 MINIMIZE。
- 指定超参数：
  - PARAMETER_ID：此超参数的名称。
  - PARAMETER_SCALE：（可选）应如何扩缩参数。不对 CATEGORICAL 参数进行设置。可以是 UNIT_LINEAR_SCALE、UNIT_LOG_SCALE、UNIT_REVERSE_LOG_SCALE 或 SCALE_TYPE_UNSPECIFIED。
  - 如果此超参数的类型是 DOUBLE，请指定此超参数的最小值 (DOUBLE_MIN_VALUE) 和最大值 (DOUBLE_MAX_VALUE)。
  - 如果此超参数的类型是 INTEGER，请指定此超参数的最小值 (INTEGER_MIN_VALUE) 和最大值 (INTEGER_MAX_VALUE)。
  - 如果此超参数的类型是 CATEGORICAL，请以字符串数组的形式指定可接受的值 (CATEGORICAL_VALUES)。
  - 如果此超参数的类型是 DISCRETE，请以数字数组的形式指定可接受的值 (DISCRETE_VALUES)。
- ALGORITHM：（可选）在此超参数调节作业中使用的搜索算法。可以是 ALGORITHM_UNSPECIFIED、GRID_SEARCH 或 RANDOM_SEARCH。
- MAX_TRIAL_COUNT：此作业中运行的试验数上限。
- PARALLEL_TRIAL_COUNT：可以并行运行的试验数上限。
- MAX_FAILED_TRIAL_COUNT：超参数调节作业失败前允许失败的作业数量。
- 定义试验自定义训练作业：
  - MACHINE_TYPE：机器类型。请参阅可用于训练的机器类型。
  - ACCELERATOR_TYPE：（可选）要附加到每个试验的加速器类型。
  - ACCELERATOR_COUNT：（可选）要附加到每个试验的加速器数量。
  - REPLICA_COUNT：每次试验要使用的工作器副本数。
  - 如果训练应用在自定义容器中运行，请指定以下内容：
    - CUSTOM_CONTAINER_IMAGE_URI：要在每个工作器副本上运行的容器映像的 Artifact Registry、Container Registry 或 Docker Hub URI。
    - CUSTOM_CONTAINER_COMMAND：（可选）启动容器时要调用的命令。此命令会替换容器的默认入口点。
    - CUSTOM_CONTAINER_ARGS：（可选）启动容器时要传递的参数。
  - 如果训练应用是在预构建容器中运行的 Python 软件包，请指定以下内容：
    - PYTHON_PACKAGE_EXECUTOR_IMAGE_URI：运行提供的 Python 软件包的容器映像的 URI。请参阅可用于训练的预构建容器。
    - PYTHON_PACKAGE_URIS：Python 软件包文件的 Cloud Storage 位置，这些文件是训练程序及其从属软件包。软件包 URI 的数量上限为 100。
    - PYTHON_MODULE：安装软件包后要运行的 Python 模块名称。
    - PYTHON_PACKAGE_ARGS：（可选）要传递给 Python 模块的命令行参数。
- 了解作业调度选项。
- TIMEOUT：（可选）每次试验的最长运行时间。
- 指定要应用于此超参数调节作业的任何标签的 LABEL_NAME 和 LABEL_VALUE。
MODEL_NAME：TrainingPipeline 上传（创建）的模型的显示名。
MODEL_DESCRIPTION：可选。模型说明。
PREDICTION_IMAGE_URI：必填。指定以下两个选项之一：
- 用于预测的预构建容器的映像 URI，例如“tf2-cpu.2-1:latest”。
- 您自己的用于预测的自定义容器的映像 URI。
modelToUpload.labels：可选。用于组织模型的任何键值对。例如：
- "env": "prod"
- "tier": "backend"
指定要应用于此训练流水线的任何标签的 LABEL_NAME 和 LABEL_VALUE。

HTTP 方法和网址：

POST https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/trainingPipelines

请求 JSON 正文：

{
  "displayName": "TRAINING_PIPELINE_NAME",
  "inputDataConfig": {
    "datasetId": DATASET_ID,
    "annotationsFilter": ANNOTATIONS_FILTER,
    "annotationSchemaUri": ANNOTATION_SCHEMA_URI,

    // Union field split can be only one of the following:
    "fractionSplit": {
      "trainingFraction": TRAINING_FRACTION,
      "validationFraction": VALIDATION_FRACTION,
      "testFraction": TEST_FRACTION
    },
    "filterSplit": {
      "trainingFilter": TRAINING_FILTER,
      "validationFilter": VALIDATION_FILTER,
      "testFilter": TEST_FILTER
    },
    "predefinedSplit": {
      "key": PREDEFINED_SPLIT_KEY
    },
    "timestampSplit": {
      "trainingFraction": TIMESTAMP_TRAINING_FRACTION,
      "validationFraction": TIMESTAMP_VALIDATION_FRACTION,
      "testFraction": TIMESTAMP_TEST_FRACTION,
      "key": TIMESTAMP_SPLIT_KEY
    }
    // End of list of possible types for union field split.
    "gcsDestination": {
      "outputUriPrefix": OUTPUT_URI_PREFIX
    }
  },
  "trainingTaskDefinition": "gs://google-cloud-aiplatform/schema/trainingjob/definition/hyperparameter_tuning_task_1.0.0.yaml",
  "trainingTaskInputs": {
    "studySpec": {
    "metrics": [
      {
        "metricId": METRIC_ID,
        "goal": METRIC_GOAL
      }
    ],
    "parameters": [
      {
        "parameterId": PARAMETER_ID,
        "scaleType": PARAMETER_SCALE,

        // Union field parameter_value_spec can be only one of the following:
        "doubleValueSpec": {
            "minValue": DOUBLE_MIN_VALUE,
            "maxValue": DOUBLE_MAX_VALUE
        },
        "integerValueSpec": {
            "minValue": INTEGER_MIN_VALUE,
            "maxValue": INTEGER_MAX_VALUE
        },
        "categoricalValueSpec": {
            "values": [
              CATEGORICAL_VALUES
            ]
        },
        "discreteValueSpec": {
            "values": [
              DISCRETE_VALUES
            ]
        }
        // End of list of possible types for union field parameter_value_spec.
      }
    ],
    "ALGORITHM": ALGORITHM
  },
  "maxTrialCount": MAX_TRIAL_COUNT,
  "parallelTrialCount": PARALLEL_TRIAL_COUNT,
  "maxFailedTrialCount": MAX_FAILED_TRIAL_COUNT,
  "trialJobSpec": {
      "workerPoolSpecs": [
        {
          "machineSpec": {
            "machineType": MACHINE_TYPE,
            "acceleratorType": ACCELERATOR_TYPE,
            "acceleratorCount": ACCELERATOR_COUNT
          },
          "replicaCount": REPLICA_COUNT,

          // Union field task can be only one of the following:
          "containerSpec": {
            "imageUri": CUSTOM_CONTAINER_IMAGE_URI,
            "command": [
              CUSTOM_CONTAINER_COMMAND
            ],
            "args": [
              CUSTOM_CONTAINER_ARGS
            ]
          },
          "pythonPackageSpec": {
            "executorImageUri": PYTHON_PACKAGE_EXECUTOR_IMAGE_URI,
            "packageUris": [
              PYTHON_PACKAGE_URIS
            ],
            "pythonModule": PYTHON_MODULE,
            "args": [
              PYTHON_PACKAGE_ARGS
            ]
          }
          // End of list of possible types for union field task.
        }
      ],
      "scheduling": {
        "TIMEOUT": TIMEOUT
      }
    },
    "labels": {
      LABEL_NAME_1": LABEL_VALUE_1,
      LABEL_NAME_2": LABEL_VALUE_2
    }
  },
  "modelToUpload": {
    "displayName": "MODEL_NAME",
    "description": "MODEL_DESCRIPTION",
    "predictSchemata": {},
    "containerSpec": {
      "imageUri": "PREDICTION_IMAGE_URI"
    }
  },
  "labels": {
    LABEL_NAME_1": LABEL_VALUE_1,
    LABEL_NAME_2": LABEL_VALUE_2
  }
}

如需发送请求，请选择以下方式之一：

curl

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/trainingPipelines"

PowerShell

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/trainingPipelines" | Select-Object -Expand Content

响应包含有关规范的信息以及 TRAININGPIPELINE_ID。

响应


{
  "name": "projects/PROJECT_ID/locations/LOCATION_ID/trainingPipelines/TRAININGPIPELINE_ID",
  "displayName": "TRAINING_PIPELINE_NAME",
  "input_data_config" {
    "dataset_id": "1234567891011121314"
    "gcs_destination" {
      "output_uri_prefix": "gs://BUCKET_NAME/data/20200915191342"
    }
    "annotation_schema_uri": "gs://google-cloud-aiplatform/schema/dataset/annotation/image_classification_1.0.0.yaml"
  },
  "trainingTaskDefinition": "gs://google-cloud-aiplatform/schema/trainingjob/definition/custom_task_1.0.0.yaml",
  "trainingTaskInputs": {
    "name": "projects/12345/locations/us-central1/hyperparameterTuningJobs/6789",
  "displayName": "myHyperparameterTuningJob",
  "studySpec": {
    "metrics": [
      {
        "metricId": "myMetric",
        "goal": "MINIMIZE"
      }
    ],
    "parameters": [
      {
        "parameterId": "myParameter1",
        "integerValueSpec": {
          "minValue": "1",
          "maxValue": "128"
        },
        "scaleType": "UNIT_LINEAR_SCALE"
      },
      {
        "parameterId": "myParameter2",
        "doubleValueSpec": {
          "minValue": 1e-07,
          "maxValue": 1
        },
        "scaleType": "UNIT_LINEAR_SCALE"
      }
    ],
    "ALGORITHM": "RANDOM_SEARCH"
  },
  "maxTrialCount": 20,
  "parallelTrialCount": 1,
  "trialJobSpec": {
    "workerPoolSpecs": [
      {
        "machineSpec": {
          "machineType": "n1-standard-4"
        },
        "replicaCount": "1",
        "pythonPackageSpec": {
          "executorImageUri": "us-docker.pkg.dev/vertex-ai/training/training-tf-cpu.2-1:latest",
          "packageUris": [
            "gs://my-bucket/my-training-application/trainer.tar.bz2"
          ],
          "pythonModule": "my-trainer.trainer"
        }
      }
    ]
  },
  "state": "PIPELINE_STATE_PENDING",
  "createTime": "2020-09-15T19:09:54.342080Z",
  "startTime": "2020-09-15T19:13:42.991045Z",
}

监控训练

如需查看训练日志，请执行以下操作：

在 Google Cloud 控制台的 Vertex AI 部分中，前往训练页面。

前往“训练”页面
点击作业名称以转到自定义作业页面。
点击查看日志。

您还可以使用交互式 shell 在训练作业运行时检查训练容器。

查看经过训练的模型

自定义训练流水线完成后，您可以在 Google Cloud 控制台的 Vertex AI 部分的模型页面上找到经过训练的模型。

转到“模型”页面

后续步骤

了解如何使用 TensorBoard Profiler 找出训练性能瓶颈，以更快、更低的费用训练模型。
将模型部署到端点。
创建超参数调节作业。

创建训练流水线

创建流水线之前

训练流水线选项

CustomJob 包含的内容

配置流水线以使用代管式数据集

配置分布式训练

自定义作业和模型上传

控制台

REST

curl

PowerShell

响应

Java

Python

预构建容器

套餐

Script

自定义容器

超参数调节作业和模型上传

REST

curl

PowerShell

响应

监控训练

查看经过训练的模型

后续步骤

`CustomJob` 包含的内容