Kubeflow 流水线 - GitHub 问题摘要

1. 简介

Kubeflow 是一个面向 Kubernetes 的机器学习工具包。该项目旨在让您可以在 Kubernetes 上以可移植、可扩缩的方式轻松部署机器学习 (ML) 工作流。其目标是提供一种简单直接的方法，将机器学习领域的同类最佳开源系统部署到各种基础架构中。

机器学习工作流可能涉及许多相互依赖的步骤，从数据准备和分析到训练、评估、部署等。以临时方式（例如在一组笔记本或脚本中）撰写和跟踪这些流程非常困难，并且审核和可重现性等问题会变得越来越严重。Kubeflow Pipelines (KFP) 提供了一种部署稳健且可重复的机器学习流水线的方法，同时还提供监控、审核、版本跟踪和可重现性功能，有助于解决这些问题。借助 Cloud AI Pipelines，您可以轻松设置 KFP 安装。

构建内容

在此 Codelab 中，您将构建一个 Web 应用，该应用使用 Kubeflow Pipelines 训练和提供模型，以总结 GitHub 问题。它基于 Kubeflow 示例代码库中的示例。完成上述步骤后，您的基础设施将包含：

安装了 Kubeflow Pipelines（通过云 AI Pipelines）的 Google Kubernetes Engine (GKE) 集群。
在 GPU 上训练 Tensor2Tensor 模型的流水线
一种可根据训练后的模型提供预测结果的服务容器
一种可解读预测结果并为 GitHub 问题提供总结的界面
一个使用 Kubeflow Pipelines (KFP) SDK 从头开始创建流水线的笔记本

学习内容

您将构建的流水线会在 GitHub 问题数据上训练 Tensor2Tensor 模型，学习根据问题正文预测问题标题。然后，它会导出训练后的模型，并使用 Tensorflow Serving 部署导出的模型。流水线的最后一步会启动一个 Web 应用，该应用会与 TF-Serving 实例互动，以获取模型预测结果。

如何在 GKE 集群上安装 Kubeflow Pipelines
如何使用 Kubeflow Pipelines 构建和运行机器学习工作流
如何从 AI Platform Notebook 定义和运行流水线

所需条件

对 Kubernetes 有基本了解会很有帮助，但并非必要条件
您拥有所有者权限的有效 GCP 项目
（可选）GitHub 账号
拥有对 Google Cloud Shell 的访问权限，该 Shell 可在 Google Cloud Platform (GCP) 控制台中使用

2. 设置

Cloud Shell

在浏览器中访问 GCP Console，然后使用您的项目凭据登录：

如有必要，请点击“选择项目”，以便您使用 Codelab 项目。

然后，点击控制台右上角的“激活 Cloud Shell”图标，启动 Cloud Shell。

启动 Cloud Shell 时，系统会告知您它已设置为使用的项目名称。检查此设置是否正确。

如需查找项目 ID，请访问 GCP 控制台的“首页”面板。如果屏幕为空，请在提示中点击“是”以创建信息中心。

然后，在 Cloud Shell 终端中，根据需要运行以下命令，以配置 gcloud 使用正确的项目：

export PROJECT_ID=<your_project_id>
gcloud config set project ${PROJECT_ID}

创建存储分区

创建一个 Cloud Storage 存储分区来存储流水线文件。您需要使用全局唯一的 ID，因此定义包含项目 ID 的存储分区名称非常方便。使用 gsutil mb（创建存储分区）命令创建存储分区：

export PROJECT_ID=<your_project_id>
export BUCKET_NAME=kubeflow-${PROJECT_ID}
gsutil mb gs://${BUCKET_NAME}

或者，您也可以通过 GCP 控制台创建存储分区。

可选：创建 GitHub 令牌

此 Codelab 会调用 GitHub API 来检索公开提供的数据。为防止速率限制，尤其是在向 GitHub API 发送大量匿名化请求的活动中，请设置不含任何权限的访问令牌。这只是为了验证您的身份，而不是匿名用户。

