运行流水线

借助 Agent Platform Pipelines，您可以运行使用 Kubeflow Pipelines SDK 或 TensorFlow Extended 以无服务器方式构建的机器学习 (ML) 流水线。本文档介绍如何运行机器学习流水线。

您还可以使用模板库中预构建的模板创建流水线运行。如需详细了解模板库，请参阅使用模板库中的预构建模板。

您可以使用 Gemini Enterprise Agent Platform Experiments 通过将流水线运行与实验或实验运行相关联来跟踪、分析和比较流水线运行。通过比较流水线运行的参数、输出和效果指标，您可以确定最适合您的应用场景的流水线配置。如需了解详情，请参阅 Gemini Enterprise Agent Platform 实验简介。

准备工作

在使用 Agent Platform Pipelines 运行流水线之前，请按照以下说明来设置Google Cloud 项目和开发环境：

1. 构建流水线。

2. 如需使用 Agent Platform SDK for Python 运行流水线，请安装 Vertex SDK。

   • 安装 Agent Platform SDK。

创建流水线运行

按照以下说明，使用 Google Cloud 控制台或 Python 运行机器学习流水线。

控制台

按照以下说明，使用 Google Cloud 控制台运行机器学习流水线。

1. 在 Google Cloud 控制台的智能体平台部分中，前往流水线页面。

   打开“流水线”

2. 在区域下拉列表中，选择区域以创建流水线运行。

3. 点击 创建运行以打开创建流水线运行窗格。

