批量预测作业非阻塞

Question

我正在使用 python API 运行 Vertex AI 批量预​​测。 我使用的功能来自谷歌云文档：

def create_batch_prediction_job_dedicated_resources_sample(
    key_path,
    project: str,
    location: str,
    model_display_name: str,
    job_display_name: str,
    gcs_source: Union[str, Sequence[str]],
    gcs_destination: str,
    machine_type: str = "n1-standard-2",
    sync: bool = True,
):
    credentials = service_account.Credentials.from_service_account_file(
    key_path)

# Initilaize an aiplatfrom object
 aiplatform.init(project=project, location=location, credentials=credentials)

# Get a list of Models by Model name
 models = aiplatform.Model.list(filter=f'display_name="{model_display_name}"')
 model_resource_name = models[0].resource_name

# Get the model
 my_model = aiplatform.Model(model_resource_name)

 batch_prediction_job = my_model.batch_predict(
    job_display_name=job_display_name,
    gcs_source=gcs_source,
    gcs_destination_prefix=gcs_destination,
    machine_type=machine_type,
    sync=sync,
)

 #batch_prediction_job.wait_for_resource_creation()
 batch_prediction_job.wait()

 print(batch_prediction_job.display_name)
 print(batch_prediction_job.resource_name)
 print(batch_prediction_job.state)
 return batch_prediction_job

datetime_today = datetime.datetime.now()
model_display_name = 'test_model'
key_path = 'vertex_key.json'
project = 'my_project'
location = 'asia-south1'
job_display_name = 'batch_prediction_' + str(datetime_today)
model_name = '1234'
gcs_source = 'gs://my_bucket/Cleaned_Data/user_item_pairs.jsonl'
gcs_destination = 'gs://my_bucket/prediction'

create_batch_prediction_job_dedicated_resources_sample(key_path,project,location,model_display_name,job_display_name,
                                                      gcs_source,gcs_destination)

输出：

92 current state:
JobState.JOB_STATE_RUNNING
INFO:google.cloud.aiplatform.jobs:BatchPredictionJob projects/my_project/locations/asia-south1/batchPredictionJobs/37737350127597649

上面的输出每隔几秒钟就会在终端上一遍又一遍地打印出来。

我遇到的问题是调用此函数的 python 程序一直运行，直到它被强制停止。我尝试了batch_prediction_job.wait() 和batch_prediction_job.wait_for_resource_creation()，结果相同。

如何在不等待它完成的情况下启动 batch_prediction_job 并在创建作业后立即终止程序？

Answer 1

我在cmets上给你错误的指令，更改参数sync=False，函数执行后应该返回。

这个函数调用应该是同步的（等待管道运行完成后再终止）还是异步的（立即返回）

sync=False

def create_batch_prediction_job_dedicated_resources_sample(
# ...
    sync: bool = False,
):

更新 - 添加更多细节：

Check here my notebook code 我在其中对其进行了测试及其工作： 您必须更改 sync=False AND 删除/注释以下打印行：

#batch_prediction_job.wait()
#print(batch_prediction_job.display_name)
#print(batch_prediction_job.resource_name)
#print(batch_prediction_job.state)

您的代码已编辑：

def create_batch_prediction_job_dedicated_resources_sample(
    key_path,
    project: str,
    location: str,
    model_display_name: str,
    job_display_name: str,
    gcs_source: Union[str, Sequence[str]],
    gcs_destination: str,
    machine_type: str = "n1-standard-2",
    sync: bool = False,
):
    credentials = service_account.Credentials.from_service_account_file(key_path)

# Initilaize an aiplatfrom object
 aiplatform.init(project=project, location=location, credentials=credentials)

# Get a list of Models by Model name
 models = aiplatform.Model.list(filter=f'display_name="{model_display_name}"')
 model_resource_name = models[0].resource_name

# Get the model
 my_model = aiplatform.Model(model_resource_name)

 batch_prediction_job = my_model.batch_predict(
    job_display_name=job_display_name,
    gcs_source=gcs_source,
    gcs_destination_prefix=gcs_destination,
    machine_type=machine_type,
    sync=sync,
)


 return batch_prediction_job

datetime_today = datetime.datetime.now()
model_display_name = 'test_model'
key_path = 'vertex_key.json'
project = '<my_project_name>'
location = 'asia-south1'
job_display_name = 'batch_prediction_' + str(datetime_today)
model_name = '1234'
gcs_source = 'gs://<my_bucket_name>/Cleaned_Data/user_item_pairs.jsonl'
gcs_destination = 'gs://<my_bucket_name>/prediction'

create_batch_prediction_job_dedicated_resources_sample(key_path,
                         project,location,
                         model_display_name,
                         job_display_name,
                         gcs_source,
                         gcs_destination,
                         sync=False,)

结果 sync=False: 结果 sync=True: