在组件之间传递少量数据

参数有助于在组件之间传递少量数据，并且当组件创建的数据不表示机器学习 Artifact（例如模型、数据集或更复杂的数据类型）时也很有用。

使用内置 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 指定多个命名输出参数。您可以使用 PipelineTask 上的 .outputs['<output-key>'] 访问命名输出。

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'])

反馈

此页面有帮助吗？


上次修改时间 2025年1月24日：管道用户指南数据处理参数中的拼写错误 (#3975) (59c9e6c)