耿雨飞的个人博客

[[ slide_text ]]

[[ item.c ]]

Kubeflow 构建 kfp-component

发布时间：2023-05-23 作者：gengyufei 来源：gengyufei个人博客

Kubeflow

# 概述
本文主要介绍：如何创建component以及如何在kubeflow pipeline中使用component

kfp-component 我们称其为组件，在机器学习的工作流中，扮演step角色（即：整体流程中的一个步骤）。以下内容描述了：
- 构建组件所需的概念
- 如何开始构建组件

![](https://tcs-devops.aliyuncs.com/storage/112ufa017246e2d390285474fd251dedb9f9?Signature=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJBcHBJRCI6IjVlNzQ4MmQ2MjE1MjJiZDVjN2Y5YjMzNSIsIl9hcHBJZCI6IjVlNzQ4MmQ2MjE1MjJiZDVjN2Y5YjMzNSIsIl9vcmdhbml6YXRpb25JZCI6IjVmZTg3YjM5ZWRiY2I1MjljNjI4OGVhMyIsImV4cCI6MTY4NTQ5ODA0NywiaWF0IjoxNjg0ODkzMjQ3LCJyZXNvdXJjZSI6Ii9zdG9yYWdlLzExMnVmYTAxNzI0NmUyZDM5MDI4NTQ3NGZkMjUxZGVkYjlmOSJ9.-c9QUdc6LfXb-wcatu_y5dkTo1ap2XGycBMYpVDFka8)

# 开始之前
在开始之前，需要在自己的开发环境中，下载kfp-sdk：
```
pip3 install kfp --upgrade -i http://pypi.douban.com/simple/
```
> 注意：python 版本 ≥ 3.7
kfp-sdk ：https://kubeflow-pipelines.readthedocs.io/en/stable/index.html

# 初识component
Pipeline component 本质就是工作流中的某一个步骤，例如：数据预处理、模型训练等等。创建一个组件需要：
- 定义**组件实现**：准备component需要**执行的代码**以及**docker镜像**（待执行代码在该docker镜像所启动的容器中运行）
- 定义**组件规范**: 即创建出来的component对外提供的接口规范

## 组件规范
组件规范定义，包含以下内容：
- 组件的输入与输出
- 组件代码运行所在的**docker容器映像**，用于运行组件代码的**cmd命令**，以及传递给组件代码的**命令行参数**
- 组件的元数据：如name、描述等

> 注意：如果component是以python函数的形式提供的，则使用kfp-sdk将python函数打包为组件

# 设计组件
当kubeflow-pipeline执行一个component时，会在k8s的pod中启动一个docker容器(该容器中包含待执行的代码)，组件的输入inputs，以命令行参数的形式传入docker容器。
- 小型inputs，如string、int，可以通过命令行参数的方式传入
- 大型inputs，如csv data文件，必须以**文件路径path**传入

当组件**运行结束**后，**运行结果以文件的形式返回**。

当在设计component组件时，需要考虑以下几个因素：
- 输入inputs：
	- 值传递（即：直接通过命令行传参）：包括数字、布尔值和短字符串
	- path传递：大型数据文件，通过对路径值的引用，传递给命令行
- 输出outputs：
	- 必须将组件的产出存储为文件
	- 在定义组件时，需要让kubeflow pipeline知道组件的产出
	- 在kubeflow pipeline运行组件时，会将保存组件输出的路径作为输入，传递给组件
	- 输出一般以单个文件的形式保存；但某些情况下，需要以目录作为输出，则需要在输出路径下，创建对应的目录并输出多个文件至该目录
	- 组件的目标可能是在一个外部服务中创建一个数据集，比如一个BigQuery表。在这种情况下，组件的输出则变为某种标识符（如表名）而非数据本身。建议将此模式限制在必须将数据放入外部系统而不是将其保存在Kubeflow pipeline系统中的情况下
- 命令行参数：
	- 由于输入和输出路径是作为命令行参数传入的，因此组件的代码必须能够从命令行读取输入
	- 如果组件是用Python构建的，那么```argparse``` 和 ```absl.flags``` 等库可以更容易地读取组件的输入
- 组件代码：
	- 组件的代码可以用任何语言实现，只要它可以在容器映像中运行即可

# 简单demo
该demo实现读取指定文件的前n行数据并保存至输出文件，其中command-line parameters:
- 输入文件的path
- 需要读取文件的行数
- 输出文件的path

```
#!/usr/bin/env python3
import argparse
from pathlib import Path

# Function doing the actual work (Outputs first N lines from a text file)
def do_work(input1_file, output1_file, param1):
  for x, line in enumerate(input1_file):
    if x >= param1:
      break
    _ = output1_file.write(line)
  
# Defining and parsing the command-line arguments
parser = argparse.ArgumentParser(description='My program description')
# Paths must be passed in, not hardcoded
parser.add_argument('--input1-path', type=str,
  help='Path of the local file containing the Input 1 data.')
parser.add_argument('--output1-path', type=str,
  help='Path of the local file where the Output 1 data should be written.')
parser.add_argument('--param1', type=int, default=100,
  help='The number of lines to read from the input and write to the output.')
args = parser.parse_args()

# Creating the directory where the output file is created (the directory
# may or may not exist).
Path(args.output1_path).parent.mkdir(parents=True, exist_ok=True)

with open(args.input1_path, 'r') as input1_file:
    with open(args.output1_path, 'w') as output1_file:
        do_work(input1_file, output1_file, args.param1)
```

假设该程序保存为 program.py, 则调用该程序的命令：
```
python3 program.py --input1-path <path-to-the-input-file> \
  --param1 <number-of-lines-to-read> \
  --output1-path <path-to-write-the-output-to> 
```

# 将组件代码打包为docker镜像
kubeflow pipeline成功运行组件的前提：组件代码打包为docker镜像，并发布到k8s可以访问到的镜像仓库（如：harbor）。本节主要描述如何将组件代码打包至docker镜像。

## 创建 dockerfile
核心3要素：
- base image：基础镜像
- 相关依赖
- 组件代码

```
FROM python:3.7
RUN python3 -m pip install keras
COPY ./src /pipelines/component/src
```

> 在本示例中，
- base image： python:3.7
- 依赖下载：Keras
- 本地源码路径./src 下的内容，被拷贝至镜像中的 /pipelines/component/src 路径下

## 编写打包脚本
主要功能：
- 构建component组件的docker镜像
- 将docker镜像推送至镜像仓库

```
#!/bin/bash -e
image_name=gcr.io/my-org/my-image
image_tag=latest
full_image_name=${image_name}:${image_tag}

cd "$(dirname "$0")" 
docker build -t "${full_image_name}" .
docker push "$full_image_name"

# Output the strict image name, which contains the sha256 image digest
docker inspect --format="{{index .RepoDigests 0}}" "${full_image_name}"
```

## 执行打包脚本
执行打包脚本

## docker run 本地测试
通过docker run 对目标component镜像进行本地测试

# 创建组件规范
## component.yaml
```
##############################################
# metadata
##############################################
name: Get Lines
description: Gets the specified number of lines from the input file.

##############################################
# inputs & outputs
##############################################
inputs:
- {name: Input 1, type: String, description: 'Data for input 1'}
- {name: Parameter 1, type: Integer, default: '100', description: 'Number of lines to copy'}

outputs:
- {name: Output 1, type: String, description: 'Output 1 data.'}

##############################################
# container
##############################################
implementation:
  container:
    image: gcr.io/my-org/my-image@sha256:a172..752f
    # command is a list of strings (command-line arguments). 
    # The YAML language has two syntaxes for lists and you can use either of them. 
    # Here we use the "flow syntax" - comma-separated strings inside square brackets.
    command: [
      python3, 
      # Path of the program inside the container
      /pipelines/component/src/program.py,
      --input1-path,
      {inputPath: Input 1},
      --param1, 
      {inputValue: Parameter 1},
      --output1-path, 
      {outputPath: Output 1},
    ]
```

注解：
- metadata：包含name、description, 顾名思义
- inputs & outputs：需要根据具体实际情况，完善信息
	- name: kfp.components.load_component_from_text()工厂函数，在实例化时，会以name对应的value作为参数名，传入实际的input信息（详情看下面的具体案例《component 实际应用在 kubeflow pipeline》）
- implementation：核心是组件对应的docker镜像 & command命令行参数

# component 实际应用在 kubeflow pipeline
kfp-sdk 提供3种方式，用于将component应用至kubeflow-pipeline：
- kfp.components.load_component_from_file: 从本地path导入component.yaml
- kfp.components.load_component_from_url: 从网络上拉取component.yaml
- kfp.components.load_component_from_text: 从字符串中获取component.yaml（适用于快速迭代开发，详见下述demo）

本案例，使用 kfp.components.load_component_from_text 的方式来加载组件规范（即：component.yaml），该demo创建了一个包含2个step的pipeline，第一个step负责产出一个文件，第二个step负责读取上游产出的文件的n行数据。
```
import kfp
import kfp.components as comp

create_step_get_lines = comp.load_component_from_text("""
name: Get Lines
description: Gets the specified number of lines from the input file.

inputs:
- {name: Input 1, type: Data, description: 'Data for input 1'}
- {name: Parameter 1, type: Integer, default: '100', description: 'Number of lines to copy'}

outputs:
- {name: Output 1, type: Data, description: 'Output 1 data.'}

implementation:
  container:
    image: gcr.io/my-org/my-image@sha256:a172..752f
    # command is a list of strings (command-line arguments). 
    # The YAML language has two syntaxes for lists and you can use either of them. 
    # Here we use the "flow syntax" - comma-separated strings inside square brackets.
    command: [
      python3, 
      # Path of the program inside the container
      /pipelines/component/src/program.py,
      --input1-path,
      {inputPath: Input 1},
      --param1, 
      {inputValue: Parameter 1},
      --output1-path, 
      {outputPath: Output 1},
    ]""")

# create_step_get_lines is a "factory function" that accepts the arguments
# for the component's inputs and output paths and returns a pipeline step
# (ContainerOp instance).
#
# To inspect the get_lines_op function in Jupyter Notebook, enter 
# "get_lines_op(" in a cell and press Shift+Tab.
# You can also get help by entering `help(get_lines_op)`, `get_lines_op?`,
# or `get_lines_op??`.

# Create a simple component using only bash commands. The output of this component
# can be passed to a downstream component that accepts an input with the same type.
create_step_write_lines = comp.load_component_from_text("""
name: Write Lines
description: Writes text to a file.

inputs:
- {name: text, type: String}

outputs:
- {name: data, type: Data}

implementation:
  container:
    image: busybox
    command:
    - sh
    - -c
    - |
      mkdir -p "$(dirname "$1")"
      echo "$0" > "$1"
    args:
    - {inputValue: text}
    - {outputPath: data}
""")

# Define your pipeline 
def my_pipeline():
    write_lines_step = create_step_write_lines(
        text='one\ntwo\nthree\nfour\nfive\nsix\nseven\neight\nnine\nten')

get_lines_step = create_step_get_lines(
        # Input name "Input 1" is converted to pythonic parameter name "input_1"
        input_1=write_lines_step.outputs['data'],
        parameter_1='5',
    )

# If you run this command on a Jupyter notebook running on Kubeflow,
# you can exclude the host parameter.
# client = kfp.Client()
client = kfp.Client(host='<your-kubeflow-pipelines-host-name>')

# Compile, upload, and submit this pipeline for execution.
client.create_run_from_pipeline_func(my_pipeline, arguments={})
```

> 注意：
输入和输出名称被转换为python格式(空格和符号被下划线替换，字母被转换为小写字母)。例如，将名为```input 1```的输入转换为```input_1```。

# component 文件管理
一般的，我们自定义的component，包含若干代码文件：
- 源代码 source code
- 单元测试 unit test
- 说明文档 README。md
- Dockerfile
- 镜像打包 build_image.sh
- 组件规范 component.yaml

这里提供官方推荐的component代码管理规范：
```
components/<component group>/<component name>/

src/*            # Component source code files
    tests/*          # Unit tests
    run_tests.sh     # Small script that runs the tests
    README.md        # Documentation. If multiple files are needed, move to docs/.

Dockerfile       # Dockerfile to build the component container image
    build_image.sh   # Small script that runs docker build and docker push

component.yaml   # Component definition in YAML format
```

上一篇：Kubeflow 独立部署kubeflow pipeline 下一篇：Kubeflow 基于python函数构建component

你觉得文章怎么样