在Kubeflow管道中，如何向轻量级python组件发送元素列表？

Question

我试图将元素列表作为PipelineParameter发送到轻量级组件。

这里是一个再现问题的示例。以下是功能：

def my_func(my_list: list) -> bool:
    print(f'my_list is {my_list}')
    print(f'my_list is of type {type(my_list)}')
    print(f'elem 0 is {my_list[0]}')
    print(f'elem 1 is {my_list[1]}')
    return True

如果我用这个来执行：

test_data = ['abc', 'def']
my_func(test_data)

它的行为与预期相同：

my_list is ['abc', 'def']
my_list is of type <class 'list'>
elem 0 is abc
elem 1 is def

但是如果我把它封装在一个op中并设置一个管道：

import kfp

my_op = kfp.components.func_to_container_op(my_func)

@kfp.dsl.pipeline()
def my_pipeline(my_list: kfp.dsl.PipelineParam = kfp.dsl.PipelineParam('my_list', param_type=kfp.dsl.types.List())):
    my_op(my_list)

kfp.compiler.Compiler().compile(my_pipeline, 'my_pipeline.zip')

然后运行管道：

client = kfp.Client()
experiment = client.create_experiment('Default')
client.run_pipeline(experiment.id, 'my job', 'my_pipeline.zip', params={'my_list': test_data})

然后，似乎在某一点上，我的列表被转换成一个字符串！

my_list is ['abc', 'def']
my_list is of type <class 'str'>
elem 0 is [
elem 1 is '

Answer 1

下面是我发现的一个解决办法，将参数序列化为json字符串。不确定这是不是最好的方法..。

裸功能变成：

def my_func(json_arg_str: str) -> bool:
    import json
    args = json.loads(json_arg_str)
    my_list = args['my_list']
    print(f'my_list is {my_list}')
    print(f'my_list is of type {type(my_list)}')
    print(f'elem 0 is {my_list[0]}')
    print(f'elem 1 is {my_list[1]}')
    return True

只要您将args作为json字符串而不是列表传递，它仍然有效：

test_data ={“my_list”：“abc”，"def"}‘my_func(test_data)

产生预期结果：

my_list is ['abc', 'def']
my_list is of type <class 'list'>
elem 0 is abc
elem 1 is def

现在，管道被更改为接受str，而不是kfp.dsl.types.List类型的PipelineParam。

import kfp 

my_op = kfp.components.func_to_container_op(my_func)

@kfp.dsl.pipeline()
def my_pipeline(json_arg_str: str):
    my_op(json_arg_str)

kfp.compiler.Compiler().compile(my_pipeline, 'my_pipeline.zip')

当像这样执行时：

client = kfp.Client()
experiment = client.create_experiment('Default')
client.run_pipeline(experiment.id, 'my job', 'my_pipeline.zip', params={'json_arg_str': test_data})

产生同样的结果：

my_list is ['abc', 'def']
my_list is of type <class 'list'>
elem 0 is abc
elem 1 is def

尽管如此，我还是觉得这个解决办法很烦人。那么kfp.dsl.types.List的意义是什么，如果不允许PipelineParam是一个列表的话？

Answer 2

目前，最好的选择似乎是序列化参数。有一个与此相关的问题：https://github.com/kubeflow/pipelines/issues/1901