Get batch predictions from a custom trained model

This page shows you how to get batch predictions from your custom trained models using the Google Cloud console or the Vertex AI API.

To make a batch prediction request, you specify an input source and an output location, either Cloud Storage or BigQuery, where Vertex AI stores the batch prediction results.

Limitations and requirements

Consider the following limitations and requirements when getting batch predictions:

• To minimize processing time, your input and output locations must be in the same region or multi-region. For example, if your input is in us-central1, your output can be in us-central1 or US, but not europe-west4. To learn more, see Cloud Storage locations and BigQuery locations.
• Your input and output must also be in the same region or multi-region as your model.
• Model Garden models are not supported.
• BigQuery ML models aren't custom trained models. However, the information in this page can be used to get batch predictions from a BigQuery ML model under the following conditions:
  • The BigQuery ML model must be registered with Vertex AI Model Registry.
  • To use a BigQuery table as input, you must set InstanceConfig.instanceType to "object" using the Vertex AI API.

Input data requirements

The input for batch requests specifies the items to send to your model for prediction. We support the following input formats:

[1, 2, 3, 4]
[5, 6, 7, 8]

{"instances": [ [1, 2, 3, 4], [5, 6, 7, 8] ]}

{"instances": [
{ "data": [1, 2, 3, 4] },
{ "data": [5, 6, 7, 8] } ]}

{ "values": [1, 2, 3, 4], "key": 1 }
{ "values": [5, 6, 7, 8], "key": 2 }

{"instances": [
  { "values": [1, 2, 3, 4], "key": 1 },
  { "values": [5, 6, 7, 8], "key": 2 }
]}

{ "data": { "values": [1, 2, 3, 4], "key": 1 } }
{ "data": { "values": [5, 6, 7, 8], "key": 2 } }

{"instances": [
  { "data": { "values": [1, 2, 3, 4], "key": 1 } },
  { "data": { "values": [5, 6, 7, 8], "key": 2 } }
]}

{"instances": [
{ "b64": "b64EncodedASCIIString" },
{ "b64": "b64EncodedASCIIString" } ]}

{"instances": [ { "data": {"b64": "b64EncodedASCIIString" } }, { "data": {"b64": "b64EncodedASCIIString" } }
]}

"input1","input2","input3"
0.1,1.2,"cat1"
4.0,5.0,"cat2"

{"instances": [ [0.1,1.2,"cat1"], [4.0,5.0,"cat2"] ]}

{"instances": [
{ "data": [0.1,1.2,"cat1"] },
{ "data": [4.0,5.0,"cat2"] } ]}

gs://path/to/image/image1.jpg
gs://path/to/image/image2.jpg

{ "instances": [
{ "b64": "b64EncodedASCIIString" },
{ "b64": "b64EncodedASCIIString" } ]}

{ "instances": [ { "data": { "b64": "b64EncodedASCIIString" } }, { "data": { "b64": "b64EncodedASCIIString" } }
]}

{"instances": [ [1.0,3.0,"cat1"], [2.0,4.0,"cat2"] ]}

{"instances": [
{ "data": [1.0,3.0,"cat1"] },
{ "data": [2.0,4.0,"cat2"] } ]}

Partition data

Batch prediction uses MapReduce to shard the input to each replica. To make use of the MapReduce features, the input should be partitionable.

Vertex AI automatically partitions BigQuery, file list, and JSON lines input.

Vertex AI does not automatically partition CSV files because they are not naturally partition-friendly. Rows in CSV files are not self-descriptive, typed, and may contain newlines. We recommend against using CSV input for throughput-sensitive applications.