前往 https://github.com/settings/tokens，然后生成一个不包含任何范围的新令牌。
将其保存在安全的地方。如果丢失，您需要删除并创建一个新密钥。

如果您跳过此步骤，实验仍可正常进行，只是在生成输入数据来测试模型时，您的选择会受到一些限制。

可选：固定有用的信息中心

在 GCP 控制台中，固定 Kubernetes Engine 和 Storage 信息中心，以便更轻松地访问。

创建 AI Platform Pipelines（托管式 Kubeflow Pipelines）安装

按照此处“准备工作”和“设置实例”部分中的说明，设置已安装 KFP 的 GKE 实例。务必选中允许访问以下 Cloud API 对应的复选框，如文档中所示。（如果您不这样做，示例流水线将无法成功运行）。将安装命名空间保留为 default。

您需要选择支持 Nvidia K80 的可用区。您可以将 us-central1-a 或 us-central1-c 用作默认值。

安装完成后，在 AI Pipelines 信息中心内记下安装所列的 GKE 集群名称和地区，并为方便起见，将环境变量设置为这些值。

export ZONE=<your zone>
export CLUSTER_NAME=<your cluster name>

设置 kubectl 以使用新 GKE 集群的凭据

创建 GKE 集群后，在 Cloud Shell 中运行以下命令，将 kubectl 配置为使用新集群的凭据：

gcloud container clusters get-credentials ${CLUSTER_NAME} \
  --project ${PROJECT_ID} \
  --zone ${ZONE}

或者，点击 AI Pipelines 信息中心内的集群名称以访问其 GKE 页面，然后点击页面顶部的“连接”。在弹出式窗口中，将该命令粘贴到 Cloud Shell 中。

这会配置您的 kubectl 上下文，以便您与集群进行互动。如需验证配置，请运行以下命令：

kubectl get nodes -o wide

您应该会看到列出的节点，其状态为“Ready”，以及有关节点使用时长、版本、外部 IP 地址、操作系统映像、内核版本和容器运行时的其他信息。

配置集群以在支持 GPU 的节点池上安装 Nvidia 驱动程序

接下来，我们将向集群应用一个 daemonset，该 daemonset 会在任何支持 GPU 的集群节点上安装 Nvidia 驱动程序：

kubectl apply -f https://raw.githubusercontent.com/GoogleCloudPlatform/container-engine-accelerators/master/nvidia-driver-installer/cos/daemonset-preloaded.yaml

然后运行以下命令，该命令会授予 KFP 组件创建新 Kubernetes 资源的权限：

kubectl create clusterrolebinding sa-admin --clusterrole=cluster-admin --serviceaccount=kubeflow:pipeline-runner

创建 GPU 节点池

然后，我们将设置一个大小为 1 的 GPU 节点池：

gcloud container node-pools create gpu-pool \
    --cluster=${CLUSTER_NAME} \
    --zone ${ZONE} \
    --num-nodes=1 \
    --machine-type n1-highmem-8 \
    --scopes cloud-platform --verbosity error \
    --accelerator=type=nvidia-tesla-k80,count=1

3. 从“流水线”信息中心运行流水线

打开流水线信息中心

在 Cloud 控制台中，访问流水线面板（如果您尚未访问）。然后，点击相应安装的“打开流水线信息中心”，再点击左侧菜单栏中的流水线。如果您收到加载错误消息，请刷新相应标签页。您应该会看到如下所示的新页面：

流水线说明

您将运行的流水线包含多个步骤（详情请参阅本 Codelab 的附录）：

现有模型检查点已复制到您的存储分区。
使用预处理的数据训练 Tensor2Tensor 模型。

训练从第一步中复制的现有模型检查点开始，然后继续训练几百步。（在 Codelab 期间，完全训练模型需要花费太长时间）。
训练完成后，流水线步骤会以适合 TensorFlow Serving 的形式导出模型。

