Kubeflow Pipeline编译报错：无法访问Worker多输出属性

当你的组件使用OutputPath定义多个输出时，不能直接通过ContainerOp对象的属性（如merge_promo_sales_nl.output_data_hd）访问输出，因为这类输出会被存储在ContainerOp的outputs字典中，需通过键名获取。以下是具体解决方法：

1. 确保组件函数正确定义多输出 #

先确认你的组件函数用@component装饰器正确声明输出参数：

from kfp import dsl
from kfp.dsl import component, Input, OutputPath, Dataset

@component
def merge_promo_sales(
    input_data: Input[Dataset],
    output_data_hd: OutputPath("Dataset"),
    output_data_shop: OutputPath("Dataset")
):
    # 你的业务逻辑：读取输入、拆分数据集并写入指定路径
    import pandas as pd
    df = pd.read_csv(input_data.path)
    df_hd = df[df['type'] == 'hd']
    df_shop = df[df['type'] == 'shop']
    
    df_hd.to_csv(output_data_hd, index=False)
    df_shop.to_csv(output_data_shop, index=False)

2. 在Pipeline中正确获取输出 #

调用组件后，通过outputs字典结合输出参数名来获取对应输出，再传递给后续组件：

@dsl.pipeline(name="promo-sales-merge-pipeline")
def pipeline(input_dataset: Input[Dataset]):
    # 调用拆分组件
    merge_op = merge_promo_sales(input_data=input_dataset)
    
    # 正确获取两个输出
    hd_dataset = merge_op.outputs['output_data_hd']
    shop_dataset = merge_op.outputs['output_data_shop']
    
    # 示例：将输出传递给后续处理组件
    # process_hd_op = process_hd_data(input_data=hd_dataset)
    # process_shop_op = process_shop_data(input_data=shop_dataset)

可选：改用Output[Dataset]实现属性式访问 #

如果希望用属性（如merge_op.output_data_hd）直接访问输出，可以改用Output[Dataset]类型，并为输出指定名称：

@component
def merge_promo_sales(
    input_data: Input[Dataset]
) -> (Output[Dataset](name="output_data_hd"), Output[Dataset](name="output_data_shop")):
    import pandas as pd
    df = pd.read_csv(input_data.path)
    df_hd = df[df['type'] == 'hd']
    df_shop = df[df['type'] == 'shop']
    
    return df_hd, df_shop

此时在Pipeline中即可直接通过属性访问：

@dsl.pipeline(name="promo-sales-merge-pipeline")
def pipeline(input_dataset: Input[Dataset]):
    merge_op = merge_promo_sales(input_data=input_dataset)
    
    # 直接通过属性获取输出
    hd_dataset = merge_op.output_data_hd
    shop_dataset = merge_op.output_data_shop

内容的提问来源于stack exchange，提问作者JordanBH