For TFRecord input, make sure you manually partition the data by splitting the instances into smaller files and passing the files to the job with a wildcard (for example, gs://my-bucket/*.tfrecord). The number of files should be at least the number of replicas specified.

Filter and transform input data

You can filter and transform your batch input by specifying instanceConfig in your BatchPredictionJob request.

Filtering lets you either exclude certain fields that are in the input data from your prediction request, or include only a subset of fields from the input data in your prediction request, without having to do any custom pre-processing or post-processing in the prediction container. This is useful when your input data file has extra columns that the model doesn't need, such as keys or additional data.

Transforming lets you send the instances to your prediction container in either a JSON array or object format. See instanceType for more information.

For example, if your input table contains the following:

and you specify the following instanceConfig:

{
  "name": "batchJob1",
  ...
  "instanceConfig": {
    "excludedFields":["customerId"]
    "instanceType":"object"
  }
}

Next, the instances in your prediction request are sent as JSON objects, and the customerId column is excluded:

{"col1":1,"col2":2}
{"col1":5,"col2":6}

Note that specifying the following instanceConfig would yield the same result:

{
  "name": "batchJob1",
  ...
  "instanceConfig": {
    "includedFields": ["col1","col2"]
    "instanceType":"object"
  }
}

For a demonstration on how to use feature filters, see the Custom model batch prediction with feature filtering notebook.

Request a batch prediction

For batch prediction requests, you can use the Google Cloud console or the Vertex AI API. Depending on the number of input items that you've submitted, a batch prediction task can take some time to complete.

When you request a batch prediction, the prediction container runs as the user-provided custom service account. The read-write operations, such as reading the prediction instances from the data source or writing the prediction results, are done using the Vertex AI service agent, which by default has access to BigQuery and Cloud Storage.

POST https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/batchPredictionJobs

{
  "displayName": "BATCH_JOB_NAME",
  "model": "projects/PROJECT_ID/locations/LOCATION_ID/models/MODEL_ID",
  "inputConfig": {
    "instancesFormat": "INPUT_FORMAT",
    "gcsSource": {
      "uris": ["INPUT_URI"],
    },
  },
  "outputConfig": {
    "predictionsFormat": "jsonl",
    "gcsDestination": {
      "outputUriPrefix": "OUTPUT_DIRECTORY",
    },
  },
  "dedicatedResources" : {
    "machineSpec" : {
      "machineType": MACHINE_TYPE
    },
    "startingReplicaCount": STARTING_REPLICA_COUNT
  },
  "manualBatchTuningParameters": {
    "batch_size": BATCH_SIZE,
  }
}

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/batchPredictionJobs"

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://LOCATION_ID-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID/batchPredictionJobs" | Select-Object -Expand Content

{
  "name": "projects/PROJECT_NUMBER/locations/LOCATION_ID/batchPredictionJobs/BATCH_JOB_ID",
  "displayName": "BATCH_JOB_NAME 202005291958",
  "model": "projects/PROJECT_ID/locations/LOCATION_ID/models/MODEL_ID",
  "inputConfig": {
    "instancesFormat": "jsonl",
    "gcsSource": {
      "uris": [
        "INPUT_URI"
      ]
    }
  },
  "outputConfig": {
    "predictionsFormat": "jsonl",
    "gcsDestination": {
      "outputUriPrefix": "OUTPUT_DIRECTORY"
    }
  },
  "state": "JOB_STATE_PENDING",
  "createTime": "2020-05-30T02:58:44.341643Z",
  "updateTime": "2020-05-30T02:58:44.341643Z",
}

import com.google.cloud.aiplatform.util.ValueConverter;
import com.google.cloud.aiplatform.v1.AcceleratorType;
import com.google.cloud.aiplatform.v1.BatchDedicatedResources;
import com.google.cloud.aiplatform.v1.BatchPredictionJob;
import com.google.cloud.aiplatform.v1.GcsDestination;
import com.google.cloud.aiplatform.v1.GcsSource;
import com.google.cloud.aiplatform.v1.JobServiceClient;
import com.google.cloud.aiplatform.v1.JobServiceSettings;
import com.google.cloud.aiplatform.v1.LocationName;
import com.google.cloud.aiplatform.v1.MachineSpec;
import com.google.cloud.aiplatform.v1.ModelName;
import com.google.protobuf.Value;
import java.io.IOException;

public class CreateBatchPredictionJobSample {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String project = "PROJECT";
    String displayName = "DISPLAY_NAME";
    String modelName = "MODEL_NAME";
    String instancesFormat = "INSTANCES_FORMAT";
    String gcsSourceUri = "GCS_SOURCE_URI";
    String predictionsFormat = "PREDICTIONS_FORMAT";
    String gcsDestinationOutputUriPrefix = "GCS_DESTINATION_OUTPUT_URI_PREFIX";
    createBatchPredictionJobSample(
        project,
        displayName,
        modelName,
        instancesFormat,
        gcsSourceUri,
        predictionsFormat,
        gcsDestinationOutputUriPrefix);
  }

  static void createBatchPredictionJobSample(
      String project,
      String displayName,
      String model,
      String instancesFormat,
      String gcsSourceUri,
      String predictionsFormat,
      String gcsDestinationOutputUriPrefix)
      throws IOException {
    JobServiceSettings settings =
        JobServiceSettings.newBuilder()
            .setEndpoint("us-central1-aiplatform.googleapis.com:443")
            .build();
    String location = "us-central1";

    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests. After completing all of your requests, call
    // the "close" method on the client to safely clean up any remaining background resources.
    try (JobServiceClient client = JobServiceClient.create(settings)) {

      // Passing in an empty Value object for model parameters
      Value modelParameters = ValueConverter.EMPTY_VALUE;

      GcsSource gcsSource = GcsSource.newBuilder().addUris(gcsSourceUri).build();
      BatchPredictionJob.InputConfig inputConfig =
          BatchPredictionJob.InputConfig.newBuilder()
              .setInstancesFormat(instancesFormat)
              .setGcsSource(gcsSource)
              .build();
      GcsDestination gcsDestination =
          GcsDestination.newBuilder().setOutputUriPrefix(gcsDestinationOutputUriPrefix).build();
      BatchPredictionJob.OutputConfig outputConfig =
          BatchPredictionJob.OutputConfig.newBuilder()
              .setPredictionsFormat(predictionsFormat)
              .setGcsDestination(gcsDestination)
              .build();
      MachineSpec machineSpec =
          MachineSpec.newBuilder()
              .setMachineType("n1-standard-2")
              .setAcceleratorType(AcceleratorType.NVIDIA_TESLA_T4)
              .setAcceleratorCount(1)
              .build();
      BatchDedicatedResources dedicatedResources =
          BatchDedicatedResources.newBuilder()
              .setMachineSpec(machineSpec)
              .setStartingReplicaCount(1)
              .setMaxReplicaCount(1)
              .build();
      String modelName = ModelName.of(project, location, model).toString();
      BatchPredictionJob batchPredictionJob =
          BatchPredictionJob.newBuilder()
              .setDisplayName(displayName)
              .setModel(modelName)
              .setModelParameters(modelParameters)
              .setInputConfig(inputConfig)
              .setOutputConfig(outputConfig)
              .setDedicatedResources(dedicatedResources)
              .build();
      LocationName parent = LocationName.of(project, location);
      BatchPredictionJob response = client.createBatchPredictionJob(parent, batchPredictionJob);
      System.out.format("response: %s\n", response);
      System.out.format("\tName: %s\n", response.getName());
    }
  }
}