使用该模型部署 TensorFlow Serving 实例。
启动一个 Web 应用，用于与已部署的模型交互以检索预测结果。

下载并编译流水线

在本部分中，我们将了解如何编译流水线定义。首先，我们需要安装 KFP SDK。在 Cloud Shell 中运行以下命令：

pip3 install -U kfp

如需下载流水线定义文件，请在 Cloud Shell 中执行以下命令：

curl -O https://raw.githubusercontent.com/amygdala/kubeflow-examples/ghsumm/github_issue_summarization/pipelines/example_pipelines/gh_summ_hosted_kfp.py

然后，运行流水线定义文件，如下所示，以编译该文件：

python3 gh_summ_hosted_kfp.py

您会看到文件 gh_summ_hosted_kfp.py.tar.gz 显示为结果。

上传已编译的流水线

在 Kubeflow Pipelines Web 界面中，点击上传流水线，然后选择通过网址导入。复制以下网址（指向您刚刚编译的同一流水线），然后粘贴。（从 Cloud Shell 上传文件需要执行一些额外的步骤，因此我们采取了捷径）。

https://storage.googleapis.com/aju-dev-demos-codelabs/KF/compiled_pipelines/gh_summ_hosted_kfp.py.tar.gz

为流水线命名（例如 gh_summ）。

运行流水线

点击列表中的已上传流水线，即可查看流水线的静态图表，然后点击创建实验，以使用该流水线创建新的实验。实验是一种将语义上相关的运行分组在一起的方式。

为实验命名（例如与流水线相同的名称 gh_summ），然后点击下一步以创建实验。

系统随即会打开一个页面，您可以在其中输入运行的参数并开始运行。

您可能需要在 Cloud Shell 中执行以下命令，以帮助填写参数。

gcloud config get-value project
echo "gs://${BUCKET_NAME}/codelab"

系统会自动填充跑步名称，但您可以根据需要为其指定其他名称。

然后，填写三个参数字段：

project
（可选）github-token
working-dir

对于工作目录，请输入您创建的 GCS 存储分区下的某个路径。添加“gs://”前缀。对于 github-token 字段，请输入您之前选择性生成的令牌，或者如果您未生成令牌，请将占位符字符串保留原样。

填写完相应字段后，点击开始，然后点击列出的跑步活动以查看其详细信息。在给定的流水线步骤运行时，您可以点击该步骤以获取有关它的更多信息，包括查看其 pod 日志。（您还可以通过指向流水线步骤的 Cloud Logging (Stackdriver) 日志的链接查看流水线步骤的日志，即使集群节点已被拆除）。

查看流水线定义

在流水线运行期间，您可能需要仔细了解其组成方式和运行方式。如需了解详情，请参阅此 Codelab 的附录部分。

在 TensorBoard 中查看模型训练信息

训练步骤完成后，选择其可视化图表标签页，然后点击蓝色的启动 TensorBoard 按钮，待其准备就绪后，点击打开 TensorBoard。

探索“工件和执行”信息中心

Kubeflow 流水线会在执行时自动记录有关流水线步骤的元数据。系统会记录制品和执行信息。点击信息中心左侧导航栏中的这些条目可进一步探索。

对于制品，您可以同时查看概览面板和沿袭浏览器面板。

启动流水线创建的 Web 应用并进行一些预测

流水线的最后一步是部署一个 Web 应用，该应用提供了一个界面，用于查询通过 TF Serving 提供的经过训练的模型，以进行预测。

流水线完成后，通过端口转发连接到 Web 应用的服务（我们之所以进行端口转发，是因为在本 Codelab 中，Web 应用服务未设置为具有外部端点）。

在 Cloud Shell 中运行以下命令，找到服务名称：

kubectl get services

在列表中查找类似 ghsumm-*-webappsvc 的服务名称。

然后，在 Cloud Shell 中，按如下方式将端口转发到该服务，将以下命令更改为使用您的 webappsvc 的名称：

