在组件之间传递少量数据

参数对于在组件之间传递少量数据非常有用，当组件创建的数据不表示机器学习工件，如模型、数据集或更复杂的数据类型时。

使用内置的Python类型注释指定参数输入和输出：

from kfp import dsl

@dsl.component
def join_words(word: str, count: int = 10) -> str:
    return ' '.join(word for _ in range(count))

KFP 根据以下表格将 Python 类型注释映射到存储在 ML Metadata 中的类型：

Python对象KFP类型
strstring
intnumber
floatnumber
boolboolean
typing.List / listobject
typing.Dict / dictobject

与普通的Python函数一样，输入参数可以有默认值，按标准方式表示：def func(my_string: str = 'default'):

在后台，KFP通过将所有参数序列化为JSON来传递组件之间的参数。

对于所有Python组件（轻量级Python组件和容器化Python组件），参数的序列化和反序列化对用户是不可见的；KFP会自动处理这一过程。

对于 容器组件，输入参数反序列化对用户是不可见的；KFP自动将输入传递给组件。对于容器组件 输出，容器组件中的用户代码必须处理输出参数的序列化，如容器组件：创建组件输出中所述。

输入参数

使用输入参数非常简单。只需用类型和可选的默认值注释您的组件函数。以下管道演示了这一点，使用了一个Python组件、一个容器组件和一个包含所有参数类型作为输入的管道：

from typing import Dict, List
from kfp import dsl

@dsl.component
def python_comp(
    string: str = 'hello',
    integer: int = 1,
    floating_pt: float = 0.1,
    boolean: bool = True,
    dictionary: Dict = {'key': 'value'},
    array: List = [1, 2, 3],
):
    print(string)
    print(integer)
    print(floating_pt)
    print(boolean)
    print(dictionary)
    print(array)


@dsl.container_component
def container_comp(
    string: str = 'hello',
    integer: int = 1,
    floating_pt: float = 0.1,
    boolean: bool = True,
    dictionary: Dict = {'key': 'value'},
    array: List = [1, 2, 3],
):
    return dsl.ContainerSpec(
        image='alpine',
        command=['sh', '-c', """echo $0 $1 $2 $3 $4 $5 $6"""],
        args=[
            string,
            integer,
            floating_pt,
            boolean,
            dictionary,
            array,
        ])

@dsl.pipeline
def my_pipeline(
    string: str = 'Hey!',
    integer: int = 100,
    floating_pt: float = 0.1,
    boolean: bool = False,
    dictionary: Dict = {'key': 'value'},
    array: List = [1, 2, 3],
):
    python_comp(
        string='howdy',
        integer=integer,
        array=[4, 5, 6],
    )
    container_comp(
        string=string,
        integer=20,
        dictionary={'other key': 'other val'},
        boolean=boolean,
    )

输出参数

对于 Python 组件和管道，输出参数通过返回注释表示：

from kfp import dsl

@dsl.component
def my_comp() -> int:
    return 1

@dsl.pipeline
def my_pipeline() -> int:
    task = my_comp()
    return task.output

对于容器组件，输出参数使用 dsl.OutputPath 注解进行指示：

from kfp import dsl

@dsl.container_component
def my_comp(int_path: dsl.OutputPath(int)):
    return dsl.ContainerSpec(
        image='alpine',
        command=[
            'sh', '-c', f"""mkdir -p $(dirname {int_path})\
                            && echo 1 > {int_path}"""
        ])

@dsl.pipeline
def my_pipeline() -> int:
    task = my_comp()
    return task.outputs['int_path']

有关如何使用 dsl.OutputPath 的更多信息，请参见 容器组件：创建组件输出

多个输出参数

您可以使用一个 typing.NamedTuple 指定多个命名输出参数。您可以通过 .outputs[''] 在 PipelineTask 上访问命名输出:

from kfp import dsl
from typing import NamedTuple

@dsl.component
def my_comp() -> NamedTuple('outputs', a=int, b=str):
    outputs = NamedTuple('outputs', a=int, b=str)
    return outputs(1, 'hello')

@dsl.pipeline
def my_pipeline() -> NamedTuple('pipeline_outputs', c=int, d=str):
    task = my_comp()
    pipeline_outputs = NamedTuple('pipeline_outputs', c=int, d=str)
    return pipeline_outputs(task.outputs['a'], task.outputs['b'])

反馈

此页面有帮助吗？


Last modified January 24, 2025: 管道用户指南数据处理参数中的错字 (#3975) (59c9e6c)