Merge branch 'main' into batch_image_model

notebooks/community/ml_ops/stage6/get_started_with_custom_image_model_batch.ipynb

@@ -82,9 +82,11 @@
         "The steps performed include:\n",
         "\n",
         "- Download a pretrained image classification model from TensorFlow Hub.\n",
+        "- Upload the TensorFlow Hub model as a `Vertex AI Model` resource.\n",
+        "- Make batch prediction with raw (uncompressed) image data to the `Model` resource, in JSONL format.\n",
         "- Create a serving function to receive compressed image data, and output decomopressed preprocessed data for the model input.\n",
         "- Upload the TensorFlow Hub model and serving function as a `Vertex AI Model` resource.\n",
-        "- Make batch prediction to the `Model` resource, in JSONL format.\n",
+        "- Make batch prediction with compressed image data to the `Model` resource, in File-List format.\n",
         "\n",
         "There is one key difference between using batch prediction and using online prediction:\n",
         "\n",
@@ -697,7 +699,7 @@
       "source": [
         "### Upload the TensorFlow Hub model to a `Vertex AI Model` resource\n",
         "\n",
-        "Finally, you upload the model artifacts from the TFHub model and serving function into a `Vertex AI Model` resource using the method `upload()`, with the following parameters:\n",
+        "Finally, you upload the model artifacts from the TFHub model into a `Vertex AI Model` resource using the method `upload()`, with the following parameters:\n",
         "\n",
         "- `display_name`: A human readable name for the `Model` resource.\n",
         "- `artifact_uri`: The Cloud Storage location of the model package.\n",
@@ -1035,6 +1037,385 @@
         "            break"
       ]
     },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "12ed21da6f1f"