kubectl port-forward svc/ghsumm-xxxxx-webappsvc 8080:80

端口转发运行后，点击 Cloud Shell 窗格上方的“预览”图标，然后在下拉菜单中点击“在端口 8080 上预览”。

您应该会在新标签页中看到如下所示的页面：

点击 Populate Random Issue 按钮可检索一段文字。点击生成标题以调用训练好的模型并显示预测结果。

如果您的流水线参数包含有效的 GitHub 令牌，您也可以尝试在第二个字段中输入 GitHub 网址，然后点击“生成标题”。如果您未设置有效的 GitHub 令牌，请仅使用“填充随机问题”字段。

4. 从 AI Platform Notebook 运行流水线

您还可以使用 KFP SDK 从 Jupyter 笔记本中以交互方式定义和运行 Kubeflow Pipelines。AI Platform Notebooks（我们将在此 Codelab 中使用）可让您轻松完成此任务。

创建笔记本实例

我们将使用 Cloud Shell 的 API 创建一个笔记本实例。（或者，您也可以通过 Cloud 控制台创建笔记本。如需了解详情，请参阅相关文档）。

在 Cloud Shell 中，设置以下环境变量：

export INSTANCE_NAME="kfp-ghsumm"
export VM_IMAGE_PROJECT="deeplearning-platform-release"
export VM_IMAGE_FAMILY="tf2-2-3-cpu"
export MACHINE_TYPE="n1-standard-4"
export LOCATION="us-central1-c"

然后，在 Cloud Shell 中，运行命令以创建笔记本实例：

gcloud beta notebooks instances create $INSTANCE_NAME \
  --vm-image-project=$VM_IMAGE_PROJECT \
  --vm-image-family=$VM_IMAGE_FAMILY \
  --machine-type=$MACHINE_TYPE --location=$LOCATION

首次运行此命令时，系统可能会要求您为项目启用 notebooks API。如果符合，请回复“y”。

几分钟后，笔记本服务器将启动并正常运行。您可以在 Cloud 控制台中看到列出的笔记本实例。

上传 Codelab 笔记本

创建笔记本实例后，点击此链接上传 Codelab 的 Jupyter 笔记本。选择要使用的笔记本实例。系统会自动打开笔记本。

执行笔记本

在实验的剩余步骤中，请按照笔记本中的说明操作。请注意，在笔记本的“设置”部分，您需要先填写自己的值，然后才能运行笔记本的其余部分。

（如果您使用的是自己的项目，请务必返回并完成本实验的“清理”部分）。

5. 清理

如果您使用的是临时 Codelab 账号，则无需执行此操作；但如果您使用的是自己的项目，则可能需要移除流水线安装和笔记本。

关闭 Pipelines GKE 集群

您可以从 Cloud 控制台中删除 Pipelines 集群。（如果您想重复使用 GKE 集群，可以选择仅删除 Pipelines 安装）。

删除 AI Notebook 实例

如果您已运行此 Codelab 的“笔记本”部分，可以从 Cloud 控制台删除或停止笔记本实例。

可选：移除 GitHub 令牌

前往 https://github.com/settings/tokens，然后移除生成的令牌。

6. 附录

代码一览

定义流水线

此 Codelab 中使用的流水线定义在此处。

我们来看看它的定义方式，以及其组成部分（步骤）的定义方式。我们会介绍一些亮点，但如需了解详情，请参阅文档。

Kubeflow 流水线步骤基于容器。构建流水线时，您可以使用已构建容器映像的预构建组件，也可以自行构建组件。在此 Codelab 中，我们构建了自己的。

其中四个流水线步骤是通过其组件定义文件访问的可重用组件定义的。在第一个代码段中，我们通过网址访问这些组件定义文件，并使用这些定义创建“操作”，然后使用这些操作创建流水线步骤。

import kfp.dsl as dsl
import kfp.gcp as gcp
import kfp.components as comp

...

