使用 Python SDK编写 Kubeflow Pipeline

Code #

import os

import kfp
import kfp.components as comp


# 定义一个函数，接受两个数字作为输入，返回它们的和
def add(a: float, b: float) -> float:
    '''Calculates sum of two arguments'''
    return a + b


# 定义一个函数，接受两个数字作为输入，返回它们的差
def subtract(a: float, b: float) -> float:
    '''Calculates difference of two arguments'''
    return a - b


# 定义一个函数，接受两个数字作为输入，返回它们的积
def multiply(a: float, b: float) -> float:
    '''Calculates product of two arguments'''
    return a * b


# 定义一个函数，接受两个数字作为输入，返回它们的商
def divide(a: float, b: float) -> float:
    '''Calculates quotient of two arguments'''
    return a / b


# 把函数转换成组件
add_op = comp.create_component_from_func(add, base_image='python:3.7')
subtract_op = comp.create_component_from_func(subtract, base_image='python:3.7')
multiply_op = comp.create_component_from_func(multiply, base_image='python:3.7')
divide_op = comp.create_component_from_func(divide, base_image='python:3.7')


# 定义一个 pipeline，接受两个数字作为输入，依次执行加减乘除的操作，并输出结果
@kfp.dsl.pipeline(
    name='Calculator',
    description='A simple calculator pipeline.'
)
def calculator_pipeline(
        a='1',
        b='1'
):
    # 创建第一个任务，调用 add_op 组件，把 a 和 b 作为输入
    add_task = add_op(a, b)
    # 创建第二个任务，调用 subtract_op 组件，把 a 和 b 作为输入
    subtract_task = subtract_op(a, b)
    # 创建第三个任务，调用 multiply_op 组件，把 a 和 b 作为输入
    multiply_task = multiply_op(a, b)
    # 创建第四个任务，调用 divide_op 组件，把 a 和 b 作为输入
    divide_task = divide_op(a, b)
    # 打印输出结果
    print(add_task.output)
    print(subtract_task.output)
    print(multiply_task.output)
    print(divide_task.output)


if __name__ == '__main__':
    # 创建一个编译器对象
    compiler = kfp.compiler.Compiler()
    # 把 pipeline 编译成 yaml 文件
    compiler.compile(calculator_pipeline, 'calculator.yaml')