如何将 kfp Artifact 与 sklearn 一起使用？

Posted 2023-03-29

【中文标题】如何将 kfp Artifact 与 sklearn 一起使用？

【英文标题】：How to use kfp Artifact with sklearn?

【发布时间】：2021-12-26 20:09:12

【问题描述】：

我正在尝试使用 Vertex AI（谷歌云平台）内的 kubeflow 管道 (kfp) 组件开发一个自定义管道。流水线的步骤是：

从大查询表中读取数据 创建熊猫DataFrame 使用 DataFrame 训练 K-Means 模型 将模型部署到端点

这里是第 2 步的代码。我不得不使用Output[Artifact] 作为输出，因为我发现here 的pd.DataFrame 类型不起作用。

@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
    
    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 = client.list_rows(table).to_dataframe()

这里是第3步的代码：

@component(base_image="python:3.9", packages_to_install=['sklearn'])
def kmeans_training(
        dataset: Input[Artifact],
        model: Output[Model],
        num_clusters: int,
):
    from sklearn.cluster import KMeans
    model = KMeans(num_clusters, random_state=220417)
    model.fit(dataset)

管道的运行由于以下错误而停止：

TypeError: float() argument must be a string or a number, not 'Artifact'

是否可以将 Artifact 转换为 numpy array 或 Dataframe？

【问题讨论】：

您可以查看以下链接 [1] 以检查如何将工件转换为数据框，或者您可以在构建管道时使用 Google Cloud 库[2] 将工件导入一个云存储桶，您只需更改 pipeline_root_path .[1]forum.qiime2.org/t/load-distancematrix-artifact-to-dataframe/… 。 [2]cloud.google.com/vertex-ai/docs/pipelines/….

【参考方案1】：

我找到了使用以下代码的解决方案。现在我可以使用步骤 2 的输出来训练步骤 3 中的模型。

第 2 步：

@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[Dataset],
):
    
    from google.cloud import bigquery
    
    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)

    train = client.list_rows(table).to_dataframe()
    
    train.to_csv(df.path)

第 3 步：

@component(base_image="python:3.9", packages_to_install=['sklearn','pandas','joblib'])
def kmeans_training(
        dataset: Input[Dataset],
        model_artifact: Output[Model],
        num_clusters: int,
):
    from sklearn.cluster import KMeans
    import pandas as pd
    from joblib import dump
    
    data = pd.read_csv(dataset.path)
    
    model = KMeans(num_clusters, random_state=220417)
    model.fit(data)
    
    dump(model, model_artifact.path)