def create_batch_prediction_job_dedicated_resources_sample(
    project: str,
    location: str,
    model_resource_name: str,
    job_display_name: str,
    gcs_source: Union[str, Sequence[str]],
    gcs_destination: str,
    instances_format: str = "jsonl",
    machine_type: str = "n1-standard-2",
    accelerator_count: int = 1,
    accelerator_type: Union[str, aiplatform_v1.AcceleratorType] = "NVIDIA_TESLA_K80",
    starting_replica_count: int = 1,
    max_replica_count: int = 1,
    sync: bool = True,
):
    aiplatform.init(project=project, location=location)

    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,
        instances_format=instances_format,
        machine_type=machine_type,
        accelerator_count=accelerator_count,
        accelerator_type=accelerator_type,
        starting_replica_count=starting_replica_count,
        max_replica_count=max_replica_count,
        sync=sync,
    )

    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

Choose machine type and replica count

Scaling horizontally by increasing the number of replicas improves throughput more linearly and predictably than by using larger machine types.

In general, we recommend that you specify the smallest machine type possible for your job and increase the number of replicas.

For cost-effectiveness, we recommend that you choose the replica count such that your batch prediction job runs for at least 10 minutes. This is because you are billed per replica node hour, which includes the approximately 5 minutes it takes for each replica to start up. It isn't cost-effective to process for only a few seconds and then shut down.