copydata_op = comp.load_component_from_url(
  'https://raw.githubusercontent.com/kubeflow/examples/master/github_issue_summarization/pipelines/components/t2t/datacopy_component.yaml'
  )

train_op = comp.load_component_from_url(
  'https://raw.githubusercontent.com/kubeflow/examples/master/github_issue_summarization/pipelines/components/t2t/train_component.yaml'
  )

以下是训练操作的组件定义（采用 YAML 格式）。您可以看到，其输入、输出、容器映像和容器入口点实参均已定义。

name: Train T2T model
description: |
  A Kubeflow Pipeline component to train a Tensor2Tensor
  model
metadata:
  labels:
    add-pod-env: 'true'
inputs:
  - name: train_steps
    description: '...'
    type: Integer
    default: 2019300
  - name: data_dir
    description: '...'
    type: GCSPath
  - name: model_dir
    description: '...'
    type: GCSPath
  - name: action
    description: '...'
    type: String
  - name: deploy_webapp
    description: '...'
    type: String
outputs:
  - name: launch_server
    description: '...'
    type: String
  - name: train_output_path
    description: '...'
    type: GCSPath
  - name: MLPipeline UI metadata
    type: UI metadata
implementation:
  container:
    image: gcr.io/google-samples/ml-pipeline-t2ttrain:v3ap
    args: [
      --data-dir, {inputValue: data_dir},
      --action, {inputValue: action},
      --model-dir, {inputValue: model_dir},
      --train-steps, {inputValue: train_steps},
      --deploy-webapp, {inputValue: deploy_webapp},
      --train-output-path, {outputPath: train_output_path}
    ]
    env:
      KFP_POD_NAME: "{{pod.name}}"
    fileOutputs:
      launch_server: /tmp/output
      MLPipeline UI metadata: /mlpipeline-ui-metadata.json

您还可以通过 dsl.ContainerOp 构造函数定义流水线步骤，如下所示。

以下是流水线定义的大部分内容。我们正在定义流水线输入（及其默认值）。然后，我们定义流水线步骤。对于大多数步骤，我们使用的是上面定义的“ops”，但我们还通过 ContainerOp 内联定义了“serve”步骤，直接指定容器映像和入口点实参。

您可以看到，train、log_model 和 serve 步骤正在访问之前步骤的输出作为输入。如需详细了解如何指定此项，请点击此处。

@dsl.pipeline(
 name='Github issue summarization',
 description='Demonstrate Tensor2Tensor-based training and TF-Serving'
)
def gh_summ(  #pylint: disable=unused-argument
 train_steps: 'Integer' = 2019300,
 project: str = 'YOUR_PROJECT_HERE',
 github_token: str = 'YOUR_GITHUB_TOKEN_HERE',
 working_dir: 'GCSPath' = 'gs://YOUR_GCS_DIR_HERE',
 checkpoint_dir: 'GCSPath' = 'gs://aju-dev-demos-codelabs/kubecon/model_output_tbase.bak2019000/',
 deploy_webapp: str = 'true',
 data_dir: 'GCSPath' = 'gs://aju-dev-demos-codelabs/kubecon/t2t_data_gh_all/'
 ):


 copydata = copydata_op(
   data_dir=data_dir,
   checkpoint_dir=checkpoint_dir,
   model_dir='%s/%s/model_output' % (working_dir, dsl.RUN_ID_PLACEHOLDER),
   action=COPY_ACTION,
   )


 train = train_op(
   data_dir=data_dir,
   model_dir=copydata.outputs['copy_output_path'],
   action=TRAIN_ACTION, train_steps=train_steps,
   deploy_webapp=deploy_webapp
   )

 serve = dsl.ContainerOp(
     name='serve',
     image='gcr.io/google-samples/ml-pipeline-kubeflow-tfserve:v6',
     arguments=["--model_name", 'ghsumm-%s' % (dsl.RUN_ID_PLACEHOLDER,),
         "--model_path", train.outputs['train_output_path']
         ]
     )

 train.set_gpu_limit(1)

