创建训练流水线

本页面介绍了如何创建训练流水线以在 Vertex AI 上运行自定义训练应用。

创建流水线之前

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

训练流水线选项

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

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

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

自定义作业包含的内容

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

在工作器池中,您可以指定以下设置:

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

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

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

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

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

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

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

配置分布式训练

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

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

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

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

自定义作业和模型上传

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

  1. 训练流水线会创建一个自定义作业资源。自定义作业使用您指定的计算资源运行训练应用。

  2. 自定义作业完成后,训练流水线会在您为 Cloud Storage 存储分区指定的输出目录中查找训练应用创建的模型工件。训练流水线使用这些工件来创建模型资源,帮助您为模型部署做好准备。

    模型工件的文件总大小不能超过 10 GB。

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

如果您同时指定 baseOutputDirectorymodelToUpload.artifactUri,则 Vertex AI 会使用 modelToUpload.artifactUri

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

控制台

  1. 在 Cloud Console 的 Vertex AI 部分中,转到训练流水线页面。

    转到“训练流水线”

  2. 点击 创建以打开训练新模型窗格。

  3. 选择训练方法步骤中,指定以下设置:

    1. 如果您要使用代管式数据集进行训练,请指定数据集注释集

      否则,在数据集下拉列表中,选择无代管式数据集

    2. 选择自定义训练(高级)

    点击继续

  4. 定义模型步骤中,输入您为模型选择的名称 MODEL_NAME。点击 Continue

  5. 训练容器步骤中,指定以下设置:

    1. 选择要使用预构建容器还是自定义容器进行训练。

    2. 根据您的选择,执行以下相应操作:

    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 开始运行训练代码时要使用的参数。这些参数的行为因您使用的容器类型而异:

      • 如果您使用的是预构建容器,Vertex AI 会将参数作为命令行标志传递给 Python 模块

      • 如果您使用的是自定义容器,则 Vertex AI 会使用参数替换容器的 CMD 指令

    点击继续

  6. 超参数调节步骤中,确保未选中启用超参数调节复选框。点击继续

  7. 计算和价格步骤中,指定以下设置:

    1. 区域下拉列表中,选择支持自定义训练的区域

    2. 工作器池 0 部分中,指定用于训练的计算资源

      如果指定加速器,请确保所选加速器类型在所选区域可用

      如果您要执行分布式训练,则点击 Add more worker pools,并为您需要的每个额外工作器池指定一组额外的计算资源。

    点击继续

  8. 预测容器步骤中,指定以下设置:

    1. 选择要使用预构建容器还是自定义容器以在经过训练的模型上执行预测。

    2. 根据您的选择,执行以下相应操作:

    3. 模型目录 (Model directory) 字段包含您之前在训练容器步骤的 Model output directory 字段中设置的值。更改这两个字段具有相同的效果。如需详细了解此字段,请参阅之前的说明

    4. 预测架构部分中的字段留空。

  9. 点击开始训练以启动自定义训练流水线。

REST 和命令行

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

在使用下面的请求数据之前,请先进行以下替换:

  • 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:(可选)启动容器时要传递的参数。
    • 如果训练应用是在预构建容器中运行的 Python 软件包,请指定以下内容:
      • PYTHON_PACKAGE_EXECUTOR_IMAGE_URI:运行提供的 Python 软件包的容器映像的 URI。请参阅可用于训练的预构建容器
      • PYTHON_PACKAGE_URIS:Python 软件包文件的 Cloud Storage 位置,这些文件是训练程序及其从属软件包。软件包 URI 的数量上限为 100。
      • PYTHON_MODULE:安装软件包后要运行的 Python 模块名称。
      • PYTHON_PACKAGE_ARGS:(可选)要传递给 Python 模块的命令行参数。
    • 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_NAMELABEL_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

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

curl -X POST \
-H "Authorization: Bearer "$(gcloud auth application-default 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 application-default 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

Java

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

      JsonArray jsonArgs = new JsonArray();
      jsonArgs.add("--model_dir=$(AIP_MODEL_DIR)");

      // A working docker image can be found at
      // gs://cloud-samples-data/ai-platform/mnist_tfrecord/custom_job
      JsonObject jsonContainerSpec = new JsonObject();
      jsonContainerSpec.addProperty("imageUri", containerImageUri);
      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 版 Vertex SDK 创建自定义训练流水线。选择要使用自定义容器还是预建容器进行训练:

预构建容器

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

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

    (如果不使用 Python 版 Vertex SDK 创建训练流水线时,此选项也可用。)

  • 指定本地机器上 Python 脚本的路径。在创建训练流水线之前,Python 版 Vertex SDK 会将脚本打包为源分发并上传到您选择的 Cloud Storage 存储分区。

    (只有使用 Python 版 Vertex 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,
):
    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,
    )

    model.wait()

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

脚本

以下示例使用 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,
):
    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,
    )

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

    model.wait()

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

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

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

  1. 训练流水线会创建一个超参数调节作业资源。超参数调节作业会创建多个试验。每次试验时,自定义作业都会使用您指定的计算资源和超参数运行训练应用。

  2. 超参数调节作业完成后,训练流水线会在您为 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:此指标的目标。可以是 MAXIMIZEMINIMIZE
    • 指定超参数:
      • PARAMETER_ID:此超参数的名称。
      • PARAMETER_SCALE:(可选)应如何扩缩参数。不对 CATEGORICAL 参数进行设置。可以是 UNIT_LINEAR_SCALEUNIT_LOG_SCALEUNIT_REVERSE_LOG_SCALESCALE_TYPE_UNSPECIFIED
      • 如果此超参数的类型是 DOUBLE,请指定此超参数的最小值 (DOUBLE_MIN_VALUE) 和最大值 (DOUBLE_MAX_VALUE)。
      • 如果此超参数的类型是 INTEGER,请指定此超参数的最小值 (INTEGER_MIN_VALUE) 和最大值 (INTEGER_MAX_VALUE)。
      • 如果此超参数的类型是 CATEGORICAL,请以字符串数组的形式指定可接受的值 (CATEGORICAL_VALUES)。
      • 如果此超参数的类型是 DISCRETE,请以数字数组的形式指定可接受的值 (DISCRETE_VALUES)。
    • ALGORITHM:(可选)在此超参数调节作业中使用的搜索算法。可以是 ALGORITHM_UNSPECIFIEDGRID_SEARCHRANDOM_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_NAMELABEL_VALUE
  • MODEL_NAME:TrainingPipeline 上传(创建)的模型的显示名。
  • MODEL_DESCRIPTION:可选。模型说明。
  • PREDICTION_IMAGE_URI:必填。指定以下两个选项之一:
  • modelToUpload.labels:可选。用于组织模型的任何键值对。例如:
    • "env": "prod"
    • "tier": "backend"
  • 指定要应用于此训练流水线的任何标签的 LABEL_NAMELABEL_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 application-default 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 application-default 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

监控训练

如需查看训练日志,请执行以下操作:

  1. 在 Cloud Console 的 Vertex AI 部分中,转到训练页面。

    转到“训练”页面

  2. 点击作业名称以转到自定义作业页面。

  3. 点击查看日志

查看经过训练的模型

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

转到“模型”页面

后续步骤