+      },
+      "source": [
+        "#### Delete the batch prediction job\n",
+        "\n",
+        "You can delete your `Vertex AI Batch Prediction` job with the `delete()` method."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "286a90d9b6e7"
+      },
+      "outputs": [],
+      "source": [
+        "batch_prediction_job.delete()"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "6cdd298348a1"
+      },
+      "source": [
+        "#### Delete the model\n",
+        "\n",
+        "You can delete your `Vertex AI Model` resource with the `delete()` method."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "4c9484c41c39"
+      },
+      "outputs": [],
+      "source": [
+        "model.delete()"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "68f6562b12cb"
+      },
+      "source": [
+        "## Image models with serving functions\n",
+        "\n",
+        "Previously, your model server took input as a 3-dimensional array. Image models typically take a compressed image and use a serving function fused to the model to decompress the compressed image into a 3-dimensional array, and other preprocesing -- such as normalizing the pixel values.\n",
+        "\n",
+        "Next, you upload your custom image model as a `Vertex AI Model` resource with a serving function. During upload, you define a serving function to convert data to the format your model expects. If you send encoded data to Vertex AI, your serving function ensures that the data is decoded on the model server before it is passed as input to your model.\n",
+        "\n",
+        "### How does the serving function work\n",
+        "\n",
+        "When you send a request to an online prediction server, the request is received by a HTTP server. The HTTP server extracts the prediction request from the HTTP request content body. The extracted prediction request is forwarded to the serving function. For Google pre-built prediction containers, the request content is passed to the serving function as a `tf.string`.\n",
+        "\n",
+        "The serving function consists of two parts:\n",
+        "\n",
+        "- `preprocessing function`:\n",
+        "  - Converts the input (`tf.string`) to the input shape and data type of the underlying model (dynamic graph).\n",
+        "  - Performs the same preprocessing of the data that was done during training the underlying model -- e.g., normalizing, scaling, etc.\n",
+        "- `post-processing function`:\n",
+        "  - Converts the model output to format expected by the receiving application -- e.q., compresses the output.\n",
+        "  - Packages the output for the the receiving application -- e.g., add headings, make JSON object, etc.\n",
+        "\n",
+        "Both the preprocessing and post-processing functions are converted to static graphs which are fused to the model. The output from the underlying model is passed to the post-processing function. The post-processing function passes the converted/packaged output back to the HTTP server. The HTTP server returns the output as the HTTP response content.\n",
+        "\n",
+        "One consideration you need to consider when building serving functions for TF.Keras models is that they run as static graphs. That means, you cannot use TF graph operations that require a dynamic graph. If you do, you will get an error during the compile of the serving function which will indicate that you are using an EagerTensor which is not supported."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "serving_function_image:post"
+      },
+      "source": [
+        "### Serving function for image data\n",
+        "\n",
+        "#### Preprocessing\n",
+        "\n",
+        "To pass images to the prediction service, you encode the compressed (e.g., JPEG) image bytes into base 64 -- which makes the content safe from modification while transmitting binary data over the network. Since this deployed model expects input data as raw (uncompressed) bytes, you need to ensure that the base 64 encoded data gets converted back to raw bytes, and then preprocessed to match the model input requirements, before it is passed as input to the deployed model.\n",
+        "\n",
+        "To resolve this, you define a serving function (`serving_fn`) and attach it to the model as a preprocessing step. Add a `@tf.function` decorator so the serving function is fused to the underlying model (instead of upstream on a CPU).\n",
+        "\n",
+        "When you send a prediction or explanation request, the content of the request is base 64 decoded into a Tensorflow string (`tf.string`), which is passed to the serving function (`serving_fn`). The serving function preprocesses the `tf.string` into raw (uncompressed) numpy bytes (`preprocess_fn`) to match the input requirements of the model:\n",
+        "\n",
+        "- `io.decode_jpeg`- Decompresses the JPG image which is returned as a Tensorflow tensor with three channels (RGB).\n",
+        "- `image.convert_image_dtype` - Changes integer pixel values to float 32, and rescales pixel data between 0 and 1.\n",
+        "- `image.resize` - Resizes the image to match the input shape for the model.\n",
+        "\n",
+        "At this point, the data can be passed to the model (`m_call`), via a concrete function. The serving function is a static graph, while the model is a dynamic graph. The concrete function performs the tasks of marshalling the input data from the serving function to the model, and marshalling the prediction result from the model back to the serving function."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "serving_function_image"
+      },
+      "outputs": [],
+      "source": [
+        "CONCRETE_INPUT = \"numpy_inputs\"\n",
+        "\n",
+        "\n",
+        "def _preprocess(bytes_input):\n",
+        "    decoded = tf.io.decode_jpeg(bytes_input, channels=3)\n",
+        "    decoded = tf.image.convert_image_dtype(decoded, tf.float32)\n",
+        "    resized = tf.image.resize(decoded, size=(224, 224))\n",
+        "    return resized\n",
+        "\n",
+        "\n",
+        "@tf.function(input_signature=[tf.TensorSpec([None], tf.string)])\n",
+        "def preprocess_fn(bytes_inputs):\n",
+        "    decoded_images = tf.map_fn(\n",
+        "        _preprocess, bytes_inputs, dtype=tf.float32, back_prop=False\n",
+        "    )\n",
+        "    return {\n",
+        "        CONCRETE_INPUT: decoded_images\n",
+        "    }  # User needs to make sure the key matches model's input\n",
+        "\n",
+        "\n",
+        "@tf.function(input_signature=[tf.TensorSpec([None], tf.string)])\n",
+        "def serving_fn(bytes_inputs):\n",
+        "    images = preprocess_fn(bytes_inputs)\n",
+        "    prob = m_call(**images)\n",
+        "    return prob\n",
+        "\n",
+        "\n",
+        "m_call = tf.function(tfhub_model.call).get_concrete_function(\n",
+        "    [tf.TensorSpec(shape=[None, 224, 224, 3], dtype=tf.float32, name=CONCRETE_INPUT)]\n",
+        ")\n",
+        "\n",
+        "tf.saved_model.save(tfhub_model, MODEL_DIR, signatures={\"serving_default\": serving_fn})"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "serving_function_signature:image"
+      },
+      "source": [
+        "## Get the serving function signature\n",
+        "\n",
+        "You can get the signatures of your model's input and output layers by reloading the model into memory, and querying it for the signatures corresponding to each layer.\n",
+        "\n",
+        "For your purpose, you need the signature of the serving function. Why? Well, when we send our data for prediction as a HTTP request packet, the image data is base64 encoded, and our TF.Keras model takes numpy input. Your serving function will do the conversion from base64 to a numpy array.\n",
+        "\n",
+        "When making a prediction request, you need to route the request to the serving function instead of the model, so you need to know the input layer name of the serving function -- which you will use later when you make a prediction request."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "serving_function_signature:image"
+      },
+      "outputs": [],
+      "source": [
+        "loaded = tf.saved_model.load(MODEL_DIR)\n",
+        "\n",
+        "serving_input = list(\n",
+        "    loaded.signatures[\"serving_default\"].structured_input_signature[1].keys()\n",
+        ")[0]\n",
+        "print(\"Serving function input:\", serving_input)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "e8ce91147c93"
+      },
+      "source": [
+        "### Upload the TensorFlow Hub model to a `Vertex AI Model` resource\n",
+        "\n",
+        "Finally, you upload the model artifacts from the TFHub model and serving function into a `Vertex AI Model` resource using the method `upload()`, with the following parameters:\n",
+        "\n",
+        "- `display_name`: A human readable name for the `Model` resource.\n",
+        "- `artifact_uri`: The Cloud Storage location of the model package.\n",
+        "- `serving_container_image_uri`: The serving container image.\n",
+        "\n",
+        "Uploading a model into a Vertex AI Model resource returns a long running operation, since it may take a few moments. \n",
+        "\n",
+        "*Note:* When you upload the model artifacts to a `Vertex AI Model` resource, you specify the corresponding deployment container image."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "ad61e1429512"
+      },
+      "outputs": [],
+      "source": [
+        "model = aip.Model.upload(\n",
+        "    display_name=\"resnet_\" + UUID,\n",
+        "    artifact_uri=MODEL_DIR,\n",
+        "    serving_container_image_uri=DEPLOY_IMAGE,\n",
+        ")\n",
+        "\n",
+        "print(model)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "make_prediction"
+      },
+      "source": [
+        "### Make a batch prediction request\n",
+        "\n",
+        "Previously, you formatted the batch prediction instances using JSONL as 3-dimensional arrays. This time, you format the batch prediction instances as a file-list, where each line in the file is a Cloud Storage location of a compressed image."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "b1e29665076f"
+      },
+      "source": [
+        "#### Prepare data for batch prediction\n",
+        "\n",
+        "BLAH\n",
+        "\n",
+        "Before you can run the data through batch prediction, you need to save the data into one of a few possible formats.\n",
+        "\n",
+        "For this tutorial, use JSONL as it's compatible with the 3-dimensional list that each image is currently represented in. To do this:\n",
+        "\n",
+        "1. In a file, write each instance as JSON on its own line.\n",
+        "2. Upload this file to Cloud Storage.\n",
+        "\n",
+        "For more details on batch prediction input formats: https://cloud.google.com/vertex-ai/docs/predictions/batch-predictions#batch_request_input"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "becaaf02edde"
+      },
+      "outputs": [],
+      "source": [
+        "import tensorflow.python.ops.numpy_ops.np_config as np_config\n",
+        "\n",
+        "np_config.enable_numpy_behavior()\n",
+        "\n",
+        "(x_train, y_train), (x_test, y_test) = tf.keras.datasets.cifar10.load_data()\n",
+        "\n",
+        "n = 1\n",
+        "for image in x_test[:10]:\n",
+        "    c_image = tf.io.encode_jpeg(image)\n",
+        "    with tf.io.gfile.GFile(BUCKET_URI + f\"/images/image{n}.jpg\", \"wb\") as f:\n",
+        "        f.write(c_image.numpy())\n",
+        "    n += 1\n",
+        "\n",
+        "BATCH_PREDICTION_INSTANCES_FILE = \"batch_prediction_instances.txt\"\n",
+        "\n",
+        "BATCH_PREDICTION_GCS_SOURCE = (\n",
+        "    BUCKET_URI + \"/batch_prediction_instances/\" + BATCH_PREDICTION_INSTANCES_FILE\n",
+        ")\n",
+        "\n",
+        "with tf.io.gfile.GFile(BATCH_PREDICTION_GCS_SOURCE, \"w\") as f:\n",
+        "    for n in range(1, 11):\n",
+        "        f.write(BUCKET_URI + f\"/images/image{n}.jpg\\n\")"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "send_prediction_request:image"
+      },
+      "source": [
+        "### Send the prediction request\n",
+        "\n",
+        "BLAH\n",
+        "\n",
+        "To make a batch prediction request, call the model object's `batch_predict` method with the following parameters: \n",
+        "- `instances_format`: The format of the batch prediction request file: \"jsonl\", \"csv\", \"bigquery\", \"tf-record\", \"tf-record-gzip\" or \"file-list\"\n",
+        "- `prediction_format`: The format of the batch prediction response file: \"jsonl\", \"csv\", \"bigquery\", \"tf-record\", \"tf-record-gzip\" or \"file-list\"\n",
+        "- `job_display_name`: The human readable name for the prediction job.\n",
+        " - `gcs_source`: A list of one or more Cloud Storage paths to your batch prediction requests.\n",
+        "- `gcs_destination_prefix`: The Cloud Storage path that the service will write the predictions to.\n",
+        "- `model_parameters`: Additional filtering parameters for serving prediction results.\n",
+        "- `machine_type`: The type of machine to use for training.\n",
+        "- `accelerator_type`: The hardware accelerator type.\n",
+        "- `accelerator_count`: The number of accelerators to attach to a worker replica.\n",
+        "- `starting_replica_count`: The number of compute instances to initially provision.\n",
+        "- `max_replica_count`: The maximum number of compute instances to scale to. In this tutorial, only one instance is provisioned.\n",
+        "\n",
+        "### Compute instance scaling\n",
+        "\n",
+        "You can specify a single instance (or node) to process your batch prediction request. This tutorial uses a single node, so the variables `MIN_NODES` and `MAX_NODES` are both set to `1`.\n",
+        "\n",
+        "If you want to use multiple nodes to process your batch prediction request, set `MAX_NODES` to the maximum number of nodes you want to use. Vertex AI autoscales the number of nodes used to serve your predictions, up to the maximum number you set. Refer to the [pricing page](https://cloud.google.com/vertex-ai/pricing#prediction-prices) to understand the costs of autoscaling with multiple nodes.\n"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "1cf1076178fc"
+      },
+      "outputs": [],
+      "source": [
+        "MIN_NODES = 1\n",
+        "MAX_NODES = 1\n",
+        "\n",
+        "# The name of the job\n",
+        "BATCH_PREDICTION_JOB_NAME = \"cifar10_batch-\" + UUID\n",
+        "\n",
+        "# Folder in the bucket to write results to\n",
+        "DESTINATION_FOLDER = \"batch_prediction_results2\"\n",
+        "\n",
+        "# The Cloud Storage bucket to upload results to\n",
+        "BATCH_PREDICTION_GCS_DEST_PREFIX = BUCKET_URI + \"/\" + DESTINATION_FOLDER\n",
+        "\n",
+        "# Make SDK batch_predict method call\n",
+        "batch_prediction_job = model.batch_predict(\n",
+        "    instances_format=\"file-list\",\n",
+        "    predictions_format=\"jsonl\",\n",
+        "    job_display_name=BATCH_PREDICTION_JOB_NAME,\n",
+        "    gcs_source=BATCH_PREDICTION_GCS_SOURCE,\n",
+        "    gcs_destination_prefix=BATCH_PREDICTION_GCS_DEST_PREFIX,\n",
+        "    model_parameters=None,\n",
+        "    machine_type=DEPLOY_COMPUTE,\n",
+        "    accelerator_type=DEPLOY_GPU,\n",
+        "    accelerator_count=DEPLOY_NGPU,\n",
+        "    starting_replica_count=MIN_NODES,\n",
+        "    max_replica_count=MAX_NODES,\n",
+        "    sync=True,\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "get_batch_prediction:mbsdk,custom,icn"
+      },
+      "source": [
+        "### Get the predictions\n",
+        "\n",
+        "Next, get the results from the completed batch prediction job.\n",
+        "\n",
+        "The results are written to the Cloud Storage output bucket you specified in the batch prediction request. You call the method iter_outputs() to get a list of each Cloud Storage file generated with the results. Each file contains one or more prediction requests in a JSON format:\n",
+        "\n",
+        "- `instance`: The prediction request.\n",
+        "- `prediction`: The prediction response."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "get_batch_prediction:mbsdk,custom,icn"
+      },
+      "outputs": [],
+      "source": [
+        "import json\n",
+        "\n",
+        "bp_iter_outputs = batch_prediction_job.iter_outputs()\n",
+        "\n",
+        "prediction_results = list()\n",
+        "for blob in bp_iter_outputs:\n",
+        "    if blob.name.split(\"/\")[-1].startswith(\"prediction\"):\n",
+        "        prediction_results.append(blob.name)\n",
+        "\n",
+        "tags = list()\n",
+        "for prediction_result in prediction_results:\n",
+        "    gfile_name = f\"gs://{bp_iter_outputs.bucket.name}/{prediction_result}\"\n",
+        "    with tf.io.gfile.GFile(name=gfile_name, mode=\"r\") as gfile:\n",
+        "        for line in gfile.readlines():\n",
+        "            line = json.loads(line)\n",
+        "            print(line)\n",
+        "            break"
+      ]
+    },
     {
       "cell_type": "markdown",
       "metadata": {