Как использовать kfp Artifact со sklearn?

Question

Я пытаюсь разработать собственный конвейер с компонентами конвейера kubeflow (kfp) внутри Vertex AI (Google Cloud Platform). Этапы конвейера:

1. читать данные из большой таблицы запросов
2. создать панд
3. использовать DataFrame обучить модель K-средних
4. развернуть модель в конечной точке

Вот код шага 2. Пришлось использовать Output[Artifact] как выход, потому что 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'

Можно ли преобразовать Артефакт в numpy array или Dataframe?

Answer 1

Я нашел решение, используя следующие коды. Теперь я могу обучить модель шагу 3 с выводом шага 2.

Шаг 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)