耿雨飞的个人博客

[[ slide_text ]]

[[ item.c ]]

Kubeflow 基于python函数构建component

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

Kubeflow

# 概述
kubeflow pipeline 提供sdk，用来基于python function 构建component。

# 开始之前
```
pip3 install kfp --upgrade -i http://pypi.douban.com/simple/
```

```
import kfp
from kfp.components import create_component_from_func

# 实例化 client（具体host信息，视自身情况而定）
client = kfp.Client() # change arguments accordingly
```

# 具体步骤
## 定义独立的python函数
定义独立的python函数，需要满足以下要求：
- 不能使用函数定义之外的任何代码
- ```import xxx``` 必须在函数内部声明
- helper 函数，必须定义在函数内部

## python函数input & output 与 component的接口定义
kfp 是使用python函数的 input && output，来定义 component 的接口的。python函数的 input 和 output 需要满足以下条件：
- 若函数接收或返回大数据，又或者数据比较复杂，则输入、输出必须以文件path的形式传递
- 若函数接收数字化的数值作为参数，则必须声明参数类型
	- 数值：int or float
	- 其他： 按照 string 处理
- 若函数返回多个小型输出（如：短字符串、数字 或 布尔值），则需要：使用``` typing.NamedTuple ```类型提示对函数进行注释，并使用```collections.NamedTuple```函数将函数的输出作为元组的新子类返回
-

## （可选）构建 or 选择一个docker镜像
在python函数的依赖项比较复杂的时候，可以自己打包或选择一个docker镜像，用于执行 python 函数

## 调用 kfp-sdk 实现 python函数 至 component的转换
具体步骤：
- **func**: 定义的python函数
- **base_image(可选)**: 指定python函数运行时，使用的docker镜像（系统默认：python:3.7  docker image）
- **output_component_file（可选）**：输出component.yaml的定义文件
- **packages_to_install(可选)**：在运行python函数之前，需要提前下载指定版本的python 包

# 解析-数据是如何在components之间流转的
当Kubeflow Pipelines运行组件时，docker容器在Kubernetes Pod中启动，组件的输入作为命令行参数传入。当组件完成后，组件的输出将作为文件返回。

以下描述如何**按值**和**按文件**传递参数：
- 按值传递的参数包括数字、布尔值和短字符串。Kubeflow Pipelines通过将值作为命令行参数传递给组件，从而按值传递参数。
- 通过file传递的参数包括**CSV、docker镜像和其他复杂类型**。这些文件存储在组件可以访问到的Kubernetes系统的某个位置，例如**持久卷声明PVC**或**云存储服务**。Kubeflow Pipelines通过**将参数的路径作为命令行参数传递给组件**，从而**将参数通过文件传递给组件**。

# Input and output parameter names
当使用Kubeflow Pipelines SDK将Python函数转换为管道组件时，Kubeflow Pipelines SDK使用该函数的参数以以下方式定义组件的接口:

- 一些参数定义了输入参数。
- 一些参数定义输出参数。
- 函数的返回值被用作输出参数。如果返回值是collections.namedtuple，则命名元组用于返回几个小值。

**Kubeflow pipeline SDK**使用以下**规则**来定义组件接口中的**输入和输出参数名称**:

- 如果参数名以_path结尾，并且参数被注释为kfp.components.InputPath或kfp.components.OutputPath，则参数名是去掉了后面的_path的参数名。
- 如果实参名称以_file结尾，则形参名称是去掉后面_file的实参名称。
- 如果使用return语句从组件返回一个小值，则输出参数将命名为output。
- 如果您通过返回collections.namedtuple从组件返回几个小值，Kubeflow Pipelines SDK将使用元组的字段名作为输出参数名。

其他情况，Kubeflow Pipelines SDK将使用实际的函数名作为参数名。

# 案例
本案例演示如何基于python函数来创建component，其中体现了：
- 使用 ```import xxx```
- 使用 ```helper func```
- 产出 多个outputs

## step-1：定义python函数
```
from typing import NamedTuple

def my_divmod(
  dividend: float,
  divisor: float) -> NamedTuple(
    'MyDivmodOutput',
    [
      ('quotient', float),
      ('remainder', float),
      ('mlpipeline_ui_metadata', 'UI_metadata'),
      ('mlpipeline_metrics', 'Metrics')
    ]):
    '''Divides two numbers and calculate  the quotient and remainder'''

# Import the numpy package inside the component function
    import numpy as np

# Define a helper function
    def divmod_helper(dividend, divisor):
        return np.divmod(dividend, divisor)

(quotient, remainder) = divmod_helper(dividend, divisor)

from tensorflow.python.lib.io import file_io
    import json

# Export a sample tensorboard
    metadata = {
      'outputs' : [{
        'type': 'tensorboard',
        'source': 'gs://ml-pipeline-dataset/tensorboard-train',
      }]
    }

# Export two metrics
    metrics = {
      'metrics': [{
          'name': 'quotient',
          'numberValue':  float(quotient),
        },{
          'name': 'remainder',
          'numberValue':  float(remainder),
        }]}

from collections import namedtuple
    divmod_output = namedtuple('MyDivmodOutput',
        ['quotient', 'remainder', 'mlpipeline_ui_metadata',
         'mlpipeline_metrics'])
    return divmod_output(quotient, remainder, json.dumps(metadata),
                         json.dumps(metrics))
```

## step-2: 调用kfp-sdk
调用 ```kfp.components.create_component_from_func``` 产出工厂函数，此工厂函数可以创建一个 ```kfp.dsl.ContainerOp``` 类实例，提供给pipeline
```
divmod_op = comp.create_component_from_func(
    my_divmod, base_image='tensorflow/tensorflow:1.11.0-py3')
```

## step-3: 定义 pipeline
```
import kfp.dsl as dsl
@dsl.pipeline(
   name='Calculation pipeline',
   description='An example pipeline that performs arithmetic calculations.'
)
def calc_pipeline(
   a='1',
   b='7',
   c='17',
):
    # Passes a pipeline parameter and a constant value as operation arguments.
    add_task = add_op(a, 4) # The add_op factory function returns
                            # a dsl.ContainerOp class instance.

# Passes the output of the add_task and a pipeline parameter as operation
    # arguments. For an operation with a single return value, the output
    # reference is accessed using `task.output` or
    # `task.outputs['output_name']`.
    divmod_task = divmod_op(add_task.output, b)

# For an operation with multiple return values, output references are
    # accessed as `task.outputs['output_name']`.
    result_task = add_op(divmod_task.outputs['quotient'], c)
```

## 编译 && 运行 pipeline
```
# Specify pipeline argument values
arguments = {'a': '7', 'b': '8'}

# Submit a pipeline run
client.create_run_from_pipeline_func(calc_pipeline, arguments=arguments)
```

上一篇：Kubeflow 构建 kfp-component 下一篇：Kubeflow 最佳练习-设计component

你觉得文章怎么样

耿雨飞的个人博客

Kubeflow 基于python函数 构建component

Kubeflow 基于python函数构建component