4. 在运行详情部分中，执行以下操作：

   1. 点击运行来源。您可以使用以下选项：

      • 从现有流水线中选择：如需根据现有流水线模板创建流水线运行，请点击从现有流水线中选择，然后输入以下详细信息：

        1. 选择包含流水线或组件定义文件的仓库。

        2. 选择流水线或组件和版本。

        3. 指定运行名称以唯一标识流水线运行。

      • 选择模板库流水线：如需根据模板库中 Google 编写的流水线模板创建流水线运行，请点击选择模板库流水线，然后输入以下详细信息：

        1. 在模板库流水线列表中，选择流水线模板。

        2. 可选：修改唯一标识流水线运行的默认运行名称。

        注意：以下说明介绍了如何使用创建流水线运行页面的默认界面创建流水线运行，该界面包含运行详情和运行时配置部分。对于模板库中的部分模板，此页面包含额外的部分。例如，表格分类/回归 AutoML 模板还包括训练方法、训练选项以及计算和价格部分。

      • 上传文件：如需上传已编译的流水线定义，请点击上传文件，然后输入以下详细信息：

        1. 点击浏览，打开文件选择器。导航到要运行的已编译流水线 YAML 文件，选择流水线，然后点击打开。

        2. 默认情况下，流水线或组件名称显示流水线定义中指定的名称。您也可以指定其他流水线名称。

        3. 指定运行名称以唯一标识流水线运行。

      • 从 Cloud Storage 导入：如需从 Cloud Storage 导入流水线定义文件，请点击从 Cloud Storage 导入，然后输入以下详细信息：

        1. 点击浏览，导航到包含流水线定义对象的 Cloud Storage 存储桶，选择文件，然后点击选择。

           或者，输入流水线定义文件的 Cloud Storage URI 或 HTTP/HTTPS 网址。

        2. 指定流水线或组件名称。

        3. 指定运行名称以唯一标识流水线运行。

  1. Optional: To schedule recurring pipeline runs, specify the **Run schedule**, as follows:

      1. Select **Recurring**.

      1. Under **Start time**, specify when the schedule becomes active.

          * To schedule the first run to occur immediately after schedule creation, select **Immediately**.

          * To schedule the first run to occur at a specific time and date, select **On**.

      1. In the **Frequency** field, specify the frequency to schedule and execute the
      pipeline runs, using a cron schedule expression based on
      [unix-cron](https://man7.org/linux/man-pages/man5/crontab.5.html).

      1. Under **Ends**, specify when the schedule ends.

          * To indicate that the schedule creates pipeline runs indefinitely, select **Never**.

          * To indicate that the schedule ends on a specific date and time, select **On**, and specify the end date and time for the schedule.

      1. Optional: To specify that the pipeline run uses a custom service account,
      Customer-managed encryption keys (CMEK), or a peered VPC network, click
      **Advanced options**, and then follow these instructions:

          *  To specify a service account, select a service account from the
            **Service account** drop-down list.

              If you don't specify a service account,
              Agent Platform Pipelines runs your pipeline using the default
              Compute Engine service account.

              Learn more about [configuring a service account for use with
              Agent Platform Pipelines](/gemini-enterprise-agent-platform/machine-learning/pipelines/configure-project#service-account).

          *  To use a CMEK, select **Use a customer-managed encryption key**. The **Select a customer-managed key** drop-down list appears. In the **Select a customer-managed key** drop-down list, select the key that you want to use.

          *  To use a peered VPC network in this pipeline run, enter the VPC
          network name in the **Peered VPC network** box.

  1. Click **Continue**.

1. 在运行时配置部分中，配置流水线运行，如下所示：

   1. 在 Cloud Storage 位置下，点击浏览以选择用于存储流水线输出工件的 Cloud Storage 存储桶，然后点击选择。

   2. 可选：如需为流水线运行配置失败政策和缓存，请点击高级选项，然后按照以下说明操作：

      • 在失败政策下，指定整个流水线的失败政策。详细了解流水线故障政策。

        • 如需将流水线配置为在一个任务失败后继续安排任务，请选择运行所有步骤，直到完成。默认情况下，此选项处于选中状态。

        • 如需将流水线配置为在一个任务失败后失败，请选择只要有一个步骤失败，就让这个运行作业失败。

      • 在缓存配置下，指定整个流水线的缓存配置。

        • 如需将任务级缓存配置用于流水线中的任务，请选择不替换任务级缓存配置。

        • 如需为流水线中的所有任务启用缓存，并替换任何任务级缓存配置，请选择为所有步骤启用“从缓存读取”（最快）。

        • 如需为流水线中的所有任务关闭缓存，并替换任何任务级缓存配置，请选择为所有步骤停用“从缓存读取”（最快）。

   3. 可选：如果您的流水线具有参数，请在流水线参数下指定流水线运行参数。

2. 如需创建流水线运行，请点击提交。

Python

按照以下说明使用 Agent Platform SDK for Python 运行机器学习流水线。在运行以下代码示例之前，必须先设置身份验证。

设置身份验证

如需设置身份验证，您必须创建服务账号密钥，并为服务账号密钥的路径设置环境变量。

1. 创建服务账号：

   1. 在 Google Cloud 控制台中，前往创建服务账号页面。

      转到“创建服务账号”

   2. 在服务账号名称字段中，输入一个名称。
   3. 可选：在服务账号说明字段中，输入说明。
   4. 点击创建。
   5. 点击选择角色字段。在所有角色下，依次选择 Gemini Enterprise Agent Platform > Agent Platform User。
   注意：您选择的角色允许服务账号访问资源。您日后可以使用Google Cloud 控制台查看和更改这些角色。如需了解详情，请参阅 Agent Platform 访问权限控制。

   6. 点击完成以创建服务账号。

      不要关闭浏览器窗口。您将在下一步骤中用到它。

2. 创建用于身份验证的服务账号密钥：

   1. 在 Google Cloud 控制台中，点击您创建的服务账号的邮箱。
   2. 点击密钥。
   3. 依次点击添加密钥和创建新密钥。
   4. 点击创建。JSON 密钥文件将下载到您的计算机上。
   5. 点击关闭。
3. 向新服务账号授予用于运行流水线的服务账号的访问权限。
   1. 点击 返回到服务账号列表。

   2. 点击用于运行流水线的服务账号的名称。此时会显示服务账号详情页面。

      如果您按照指南中的说明为 Gemini Enterprise Agent Platform Pipelines 配置了项目，则此服务账号与您在配置具有精细权限的服务账号部分中创建的服务账号相同。否则，Gemini Enterprise Agent Platform 会使用 Compute Engine 默认服务账号来运行流水线。Compute Engine 默认服务账号的名称如下：PROJECT_NUMBER-compute@

   3. 点击权限标签页。
   4. 点击授予使用权限。此时会显示添加主账号面板。
   5. 在新的主账号框中，输入您在上一步中创建的服务账号的电子邮件地址。
   6. 在角色下拉列表中，选择服务账号 > 服务账号用户。
   7. 点击保存

4. 将环境变量 GOOGLE_APPLICATION_CREDENTIALS 设置为包含服务账号密钥的 JSON 文件的路径。此变量仅适用于当前的 shell 会话，因此，如果您打开新的会话，请重新设置该变量。

   示例：Linux 或 macOS

   将 [PATH] 替换为包含您的服务账号密钥的 JSON 文件的路径。

   export GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

   例如：

   export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

   示例：Windows

   将 [PATH] 替换为包含服务账号密钥的 JSON 文件的路径，将 [FILE_NAME] 替换为文件名。

   使用 PowerShell：

   $env:GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

   例如：

   $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\[FILE_NAME].json"

   使用命令提示符：

   set GOOGLE_APPLICATION_CREDENTIALS=[PATH]

运行流水线

运行 Gemini Enterprise Agent Platform PipelineJob 要求您创建一个 PipelineJob 资源，然后调用 submit 方法。

KFP 支持的特殊输入类型

创建流水线运行时，您还可以传入 KFP SDK 支持的以下占位符作为输入：

• {{$.pipeline_job_name_placeholder}}

• {{$.pipeline_job_resource_name_placeholder}}

• {{$.pipeline_job_id_placeholder}}

• {{$.pipeline_task_name_placeholder}}

• {{$.pipeline_task_id_placeholder}}

• {{$.pipeline_job_create_time_utc_placeholder}}

• {{$.pipeline_root_placeholder}}

如需了解详情，请参阅 Kubeflow Pipelines v2 文档中的特殊输入类型。

from google.cloud import aiplatform
job = aiplatform.PipelineJob(display_name = DISPLAY_NAME,
                             template_path = COMPILED_PIPELINE_PATH,
                             job_id = JOB_ID,
                             pipeline_root = PIPELINE_ROOT_PATH,
                             parameter_values = PIPELINE_PARAMETERS,
                             enable_caching = ENABLE_CACHING,
                             encryption_spec_key_name = CMEK,
                             labels = LABELS,
                             credentials = CREDENTIALS,
                             project = PROJECT_ID,
                             location = LOCATION,
                             failure_policy = FAILURE_POLICY)
job.submit(service_account=SERVICE_ACCOUNT,
           network=NETWORK,
           reserved_ip_ranges=RESERVED_IP_RANGES)

替换以下内容：

• DISPLAY_NAME：流水线的名称，此名称将显示在 Google Cloud 控制台中。

• COMPILED_PIPELINE_PATH：已编译的流水线 YAML 文件的路径。它可以是 Cloud Storage URI、HTTP/HTTPS 网址或本地路径。
  

  可选：如需指定已编译的流水线的特定版本，请添加采用以下任一格式的版本标记：

  • COMPILED_PIPELINE_PATH:TAG，其中 TAG 是版本标记。

  • COMPILED_PIPELINE_PATH@SHA256_TAG，其中 SHA256_TAG 是流水线版本的 sha256 哈希值。

• JOB_ID：（可选）此流水线运行的唯一标识符。如果未指定任务 ID，Agent Platform Pipelines 将用流水线名称和运行流水线的时间戳为您创建任务 ID。

• PIPELINE_ROOT_PATH：（可选）如需替换流水线定义中指定的流水线根路径，请指定流水线作业可以访问的路径，例如 Cloud Storage 存储桶 URI。

• PIPELINE_PARAMETERS：（可选）要传递给此运行的流水线参数。例如，创建一个 dict()，以参数名称为字典键，以参数值为字典值。

• ENABLE_CACHING：（可选）指定此流水线运行是否使用执行缓存。执行缓存通过跳过流水线任务（对于当前这组输入已知输出）来降低费用。如果未指定启用缓存参数，则在此流水线运行中使用执行缓存。 详细了解执行缓存。

• CMEK：（可选）要用于此流水线运行的客户管理的加密密钥的名称。

• LABELS：（可选）用户定义的标签，用于整理此 PipelineJob。 如需详细了解资源标签，请参阅 Resource Manager 文档中的创建和管理标签部分。

  Agent Platform Pipelines 会自动将以下标签关联到您的流水线运行作业：

  vertex-ai-pipelines-run-billing-id: pipeline_run_id

  其中，pipeline_run_id 是流水线运行作业的唯一 ID。

  此标签会在结算报告中关联流水线运行作业生成的 Google Cloud 资源的用量。

• CREDENTIALS：（可选）用于创建此 PipelineJob 的自定义凭据。将替换 aiplatform.init 中设置的凭据。

• PROJECT_ID（可选）：要在其中运行流水线的 Google Cloud 项目。如果未设置此参数，则使用 aiplatform.init 中设置的项目。

• LOCATION（可选）：要在其中运行流水线的区域。如需详细了解支持 Agent Platform Pipelines 的区域，请参阅 Gemini Enterprise Agent Platform 位置指南。如果您未设置此参数，则使用 aiplatform.init 中设置的默认位置。

• FAILURE_POLICY：（可选）指定整个流水线的失败政策。您可以使用以下配置：

  • 如需将流水线配置为在任务失败后失败，请输入 fast。

  • 如需将流水线配置为在任务失败后继续安排任务，请输入 slow。

  如果您未设置此参数，则失败政策配置默认设置为 slow。详细了解流水线故障政策。

• SERVICE_ACCOUNT：（可选）要用于此流水线运行的服务账号的名称。如果您未指定服务账号，Agent Platform Pipelines 会使用默认 Compute Engine 服务账号运行流水线。

• NETWORK：（可选）要用于此流水线运行的 VPC 对等互连网络的名称。

• RESERVED_IP_RANGES：可选。用于流水线运行工作负载的 VPC 网络中预留的 IP 范围的名称列表，例如：['vertex-ai-ip-range']。

后续步骤

• 了解如何创建流水线运行作业并将其添加到实验或实验运行作业。