如何在sklearn训练场景中使用Kubeflow Pipeline(KFP)的Artifact

报错的核心原因是Kubeflow Pipeline（KFP）中的Input[Artifact]、Output[Artifact]是平台用于跨组件传递数据的封装容器对象，本身并不存储实际的数据集内容，你之前的代码存在两个核心错误：

1. create_dataframe组件中直接将生成的DataFrame赋值给Output[Artifact]类型的df变量，覆盖了KFP传入的Artifact对象，实际没有把数据写入到跨组件共享的存储路径
2. kmeans_training组件中直接将Input[Artifact]类型的dataset对象传入fit方法，模型无法识别该封装对象

你提到的将Artifact转换为DataFrame或numpy数组的思路完全可行，按以下方式修改代码即可解决问题：

1. 修改create_dataframe组件代码 #

将生成的DataFrame序列化后写入Artifact对应的存储路径，推荐用parquet格式保留原生数据类型：

@component(base_image="python:3.9", packages_to_install=["google-cloud-bigquery","pandas","pyarrow"])
def create_dataframe(
    project: str,
    region: str,
    destination_dataset: str,
    destination_table_name: str,
    df: Output[Artifact],
):
    
    from google.cloud import bigquery
    import pandas as pd
    
    client = bigquery.Client(project=project, location=region)
    dataset_ref = bigquery.DatasetReference(project, destination_dataset)
    table_ref = dataset_ref.table(destination_table_name)
    table = client.get_table(table_ref)

    # 不直接覆盖df变量，将DataFrame写入Artifact对应的共享存储路径
    df_data = client.list_rows(table).to_dataframe()
    df_data.to_parquet(df.path, index=False)

df.path是KFP自动生成的跨组件共享存储路径（默认存储到GCS对应Bucket），下游组件可以通过相同路径读取数据。

2. 修改kmeans_training组件代码 #

先从Input[Artifact]的路径读取序列化的数据，转换为DataFrame或numpy数组后再传入模型训练，同时训练好的模型也要序列化写入Output[Model]的路径：

@component(base_image="python:3.9", packages_to_install=['scikit-learn', 'pandas', 'pyarrow', 'joblib'])
def kmeans_training(
        dataset: Input[Artifact],
        model: Output[Model],
        num_clusters: int,
):
    import pandas as pd
    import joblib
    from sklearn.cluster import KMeans
    
    # 读取Artifact路径中的数据，转换为DataFrame，也可按需转为numpy数组：df = df.values
    df = pd.read_parquet(dataset.path)
    
    kmeans_model = KMeans(n_clusters=num_clusters, random_state=220417)
    kmeans_model.fit(df)
    
    # 将训练好的模型写入Model Artifact的存储路径
    joblib.dump(kmeans_model, model.path)

注意：sklearn的官方安装包名是scikit-learn不是sklearn，原代码的依赖声明存在错误，会导致组件运行失败，上述代码已经修正。

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