尝试使用文件在 Kubeflow 组件之间传递数据时出现问题答案

【问题标题】：Issue when trying to pass data between Kubeflow components using files尝试使用文件在 Kubeflow 组件之间传递数据时出现问题
【发布时间】：2021-04-25 13:41:27
【问题描述】：

我使用 python 函数制作了两个组件，并尝试使用文件在它们之间传递数据，但我无法这样做。我想计算总和，然后使用文件将答案发送给其他组件。下面是部分代码（代码在没有文件传递的情况下工作）。请帮忙。

# Define your components code as standalone python functions:======================
    def add(a: float, b: float, f: comp.OutputTextFile(float)) -> NamedTuple(
        'AddOutput',
        [
            ('out', comp.OutputTextFile(float))
        ]):
        '''Calculates sum of two arguments'''
        sum = a+b

        f.write(sum)

        from collections import namedtuple

        addOutput = namedtuple(
            'AddOutput',
            ['out'])
        return addOutput(f)  # the metrics will be uploaded to the cloud


    def multiply(c:float, d:float, f: comp.InputTextFile(float) ):
        '''Calculates the product'''
        product = c * d

        print(f.read())


add_op = comp.func_to_container_op(add, output_component_file='add_component.yaml')
    product_op = comp.create_component_from_func(multiply, 
output_component_file='multiple_component.yaml')


@dsl.pipeline(
      name='Addition-pipeline',
      description='An example pipeline that performs addition calculations.'
    )
    def my_pipeline(a, b='7', c='4', d='1'):

        add_op = pl_comp_list[0]
        product_op = pl_comp_list[1]

        first_add_task = add_op(a, 4)
        second_add_task = product_op(c, d, first_add_task.outputs['out'])

【问题讨论】：

标签： python kubeflow kubeflow-pipelines

【解决方案1】：

这是我测试过的管道的一个稍微简化的版本，它可以工作。传递给OutputTextFile 和InputTextFile 的类类型并不重要。它将被读写为str。所以这是你应该改变的：

写信给OutputTextFile时：从float to str转换sum_
从InputTextFile 读取时：从str to float 转换f.read() 值

import kfp
from kfp import dsl
from kfp import components as comp


def add(a: float, b: float, f: comp.OutputTextFile()):
    '''Calculates sum of two arguments'''
    sum_ = a + b
    f.write(str(sum_)) # cast to str
    return sum_


def multiply(c: float, d: float, f: comp.InputTextFile()):
    '''Calculates the product'''
    in_ = float(f.read()) # cast to float
    product = c * d * in_
    print(product)
    return product


add_op = comp.func_to_container_op(add,
                                   output_component_file='add_component.yaml')
product_op = comp.create_component_from_func(
    multiply, output_component_file='multiple_component.yaml')


@dsl.pipeline(
    name='Addition-pipeline',
    description='An example pipeline that performs addition calculations.')
def my_pipeline(a, b='7', c='4', d='1'):

    first_add_task = add_op(a, b)
    second_add_task = product_op(c, d, first_add_task.output)


if __name__ == "__main__":
    compiled_name = __file__ + ".yaml"
    kfp.compiler.Compiler().compile(my_pipeline, compiled_name)

【讨论】：

【解决方案2】：

('out', comp.OutputTextFile(float))

这不是真的有效。 OutputTextFile 注解（和其他类似注解）只能在函数参数中使用。函数返回值仅适用于您希望作为值（而不是文件）输出的输出。

由于您已经拥有f: comp.OutputTextFile(float)，您可以在其中完全删除函数返回值。然后将f 输出传递给下游组件：product_op(c, d, first_add_task.outputs['f'])。

【讨论】：