提问者:小点点

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


我正在努力正确设置顶点AI管道,它执行以下操作:

  1. 从API读取数据并存储到GCS并作为批量预测的输入。
  2. 获取现有模型(VertexAI上的视频分类)
  3. 使用第1点的输入创建批处理预测作业。
    可以看出,我对顶点管道/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()

当运行管道时,它在运行Batch预测时抛出此异常:
详细信息="发现的错误列表: 1.字段:batch_prediction_job.model;消息:无效的模型资源名称。所以我不确定会出什么问题。我试图在笔记本中加载模型(在组件之外),它正确返回。

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

   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: PipelineParam类型的对象不是JSON可序列化的,尽管我认为这可能是ModelBatchPredicOp组件的问题。

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

我正在使用的库:

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

UPDATE经过评论、一些研究和调整,对于引用模型,这是可行的:

@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

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

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'
    )

UPDATE 2关于GCS路径,解决方法是在组件外部定义路径并将其作为输入参数传递,例如(缩写):

@dsl.pipeline(
    name="my-pipeline",
    pipeline_root=PIPELINE_ROOT,
)
def pipeline(
        gcp_project: str,
        region: str,
        bucket: str
):
    ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
    
    gcs_prediction_input_path = f'gs://{BUCKET_NAME}/prediction_input/video_batch_prediction_input_{ts}.jsonl'
    batch_input_data_op = get_input_data(gcs_prediction_input_path)  # this loads input data to GCS path

    batch_predict_op = gcc_aip.ModelBatchPredictOp(
        project=gcp_project,
        model=training_job_run_op.outputs["model"],
        job_display_name='batch-prediction',
        # gcs_source_uris=[batch_input_data_op.output],
        gcs_source_uris=[gcs_prediction_input_path],
        gcs_destination_output_uri_prefix=f'gs://{BUCKET_NAME}/prediction_output/',
    ).after(batch_input_data_op)  # we need to add 'after' so it runs after input data is prepared since get_input_data doesn't returns anything

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


共1个答案

匿名用户

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

对于你对gcs_source_urisinput.输出观察,它背后的原因是因为函数/类返回值的方式。如果你深入挖掘google_cloud_pipeline_components的类/方法,你会发现它实现了一个结构,允许你从调用的函数的返回值中使用。输出

如果你去管道的一个组件的实现,你会发现它从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的实现

    属性,python中属性的基础知识。