请注意，我们要求“train”步骤在集群中至少有 1 个可用 GPU 的节点上运行。

  train.set_gpu_limit(1)

流水线中的最后一步（也以内嵌方式定义）是条件性的。它将在“serve”步骤完成后运行，前提是训练步骤 launch_server 的输出为字符串“true”。它会启动“预测 Web 应用”，我们之前使用该应用从训练后的 T2T 模型请求问题摘要。

 with dsl.Condition(train.outputs['launch_server'] == 'true'):
   webapp = dsl.ContainerOp(
       name='webapp',
       image='gcr.io/google-samples/ml-pipeline-webapp-launcher:v1',
       arguments=["--model_name", 'ghsumm-%s' % (dsl.RUN_ID_PLACEHOLDER,),
           "--github_token", github_token]

       )
   webapp.after(serve)

组件容器映像定义

Kubeflow 流水线文档介绍了一些用于构建自有组件的最佳实践。在此过程中，您需要定义并构建容器映像。您可以点击此处查看此 Codelab 的流水线的组件步骤。Dockerfile 定义位于 containers 子目录中，例如此处。

使用配备 GPU 的抢占式虚拟机进行训练

抢占式虚拟机是 Compute Engine 虚拟机实例，最多可存在 24 小时，并且不提供任何可用性保证。抢占式虚拟机的价格低于标准 Compute Engine 虚拟机。

借助 Google Kubernetes Engine (GKE)，您可以轻松设置使用抢占式虚拟机的集群或节点池。您可以设置这样一个节点池，将 GPU 挂接到抢占式实例。这些节点的工作方式与常规的启用 GPU 的节点相同，但 GPU 只会在实例的生命周期内保留。

您可以运行类似于以下内容的命令，为集群设置抢占式、支持 GPU 的节点池。请根据您的要求，使用集群名称和可用区修改以下命令，并调整加速器类型和数量。您可以选择定义节点池，使其根据当前工作负载自动扩缩。

gcloud container node-pools create preemptible-gpu-pool \
    --cluster=<your-cluster-name> \
    --zone <your-cluster-zone> \
    --enable-autoscaling --max-nodes=4 --min-nodes=0 \
    --machine-type n1-highmem-8 \
    --preemptible \
    --node-taints=preemptible=true:NoSchedule \
    --scopes cloud-platform --verbosity error \
    --accelerator=type=nvidia-tesla-k80,count=4

您还可以通过 Cloud 控制台设置节点池。

定义使用可抢占 GKE 节点的 Kubeflow 流水线

如果您在 GKE 上运行 Kubeflow，现在可以轻松定义和运行 Kubeflow 流水线，其中一个或多个流水线步骤（组件）在可抢占节点上运行，从而降低作业运行成本。为了确保使用抢占式虚拟机时获得正确的结果，您确定为可抢占的步骤应为幂等（即，如果您多次运行某个步骤，它将产生相同的结果），或者应检查点工作，以便在中断时，该步骤可以从中断处继续执行。

定义 Kubeflow 流水线时，您可以通过修改操作来指明某个指定步骤应在可抢占节点上运行，如下所示：

your_pipelines_op.apply(gcp.use_preemptible_nodepool())

如需了解详情，请参阅文档。

如果节点被抢占，您可能还希望重试该步骤若干次。您可以按如下方式操作 - 在此示例中，我们指定了 5 次重试。

your_pipelines_op.set_gpu_limit(1).apply(gcp.use_preemptible_nodepool()).set_retry(5)

尝试修改我们在本 Codelab 中使用的 Kubeflow 流水线，以在抢占式虚拟机上运行训练步骤。

更改流水线规范中的以下行，以额外使用可抢占节点池（确保您已按照上述说明创建了一个），并重试 5 次：

  train.set_gpu_limit(1)

然后，重新编译流水线，上传新版本（为其指定新名称），然后运行新版本的流水线。