Vertex AI 模型批量预测、引用云存储上的现有模型和输入文件的问题

Question

我正在努力正确设置 Vertex AI 管道，该管道执行以下操作：

1. 从 API 读取数据并存储到 GCS 并作为批量预测的输入。
2. 获取现有模型（Vertex AI 上的视频分类）
3. 使用第 1 点的输入创建批量预测作业
   。正如所见，我对 Vertex Pipelines/Kubeflow 没有太多经验，因此我寻求帮助/建议，希望这只是一些初学者的错误。这是我用作管道的代码的要点

from google_cloud_pipeline_components import aiplatform as gcc_aip
from kfp.v2 import dsl

from kfp.v2.dsl import component
from kfp.v2.dsl import (
    Output,
    Artifact,
    Model,
)

PROJECT_ID = 'my-gcp-project'
BUCKET_NAME = "mybucket"
PIPELINE_ROOT = "{}/pipeline_root".format(BUCKET_NAME)


@component
def get_input_data() -> str:
    # getting data from API, save to Cloud Storage
    # return GS URI
    gcs_batch_input_path = 'gs://somebucket/file'
    return gcs_batch_input_path


@component(
    base_image="python:3.9",
    packages_to_install=['google-cloud-aiplatform==1.8.0']
)
def load_ml_model(project_id: str, model: Output[Artifact]):
    """Load existing Vertex model"""
    import google.cloud.aiplatform as aip

    model_id = '1234'
    model = aip.Model(model_name=model_id, project=project_id, location='us-central1')



@dsl.pipeline(
    name="batch-pipeline", pipeline_root=PIPELINE_ROOT,
)
def pipeline(gcp_project: str):
    input_data = get_input_data()
    ml_model = load_ml_model(gcp_project)

    gcc_aip.ModelBatchPredictOp(
        project=PROJECT_ID,
        job_display_name=f'test-prediction',
        model=ml_model.output,
        gcs_source_uris=[input_data.output],  # this doesn't work
        # gcs_source_uris=['gs://mybucket/output/'],  # hardcoded gs uri works
        gcs_destination_output_uri_prefix=f'gs://{PIPELINE_ROOT}/prediction_output/'
    )


if __name__ == '__main__':
    from kfp.v2 import compiler
    import google.cloud.aiplatform as aip
    pipeline_export_filepath = 'test-pipeline.json'
    compiler.Compiler().compile(pipeline_func=pipeline,
                                package_path=pipeline_export_filepath)
    # pipeline_params = {
    #     'gcp_project': PROJECT_ID,
    # }
    # job = aip.PipelineJob(
    #     display_name='test-pipeline',
    #     template_path=pipeline_export_filepath,
    #     pipeline_root=f'gs://{PIPELINE_ROOT}',
    #     project=PROJECT_ID,
    #     parameter_values=pipeline_params,
    # )

    # job.run()

运行管道时，它在运行批量预测时抛出此异常：
details = "List of found errors: 1.Field: batch_prediction_job.model; Message: Invalid Model resource name. 所以我不确定可能出了什么问题。我尝试在笔记本中加载模型（在组件之外）并且它正确返回。

我遇到的第二个问题是引用 GCS URI 作为从组件到批处理作业输入的输出。

   input_data = get_input_data2()
   gcc_aip.ModelBatchPredictOp(
        project=PROJECT_ID,
        job_display_name=f'test-prediction',
        model=ml_model.output,
        gcs_source_uris=[input_data.output],  # this doesn't work
        # gcs_source_uris=['gs://mybucket/output/'],  # hardcoded gs uri works
        gcs_destination_output_uri_prefix=f'gs://{PIPELINE_ROOT}/prediction_output/'
    )

在编译过程中，我收到以下异常TypeError: Object of type PipelineParam is not JSON serializable，尽管我认为这可能是 ModelBatchPredictOp 组件的问题。

再次感谢任何帮助/建议，我从昨天开始处理这个问题，所以也许我错过了一些明显的东西。

我正在使用的库：

google-cloud-aiplatform==1.8.0  
google-cloud-pipeline-components==0.2.0  
kfp==1.8.10  
kfp-pipeline-spec==0.1.13  
kfp-server-api==1.7.1

更新 评论后，进行一些研究和调整，以参考模型：

google-cloud-aiplatform==1.8.0  
google-cloud-pipeline-components==0.2.0  
kfp==1.8.10  
kfp-pipeline-spec==0.1.13  
kfp-server-api==1.7.1

然后我可以按预期使用它：

@component
def load_ml_model(project_id: str, model: Output[Artifact]):
    region = 'us-central1'
    model_id = '1234'
    model_uid = f'projects/{project_id}/locations/{region}/models/{model_id}'
    model.uri = model_uid
    model.metadata['resourceName'] = model_uid

关于 GCS 路径的更新 2 ，解决方法是在组件外部定义路径并将其作为输入参数传递，例如（缩写）：

batch_predict_op = gcc_aip.ModelBatchPredictOp(
        project=gcp_project,
        job_display_name=f'batch-prediction-test',
        model=ml_model.outputs['model'],
        gcs_source_uris=[input_batch_gcs_path],
gcs_destination_output_uri_prefix=f'gs://{BUCKET_NAME}/prediction_output/test'
    )

get_input_data仍然不确定，为什么当我从组件返回 GCS 路径时它不起作用/编译

Answer 1

我很高兴您解决了大部分主要问题并找到了模型声明的解决方法。

根据您input.output对 的观察gcs_source_uris，其背后的原因是函数/类返回值的方式。如果您深入研究该类/方法，google_cloud_pipeline_components 您会发现它实现了一个结构，该结构允许您使用.outputs所调用函数的返回值。

如果您查看管道组件之一的实现，您会发现它从convert_method_to_component函数返回一个输出数组。因此，为了在您的自定义类/函数中实现该功能，您的函数应该返回一个可以作为属性调用的值。下面是它的基本实现。

class CustomClass():
     def __init__(self):
       self.return_val = {'path':'custompath','desc':'a desc'}
      
     @property
     def output(self):
       return self.return_val 

hello = CustomClass()
print(hello.output['path'])

如果您想深入了解它，可以访问以下页面：

• Convert_method_to_component，这是实现convert_method_to_component

• Properties，Python 中属性的基础知识。