As general guidance, for thousands of instances, we recommend a starting_replica_count in the tens. For millions of instances, we recommend a starting_replica_count in the hundreds. You can also use the following formula to estimate the number of replicas:

N / (T * (60 / Tb))

Where:

• N: The number of batches in the job. For example, 1 million instances / 100 batch size = 10,000 batches.
• T: The expected time for the batch prediction job. For example, 10 minutes.
• Tb: The time in seconds it takes for a replica to process a single batch. For example, 1 second per batch on a 2-core machine type.

In our example, 10,000 batches / (10 minutes * (60 / 1s)) rounds up to 17 replicas.

Unlike online prediction, batch prediction jobs don't autoscale. Because all of the input data is known up front, the system partitions the data to each replica when the job starts. The system uses the starting_replica_count parameter. The max_replica_count parameter is ignored.

These recommendations are all approximate guidelines. They don't necessarily give optimal throughput for every model. They don't provide exact predictions of processing time and cost. And they don't necessarily capture the best cost-versus-throughput tradeoffs for each scenario. Use them as a reasonable starting point and adjust them as necessary. To measure characteristics such as throughput for your model, run the Finding ideal machine type notebook.

For GPU- or TPU-accelerated machines

Follow the preceding guidelines (which also apply to CPU-only models), with the following additional considerations:

• You might need more CPUs and GPUs (for example, for data preprocessing).
• GPU machine types take more time to startup (10 minutes), so you may want to target longer times (for example, at least 20 minutes instead of 10 minutes) for the batch prediction job so that a reasonable proportion of the time and cost is spent on generating predictions.

Retrieve batch prediction results

When a batch prediction task is complete, the output of the prediction is stored in the Cloud Storage bucket or BigQuery location that you specified in your request.

Example batch prediction result

The output folder contains a set of JSON Lines files.

The files are named {gcs_path}/prediction.results-{file_number}-of-{number_of_files_generated}. The number of files is nondeterministic, due to the distributed nature of batch prediction.

Each line in the file corresponds to an instance from the input and has the following key-value pairs:

• prediction: contains the value returned by prediction container.
• instance: For a file list, contains the Cloud Storage URI. For all other input formats, contains the value that was sent to the prediction container in the HTTP request body.

Example 1

If the HTTP request contains:

{
  "instances": [
    [1, 2, 3, 4],
    [5, 6, 7, 8]
]}

And the prediction container returns:

{
  "predictions": [
    [0.1,0.9],
    [0.7,0.3]
  ],
}

Then the JSON Lines output file is:

{ "instance": [1, 2, 3, 4], "prediction": [0.1,0.9]}
{ "instance": [5, 6, 7, 8], "prediction": [0.7,0.3]}

Example 2

If the HTTP request contains:

{
  "instances": [
    {"values": [1, 2, 3, 4], "key": 1},
    {"values": [5, 6, 7, 8], "key": 2}
]}

And the prediction container returns:

{
  "predictions": [
    {"result":1},
    {"result":0}
  ],
}

Then the JSON Lines output file is:

{ "instance": {"values": [1, 2, 3, 4], "key": 1}, "prediction": {"result":1}}
{ "instance": {"values": [5, 6, 7, 8], "key": 2}, "prediction": {"result":0}}

Use Explainable AI

We don't recommend running feature-based explanations on a large amount of data. This is because each input can potentially fan out to thousands of requests based on the set of possible feature values which may result in massively increased processing time and cost. In general, a small dataset is enough to understand feature importance.

Batch prediction does not support example-based explanations.

Notebooks

What's next

• Learn about Compute resources for prediction.