This is an automated email from the ASF dual-hosted git repository.
riteshghorse pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 65344d46700 [Docs] Add Vertex AI Feature Store enrichment notebook
(#30689)
65344d46700 is described below
commit 65344d467000c960ab406e70b3c059cc315e4c4f
Author: Ritesh Ghorse <[email protected]>
AuthorDate: Mon Mar 25 16:05:00 2024 -0400
[Docs] Add Vertex AI Feature Store enrichment notebook (#30689)
* add vertex ai enrichment notebook
* address review comments, add to readme
* Update examples/notebooks/beam-ml/vertex_ai_feature_store_enrichment.ipynb
Co-authored-by: Rebecca Szper <[email protected]>
* Update examples/notebooks/beam-ml/vertex_ai_feature_store_enrichment.ipynb
Co-authored-by: Rebecca Szper <[email protected]>
* update content from review
---------
Co-authored-by: Rebecca Szper <[email protected]>
---
examples/notebooks/beam-ml/README.md | 5 +
.../vertex_ai_feature_store_enrichment.ipynb | 2502 ++++++++++++++++++++
2 files changed, 2507 insertions(+)
diff --git a/examples/notebooks/beam-ml/README.md
b/examples/notebooks/beam-ml/README.md
index f1c19747fc7..8f723c2c914 100644
--- a/examples/notebooks/beam-ml/README.md
+++ b/examples/notebooks/beam-ml/README.md
@@ -56,6 +56,11 @@ This section contains the following example notebooks.
* [Use MLTransform to scale
data](https://github.com/apache/beam/blob/master/examples/notebooks/beam-ml/data_preprocessing/scale_data.ipynb)
* [Preprocessing with the Apache Beam DataFrames
API](https://github.com/apache/beam/blob/master/examples/notebooks/beam-ml/dataframe_api_preprocessing.ipynb)
+### Data enrichment
+
+* [Use Bigtable to enrich
data](https://github.com/apache/beam/blob/master/examples/notebooks/beam-ml/bigtable_enrichment_transform.ipynb)
+* [Use Vertex AI Feature Store for feature
enrichment](https://github.com/apache/beam/blob/master/examples/notebooks/beam-ml/vertex_ai_feature_store_enrichment.ipynb)
+
### Prediction and inference with pretrained models
* [Apache Beam RunInference for
PyTorch](https://github.com/apache/beam/blob/master/examples/notebooks/beam-ml/run_inference_pytorch.ipynb)
diff --git
a/examples/notebooks/beam-ml/vertex_ai_feature_store_enrichment.ipynb
b/examples/notebooks/beam-ml/vertex_ai_feature_store_enrichment.ipynb
new file mode 100644
index 00000000000..c8ae558a1ba
--- /dev/null
+++ b/examples/notebooks/beam-ml/vertex_ai_feature_store_enrichment.ipynb
@@ -0,0 +1,2502 @@
+{
+ "cells": [
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "cellView": "form",
+ "id": "fFjof1NgAJwu"
+ },
+ "outputs": [],
+ "source": [
+ "# @title ###### Licensed to the Apache Software Foundation (ASF),
Version 2.0 (the \"License\")\n",
+ "\n",
+ "# Licensed to the Apache Software Foundation (ASF) under one\n",
+ "# or more contributor license agreements. See the NOTICE file\n",
+ "# distributed with this work for additional information\n",
+ "# regarding copyright ownership. The ASF licenses this file\n",
+ "# to you under the Apache License, Version 2.0 (the\n",
+ "# \"License\"); you may not use this file except in compliance\n",
+ "# with the License. You may obtain a copy of the License at\n",
+ "#\n",
+ "# http://www.apache.org/licenses/LICENSE-2.0\n";,
+ "#\n",
+ "# Unless required by applicable law or agreed to in writing,\n",
+ "# software distributed under the License is distributed on an\n",
+ "# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n",
+ "# KIND, either express or implied. See the License for the\n",
+ "# specific language governing permissions and limitations\n",
+ "# under the License"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "A8xNRyZMW1yK"
+ },
+ "source": [
+ "# Use Apache Beam and Vertex AI Feature Store to enrich data\n",
+ "\n",
+ "<table align=\"left\">\n",
+ " <td>\n",
+ " <a target=\"_blank\"
href=\"https://colab.research.google.com/github/apache/beam/blob/master/examples/notebooks/beam-ml/vertex_ai_feature_store_enrichment.ipynb\";><img
src=\"https://raw.githubusercontent.com/google/or-tools/main/tools/colab_32px.png\";
/>Run in Google Colab</a>\n",
+ " </td>\n",
+ " <td>\n",
+ " <a target=\"_blank\"
href=\"https://github.com/apache/beam/blob/master/examples/notebooks/beam-ml/vertex_ai_feature_store_enrichment.ipynb\";><img
src=\"https://raw.githubusercontent.com/google/or-tools/main/tools/github_32px.png\";
/>View source on GitHub</a>\n",
+ " </td>\n",
+ "</table>\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "HrCtxslBGK8Z"
+ },
+ "source": [
+ "This notebook shows how to enrich data by using the Apache Beam
[enrichment
transform](https://beam.apache.org/documentation/transforms/python/elementwise/enrichment/)
with [Vertex AI Feature Store](https://cloud.google.com/vertex-ai/docs). The
enrichment transform is a turnkey transform in Apache Beam that lets you enrich
data using a key-value lookup. This transform has the following features:\n",
+ "\n",
+ "- The transform has a built-in Apache Beam handler that interacts
with Vertex AI to get precomputed feature values.\n",
+ "- The transform uses client-side throttling to manage rate limiting
the requests.\n",
+ "- Optionally, you can configure a Redis cache to improve
efficiency.\n",
+ "\n",
+ "As of Apache Beam SDK version 2.55.0, [online feature
serving](https://cloud.google.com/vertex-ai/docs/featurestore/latest/overview#online_serving)
through Bigtable online serving and the Vertex AI Feature Store (legacy)
method is supported. This notebook demonstrates how to use the Bigtable online
serving approach with the enrichment transform in an Apache Beam pipeline."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "ltn5zrBiGS9C"
+ },
+ "source": [
+ "This notebook demonstrates the following ecommerce product
recommendation use case based on the BigQuery public dataset [theLook
eCommerce](https://pantheon.corp.google.com/marketplace/product/bigquery-public-data/thelook-ecommerce):\n",
+ "\n",
+ "* Use a stream of online transactions from
[Pub/Sub](https://cloud.google.com/pubsub/docs/guides) that contains the
following fields: `product_id`, `user_id`, and `sale_price`.\n",
+ "* Deploy a pretrained model on Vertex AI based on the features
`product_id`, `user_id`, `sale_price`, `age`, `gender`, `state`, and
`country`.\n",
+ "* Precompute the feature values for the pretrained model, and store
the values in the Vertex AI Feature Store.\n",
+ "* Enrich the stream of transactions from Pub/Sub with feature values
from Vertex AI Feature Store by using the `Enrichment` transform.\n",
+ "* Send the enriched data to the Vertex AI model for online prediction
by using the `RunInference` transform, which predicts the product
recommendation for the user."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "gVCtGOKTHMm4"
+ },
+ "source": [
+ "## Before you begin\n",
+ "Set up your environment and download dependencies."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "YDHPlMjZRuY0"
+ },
+ "source": [
+ "### Install Apache Beam\n",
+ "To use the enrichment transform with the built-in Vertex AI handler,
install the Apache Beam SDK version 2.55.0 or later."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "collapsed": true,
+ "id": "jBakpNZnAhqk"
+ },
+ "outputs": [],
+ "source": [
+ "!pip install apache_beam[interactive,gcp]==2.55.0 --quiet\n",
+ "!pip install redis\n",
+ "\n",
+ "# Use TensorFlow 2.13.0, because it is the latest version that has
the prebuilt\n",
+ "# container image for Vertex AI model deployment.\n",
+ "# See
https://cloud.google.com/vertex-ai/docs/predictions/pre-built-containers#tensorflow\n";,
+ "!pip install tensorflow==2.13"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "SiJii48A2Rnb"
+ },
+ "outputs": [],
+ "source": [
+ "import json\n",
+ "import math\n",
+ "import os\n",
+ "import time\n",
+ "\n",
+ "from typing import Any\n",
+ "from typing import Dict\n",
+ "\n",
+ "import pandas as pd\n",
+ "from google.cloud import aiplatform\n",
+ "from google.cloud import pubsub_v1\n",
+ "from google.cloud import bigquery\n",
+ "from google.cloud import storage\n",
+ "from google.cloud.aiplatform_v1 import
FeatureOnlineStoreAdminServiceClient\n",
+ "from google.cloud.aiplatform_v1 import
FeatureRegistryServiceClient\n",
+ "from google.cloud.aiplatform_v1.types import feature_view as
feature_view_pb2\n",
+ "from google.cloud.aiplatform_v1.types import \\\n",
+ " feature_online_store as feature_online_store_pb2\n",
+ "from google.cloud.aiplatform_v1.types import \\\n",
+ " feature_online_store_admin_service as \\\n",
+ " feature_online_store_admin_service_pb2\n",
+ "\n",
+ "import apache_beam as beam\n",
+ "import tensorflow as tf\n",
+ "import apache_beam.runners.interactive.interactive_beam as ib\n",
+ "from apache_beam.ml.inference.base import RunInference\n",
+ "from apache_beam.ml.inference.vertex_ai_inference import
VertexAIModelHandlerJSON\n",
+ "from apache_beam.options import pipeline_options\n",
+ "from apache_beam.runners.interactive.interactive_runner import
InteractiveRunner\n",
+ "from apache_beam.transforms.enrichment import Enrichment\n",
+ "from
apache_beam.transforms.enrichment_handlers.vertex_ai_feature_store import
VertexAIFeatureStoreEnrichmentHandler\n",
+ "from tensorflow import keras\n",
+ "from tensorflow.keras import layers"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "X80jy3FqHjK4"
+ },
+ "source": [
+ "### Authenticate with Google Cloud\n",
+ "This notebook reads data from Pub/Sub and Vertex AI. To use your
Google Cloud account, authenticate this notebook."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "Kz9sccyGBqz3"
+ },
+ "outputs": [],
+ "source": [
+ "from google.colab import auth\n",
+ "auth.authenticate_user()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "nAmGgUMt48o9"
+ },
+ "source": [
+ "Replace `<PROJECT_ID>` and `<LOCATION>` with the appropriate values
for your Google Cloud account."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "wEXucyi2liij"
+ },
+ "outputs": [],
+ "source": [
+ "PROJECT_ID = \"<PROJECT_ID>\"\n",
+ "LOCATION = \"<LOCATION>\""
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "RpqZFfFfA_Dt"
+ },
+ "source": [
+ "### Train and deploy the model to Vertex AI\n",
+ "\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "8cUpV7mkB_xE"
+ },
+ "source": [
+ "Fetch the training data from the BigQuery public dataset
[thelook-ecommerce](https://pantheon.corp.google.com/marketplace/product/bigquery-public-data/thelook-ecommerce)."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";,
+ "height": 206
+ },
+ "id": "TpxDHGObBEsj",
+ "outputId": "4f7afe32-a72b-40d3-b9ae-cc999ad104b8"
+ },
+ "outputs": [
+ {
+ "data": {
+ "application/vnd.google.colaboratory.intrinsic+json": {
+ "type": "dataframe",
+ "variable_name": "train_data"
+ },
+ "text/html": [
+ "\n",
+ " <div id=\"df-d4074964-50d7-4425-a22a-70b0e5045dd1\"
class=\"colab-df-container\">\n",
+ " <div>\n",
+ "<style scoped>\n",
+ " .dataframe tbody tr th:only-of-type {\n",
+ " vertical-align: middle;\n",
+ " }\n",
+ "\n",
+ " .dataframe tbody tr th {\n",
+ " vertical-align: top;\n",
+ " }\n",
+ "\n",
+ " .dataframe thead th {\n",
+ " text-align: right;\n",
+ " }\n",
+ "</style>\n",
+ "<table border=\"1\" class=\"dataframe\">\n",
+ " <thead>\n",
+ " <tr style=\"text-align: right;\">\n",
+ " <th></th>\n",
+ " <th>user_id</th>\n",
+ " <th>product_id</th>\n",
+ " <th>sale_price</th>\n",
+ " <th>age</th>\n",
+ " <th>gender</th>\n",
+ " <th>state</th>\n",
+ " <th>country</th>\n",
+ " </tr>\n",
+ " </thead>\n",
+ " <tbody>\n",
+ " <tr>\n",
+ " <th>0</th>\n",
+ " <td>68717</td>\n",
+ " <td>14235</td>\n",
+ " <td>0.02</td>\n",
+ " <td>43</td>\n",
+ " <td>f</td>\n",
+ " <td>sachsen</td>\n",
+ " <td>germany</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>1</th>\n",
+ " <td>59866</td>\n",
+ " <td>28700</td>\n",
+ " <td>1.50</td>\n",
+ " <td>17</td>\n",
+ " <td>m</td>\n",
+ " <td>chongqing</td>\n",
+ " <td>china</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>2</th>\n",
+ " <td>38322</td>\n",
+ " <td>14202</td>\n",
+ " <td>1.50</td>\n",
+ " <td>47</td>\n",
+ " <td>f</td>\n",
+ " <td>missouri</td>\n",
+ " <td>united states</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>3</th>\n",
+ " <td>7839</td>\n",
+ " <td>28700</td>\n",
+ " <td>1.50</td>\n",
+ " <td>64</td>\n",
+ " <td>m</td>\n",
+ " <td>mato grosso</td>\n",
+ " <td>brasil</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>4</th>\n",
+ " <td>40877</td>\n",
+ " <td>28700</td>\n",
+ " <td>1.50</td>\n",
+ " <td>68</td>\n",
+ " <td>m</td>\n",
+ " <td>sergipe</td>\n",
+ " <td>brasil</td>\n",
+ " </tr>\n",
+ " </tbody>\n",
+ "</table>\n",
+ "</div>\n",
+ " <div class=\"colab-df-buttons\">\n",
+ "\n",
+ " <div class=\"colab-df-container\">\n",
+ " <button class=\"colab-df-convert\"
onclick=\"convertToInteractive('df-d4074964-50d7-4425-a22a-70b0e5045dd1')\"\n",
+ " title=\"Convert this dataframe to an interactive
table.\"\n",
+ " style=\"display:none;\">\n",
+ "\n",
+ " <svg xmlns=\"http://www.w3.org/2000/svg\"; height=\"24px\"
viewBox=\"0 -960 960 960\">\n",
+ " <path
d=\"M120-120v-720h720v720H120Zm60-500h600v-160H180v160Zm220
220h160v-160H400v160Zm0 220h160v-160H400v160ZM180-400h160v-160H180v160Zm440
0h160v-160H620v160ZM180-180h160v-160H180v160Zm440 0h160v-160H620v160Z\"/>\n",
+ " </svg>\n",
+ " </button>\n",
+ "\n",
+ " <style>\n",
+ " .colab-df-container {\n",
+ " display:flex;\n",
+ " gap: 12px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-convert {\n",
+ " background-color: #E8F0FE;\n",
+ " border: none;\n",
+ " border-radius: 50%;\n",
+ " cursor: pointer;\n",
+ " display: none;\n",
+ " fill: #1967D2;\n",
+ " height: 32px;\n",
+ " padding: 0 0 0 0;\n",
+ " width: 32px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-convert:hover {\n",
+ " background-color: #E2EBFA;\n",
+ " box-shadow: 0px 1px 2px rgba(60, 64, 67, 0.3), 0px 1px
3px 1px rgba(60, 64, 67, 0.15);\n",
+ " fill: #174EA6;\n",
+ " }\n",
+ "\n",
+ " .colab-df-buttons div {\n",
+ " margin-bottom: 4px;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-convert {\n",
+ " background-color: #3B4455;\n",
+ " fill: #D2E3FC;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-convert:hover {\n",
+ " background-color: #434B5C;\n",
+ " box-shadow: 0px 1px 3px 1px rgba(0, 0, 0, 0.15);\n",
+ " filter: drop-shadow(0px 1px 2px rgba(0, 0, 0, 0.3));\n",
+ " fill: #FFFFFF;\n",
+ " }\n",
+ " </style>\n",
+ "\n",
+ " <script>\n",
+ " const buttonEl =\n",
+ "
document.querySelector('#df-d4074964-50d7-4425-a22a-70b0e5045dd1
button.colab-df-convert');\n",
+ " buttonEl.style.display =\n",
+ " google.colab.kernel.accessAllowed ? 'block' :
'none';\n",
+ "\n",
+ " async function convertToInteractive(key) {\n",
+ " const element =
document.querySelector('#df-d4074964-50d7-4425-a22a-70b0e5045dd1');\n",
+ " const dataTable =\n",
+ " await
google.colab.kernel.invokeFunction('convertToInteractive',\n",
+ " [key],
{});\n",
+ " if (!dataTable) return;\n",
+ "\n",
+ " const docLinkHtml = 'Like what you see? Visit the '
+\n",
+ " '<a target=\"_blank\"
href=https://colab.research.google.com/notebooks/data_table.ipynb>data table
notebook</a>'\n",
+ " + ' to learn more about interactive tables.';\n",
+ " element.innerHTML = '';\n",
+ " dataTable['output_type'] = 'display_data';\n",
+ " await google.colab.output.renderOutput(dataTable,
element);\n",
+ " const docLink = document.createElement('div');\n",
+ " docLink.innerHTML = docLinkHtml;\n",
+ " element.appendChild(docLink);\n",
+ " }\n",
+ " </script>\n",
+ " </div>\n",
+ "\n",
+ "\n",
+ "<div id=\"df-dcacc3b1-58b4-4ba0-96e1-2bbe81f01ffe\">\n",
+ " <button class=\"colab-df-quickchart\"
onclick=\"quickchart('df-dcacc3b1-58b4-4ba0-96e1-2bbe81f01ffe')\"\n",
+ " title=\"Suggest charts\"\n",
+ " style=\"display:none;\">\n",
+ "\n",
+ "<svg xmlns=\"http://www.w3.org/2000/svg\";
height=\"24px\"viewBox=\"0 0 24 24\"\n",
+ " width=\"24px\">\n",
+ " <g>\n",
+ " <path d=\"M19 3H5c-1.1 0-2 .9-2 2v14c0 1.1.9 2 2
2h14c1.1 0 2-.9 2-2V5c0-1.1-.9-2-2-2zM9 17H7v-7h2v7zm4 0h-2V7h2v10zm4
0h-2v-4h2v4z\"/>\n",
+ " </g>\n",
+ "</svg>\n",
+ " </button>\n",
+ "\n",
+ "<style>\n",
+ " .colab-df-quickchart {\n",
+ " --bg-color: #E8F0FE;\n",
+ " --fill-color: #1967D2;\n",
+ " --hover-bg-color: #E2EBFA;\n",
+ " --hover-fill-color: #174EA6;\n",
+ " --disabled-fill-color: #AAA;\n",
+ " --disabled-bg-color: #DDD;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-quickchart {\n",
+ " --bg-color: #3B4455;\n",
+ " --fill-color: #D2E3FC;\n",
+ " --hover-bg-color: #434B5C;\n",
+ " --hover-fill-color: #FFFFFF;\n",
+ " --disabled-bg-color: #3B4455;\n",
+ " --disabled-fill-color: #666;\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart {\n",
+ " background-color: var(--bg-color);\n",
+ " border: none;\n",
+ " border-radius: 50%;\n",
+ " cursor: pointer;\n",
+ " display: none;\n",
+ " fill: var(--fill-color);\n",
+ " height: 32px;\n",
+ " padding: 0;\n",
+ " width: 32px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart:hover {\n",
+ " background-color: var(--hover-bg-color);\n",
+ " box-shadow: 0 1px 2px rgba(60, 64, 67, 0.3), 0 1px 3px 1px
rgba(60, 64, 67, 0.15);\n",
+ " fill: var(--button-hover-fill-color);\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart-complete:disabled,\n",
+ " .colab-df-quickchart-complete:disabled:hover {\n",
+ " background-color: var(--disabled-bg-color);\n",
+ " fill: var(--disabled-fill-color);\n",
+ " box-shadow: none;\n",
+ " }\n",
+ "\n",
+ " .colab-df-spinner {\n",
+ " border: 2px solid var(--fill-color);\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " animation:\n",
+ " spin 1s steps(1) infinite;\n",
+ " }\n",
+ "\n",
+ " @keyframes spin {\n",
+ " 0% {\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " border-left-color: var(--fill-color);\n",
+ " }\n",
+ " 20% {\n",
+ " border-color: transparent;\n",
+ " border-left-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " }\n",
+ " 30% {\n",
+ " border-color: transparent;\n",
+ " border-left-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " border-right-color: var(--fill-color);\n",
+ " }\n",
+ " 40% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " }\n",
+ " 60% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " }\n",
+ " 80% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " }\n",
+ " 90% {\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " }\n",
+ " }\n",
+ "</style>\n",
+ "\n",
+ " <script>\n",
+ " async function quickchart(key) {\n",
+ " const quickchartButtonEl =\n",
+ " document.querySelector('#' + key + ' button');\n",
+ " quickchartButtonEl.disabled = true; // To prevent
multiple clicks.\n",
+ " quickchartButtonEl.classList.add('colab-df-spinner');\n",
+ " try {\n",
+ " const charts = await
google.colab.kernel.invokeFunction(\n",
+ " 'suggestCharts', [key], {});\n",
+ " } catch (error) {\n",
+ " console.error('Error during call to suggestCharts:',
error);\n",
+ " }\n",
+ "
quickchartButtonEl.classList.remove('colab-df-spinner');\n",
+ "
quickchartButtonEl.classList.add('colab-df-quickchart-complete');\n",
+ " }\n",
+ " (() => {\n",
+ " let quickchartButtonEl =\n",
+ "
document.querySelector('#df-dcacc3b1-58b4-4ba0-96e1-2bbe81f01ffe button');\n",
+ " quickchartButtonEl.style.display =\n",
+ " google.colab.kernel.accessAllowed ? 'block' :
'none';\n",
+ " })();\n",
+ " </script>\n",
+ "</div>\n",
+ "\n",
+ " </div>\n",
+ " </div>\n"
+ ],
+ "text/plain": [
+ " user_id product_id sale_price age gender state
country\n",
+ "0 68717 14235 0.02 43 f sachsen
germany\n",
+ "1 59866 28700 1.50 17 m chongqing
china\n",
+ "2 38322 14202 1.50 47 f missouri
united states\n",
+ "3 7839 28700 1.50 64 m mato grosso
brasil\n",
+ "4 40877 28700 1.50 68 m sergipe
brasil"
+ ]
+ },
+ "execution_count": 4,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "train_data_query = \"\"\"\n",
+ "WITH\n",
+ " order_items AS (\n",
+ " SELECT cast(user_id as string) AS user_id,\n",
+ " product_id,\n",
+ " sale_price,\n",
+ " FROM `bigquery-public-data.thelook_ecommerce.order_items`),\n",
+ " users AS (\n",
+ " SELECT cast(id as string) AS user_id,\n",
+ " age,\n",
+ " lower(gender) as gender,\n",
+ " lower(state) as state,\n",
+ " lower(country) as country,\n",
+ " FROM `bigquery-public-data.thelook_ecommerce.users`)\n",
+ "SELECT *\n",
+ "FROM order_items\n",
+ "LEFT OUTER JOIN users\n",
+ "USING (user_id)\n",
+ "\"\"\"\n",
+ "\n",
+ "client = bigquery.Client(project=PROJECT_ID)\n",
+ "train_data =
client.query(train_data_query).result().to_dataframe()\n",
+ "train_data.head()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "OkYcJPC0THoV"
+ },
+ "source": [
+ "Create a prediction dataframe that contains the `product_id` to
recommend to the user. Preprocess the data for columns that contain the
categorical values."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";,
+ "height": 206
+ },
+ "id": "ej6jCkMF0B29",
+ "outputId": "44cfd7f1-0c7c-40a8-813f-02af86a6f788"
+ },
+ "outputs": [
+ {
+ "data": {
+ "application/vnd.google.colaboratory.intrinsic+json": {
+ "type": "dataframe",
+ "variable_name": "train_data"
+ },
+ "text/html": [
+ "\n",
+ " <div id=\"df-6c868129-b0f4-49db-b83e-5a2914f6cf1a\"
class=\"colab-df-container\">\n",
+ " <div>\n",
+ "<style scoped>\n",
+ " .dataframe tbody tr th:only-of-type {\n",
+ " vertical-align: middle;\n",
+ " }\n",
+ "\n",
+ " .dataframe tbody tr th {\n",
+ " vertical-align: top;\n",
+ " }\n",
+ "\n",
+ " .dataframe thead th {\n",
+ " text-align: right;\n",
+ " }\n",
+ "</style>\n",
+ "<table border=\"1\" class=\"dataframe\">\n",
+ " <thead>\n",
+ " <tr style=\"text-align: right;\">\n",
+ " <th></th>\n",
+ " <th>user_id</th>\n",
+ " <th>product_id</th>\n",
+ " <th>sale_price</th>\n",
+ " <th>age</th>\n",
+ " <th>gender</th>\n",
+ " <th>state</th>\n",
+ " <th>country</th>\n",
+ " </tr>\n",
+ " </thead>\n",
+ " <tbody>\n",
+ " <tr>\n",
+ " <th>0</th>\n",
+ " <td>68717</td>\n",
+ " <td>14235</td>\n",
+ " <td>0.02</td>\n",
+ " <td>43</td>\n",
+ " <td>0</td>\n",
+ " <td>0</td>\n",
+ " <td>0</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>1</th>\n",
+ " <td>59866</td>\n",
+ " <td>28700</td>\n",
+ " <td>1.50</td>\n",
+ " <td>17</td>\n",
+ " <td>1</td>\n",
+ " <td>1</td>\n",
+ " <td>1</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>2</th>\n",
+ " <td>38322</td>\n",
+ " <td>14202</td>\n",
+ " <td>1.50</td>\n",
+ " <td>47</td>\n",
+ " <td>0</td>\n",
+ " <td>2</td>\n",
+ " <td>2</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>3</th>\n",
+ " <td>7839</td>\n",
+ " <td>28700</td>\n",
+ " <td>1.50</td>\n",
+ " <td>64</td>\n",
+ " <td>1</td>\n",
+ " <td>3</td>\n",
+ " <td>3</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>4</th>\n",
+ " <td>40877</td>\n",
+ " <td>28700</td>\n",
+ " <td>1.50</td>\n",
+ " <td>68</td>\n",
+ " <td>1</td>\n",
+ " <td>4</td>\n",
+ " <td>3</td>\n",
+ " </tr>\n",
+ " </tbody>\n",
+ "</table>\n",
+ "</div>\n",
+ " <div class=\"colab-df-buttons\">\n",
+ "\n",
+ " <div class=\"colab-df-container\">\n",
+ " <button class=\"colab-df-convert\"
onclick=\"convertToInteractive('df-6c868129-b0f4-49db-b83e-5a2914f6cf1a')\"\n",
+ " title=\"Convert this dataframe to an interactive
table.\"\n",
+ " style=\"display:none;\">\n",
+ "\n",
+ " <svg xmlns=\"http://www.w3.org/2000/svg\"; height=\"24px\"
viewBox=\"0 -960 960 960\">\n",
+ " <path
d=\"M120-120v-720h720v720H120Zm60-500h600v-160H180v160Zm220
220h160v-160H400v160Zm0 220h160v-160H400v160ZM180-400h160v-160H180v160Zm440
0h160v-160H620v160ZM180-180h160v-160H180v160Zm440 0h160v-160H620v160Z\"/>\n",
+ " </svg>\n",
+ " </button>\n",
+ "\n",
+ " <style>\n",
+ " .colab-df-container {\n",
+ " display:flex;\n",
+ " gap: 12px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-convert {\n",
+ " background-color: #E8F0FE;\n",
+ " border: none;\n",
+ " border-radius: 50%;\n",
+ " cursor: pointer;\n",
+ " display: none;\n",
+ " fill: #1967D2;\n",
+ " height: 32px;\n",
+ " padding: 0 0 0 0;\n",
+ " width: 32px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-convert:hover {\n",
+ " background-color: #E2EBFA;\n",
+ " box-shadow: 0px 1px 2px rgba(60, 64, 67, 0.3), 0px 1px
3px 1px rgba(60, 64, 67, 0.15);\n",
+ " fill: #174EA6;\n",
+ " }\n",
+ "\n",
+ " .colab-df-buttons div {\n",
+ " margin-bottom: 4px;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-convert {\n",
+ " background-color: #3B4455;\n",
+ " fill: #D2E3FC;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-convert:hover {\n",
+ " background-color: #434B5C;\n",
+ " box-shadow: 0px 1px 3px 1px rgba(0, 0, 0, 0.15);\n",
+ " filter: drop-shadow(0px 1px 2px rgba(0, 0, 0, 0.3));\n",
+ " fill: #FFFFFF;\n",
+ " }\n",
+ " </style>\n",
+ "\n",
+ " <script>\n",
+ " const buttonEl =\n",
+ "
document.querySelector('#df-6c868129-b0f4-49db-b83e-5a2914f6cf1a
button.colab-df-convert');\n",
+ " buttonEl.style.display =\n",
+ " google.colab.kernel.accessAllowed ? 'block' :
'none';\n",
+ "\n",
+ " async function convertToInteractive(key) {\n",
+ " const element =
document.querySelector('#df-6c868129-b0f4-49db-b83e-5a2914f6cf1a');\n",
+ " const dataTable =\n",
+ " await
google.colab.kernel.invokeFunction('convertToInteractive',\n",
+ " [key],
{});\n",
+ " if (!dataTable) return;\n",
+ "\n",
+ " const docLinkHtml = 'Like what you see? Visit the '
+\n",
+ " '<a target=\"_blank\"
href=https://colab.research.google.com/notebooks/data_table.ipynb>data table
notebook</a>'\n",
+ " + ' to learn more about interactive tables.';\n",
+ " element.innerHTML = '';\n",
+ " dataTable['output_type'] = 'display_data';\n",
+ " await google.colab.output.renderOutput(dataTable,
element);\n",
+ " const docLink = document.createElement('div');\n",
+ " docLink.innerHTML = docLinkHtml;\n",
+ " element.appendChild(docLink);\n",
+ " }\n",
+ " </script>\n",
+ " </div>\n",
+ "\n",
+ "\n",
+ "<div id=\"df-6a7655a2-dc94-4e64-80be-c525111b9a63\">\n",
+ " <button class=\"colab-df-quickchart\"
onclick=\"quickchart('df-6a7655a2-dc94-4e64-80be-c525111b9a63')\"\n",
+ " title=\"Suggest charts\"\n",
+ " style=\"display:none;\">\n",
+ "\n",
+ "<svg xmlns=\"http://www.w3.org/2000/svg\";
height=\"24px\"viewBox=\"0 0 24 24\"\n",
+ " width=\"24px\">\n",
+ " <g>\n",
+ " <path d=\"M19 3H5c-1.1 0-2 .9-2 2v14c0 1.1.9 2 2
2h14c1.1 0 2-.9 2-2V5c0-1.1-.9-2-2-2zM9 17H7v-7h2v7zm4 0h-2V7h2v10zm4
0h-2v-4h2v4z\"/>\n",
+ " </g>\n",
+ "</svg>\n",
+ " </button>\n",
+ "\n",
+ "<style>\n",
+ " .colab-df-quickchart {\n",
+ " --bg-color: #E8F0FE;\n",
+ " --fill-color: #1967D2;\n",
+ " --hover-bg-color: #E2EBFA;\n",
+ " --hover-fill-color: #174EA6;\n",
+ " --disabled-fill-color: #AAA;\n",
+ " --disabled-bg-color: #DDD;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-quickchart {\n",
+ " --bg-color: #3B4455;\n",
+ " --fill-color: #D2E3FC;\n",
+ " --hover-bg-color: #434B5C;\n",
+ " --hover-fill-color: #FFFFFF;\n",
+ " --disabled-bg-color: #3B4455;\n",
+ " --disabled-fill-color: #666;\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart {\n",
+ " background-color: var(--bg-color);\n",
+ " border: none;\n",
+ " border-radius: 50%;\n",
+ " cursor: pointer;\n",
+ " display: none;\n",
+ " fill: var(--fill-color);\n",
+ " height: 32px;\n",
+ " padding: 0;\n",
+ " width: 32px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart:hover {\n",
+ " background-color: var(--hover-bg-color);\n",
+ " box-shadow: 0 1px 2px rgba(60, 64, 67, 0.3), 0 1px 3px 1px
rgba(60, 64, 67, 0.15);\n",
+ " fill: var(--button-hover-fill-color);\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart-complete:disabled,\n",
+ " .colab-df-quickchart-complete:disabled:hover {\n",
+ " background-color: var(--disabled-bg-color);\n",
+ " fill: var(--disabled-fill-color);\n",
+ " box-shadow: none;\n",
+ " }\n",
+ "\n",
+ " .colab-df-spinner {\n",
+ " border: 2px solid var(--fill-color);\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " animation:\n",
+ " spin 1s steps(1) infinite;\n",
+ " }\n",
+ "\n",
+ " @keyframes spin {\n",
+ " 0% {\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " border-left-color: var(--fill-color);\n",
+ " }\n",
+ " 20% {\n",
+ " border-color: transparent;\n",
+ " border-left-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " }\n",
+ " 30% {\n",
+ " border-color: transparent;\n",
+ " border-left-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " border-right-color: var(--fill-color);\n",
+ " }\n",
+ " 40% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " }\n",
+ " 60% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " }\n",
+ " 80% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " }\n",
+ " 90% {\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " }\n",
+ " }\n",
+ "</style>\n",
+ "\n",
+ " <script>\n",
+ " async function quickchart(key) {\n",
+ " const quickchartButtonEl =\n",
+ " document.querySelector('#' + key + ' button');\n",
+ " quickchartButtonEl.disabled = true; // To prevent
multiple clicks.\n",
+ " quickchartButtonEl.classList.add('colab-df-spinner');\n",
+ " try {\n",
+ " const charts = await
google.colab.kernel.invokeFunction(\n",
+ " 'suggestCharts', [key], {});\n",
+ " } catch (error) {\n",
+ " console.error('Error during call to suggestCharts:',
error);\n",
+ " }\n",
+ "
quickchartButtonEl.classList.remove('colab-df-spinner');\n",
+ "
quickchartButtonEl.classList.add('colab-df-quickchart-complete');\n",
+ " }\n",
+ " (() => {\n",
+ " let quickchartButtonEl =\n",
+ "
document.querySelector('#df-6a7655a2-dc94-4e64-80be-c525111b9a63 button');\n",
+ " quickchartButtonEl.style.display =\n",
+ " google.colab.kernel.accessAllowed ? 'block' :
'none';\n",
+ " })();\n",
+ " </script>\n",
+ "</div>\n",
+ "\n",
+ " </div>\n",
+ " </div>\n"
+ ],
+ "text/plain": [
+ " user_id product_id sale_price age gender state
country\n",
+ "0 68717 14235 0.02 43 0 0
0\n",
+ "1 59866 28700 1.50 17 1 1
1\n",
+ "2 38322 14202 1.50 47 0 2
2\n",
+ "3 7839 28700 1.50 64 1 3
3\n",
+ "4 40877 28700 1.50 68 1 4 3"
+ ]
+ },
+ "execution_count": 5,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "# Create a prediction dataframe.\n",
+ "prediction_data = train_data['product_id'].sample(frac=1,
replace=True)\n",
+ "\n",
+ "# Preprocess data to handle categorical values.\n",
+ "train_data['gender'] = pd.factorize(train_data['gender'])[0]\n",
+ "train_data['state'] = pd.factorize(train_data['state'])[0]\n",
+ "train_data['country'] = pd.factorize(train_data['country'])[0]\n",
+ "train_data.head()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "7ffoopdQVk8W"
+ },
+ "source": [
+ "Convert the dataframe to tensors."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "vmHH26KDVkuf"
+ },
+ "outputs": [],
+ "source": [
+ "train_tensors = tf.convert_to_tensor(train_data.values,
dtype=tf.float32)\n",
+ "prediction_tensors = tf.convert_to_tensor(prediction_data.values,
dtype=tf.float32)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "CRoW8ElNV4I9"
+ },
+ "source": [
+ "Based on this data, build a basic neural network model by using
TensorFlow."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "EKrb13wsV3m4"
+ },
+ "outputs": [],
+ "source": [
+ "inputs = layers.Input(shape=(7,))\n",
+ "x = layers.Dense(7, activation='relu')(inputs)\n",
+ "x = layers.Dense(14, activation='relu')(x)\n",
+ "outputs = layers.Dense(1)(x)\n",
+ "\n",
+ "model = keras.Model(inputs=inputs, outputs=outputs)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "Duv4qzmEWFSZ"
+ },
+ "source": [
+ "Train the model. This step takes about 90 seconds for one epoch."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "bHg1kcvnk7Xb"
+ },
+ "outputs": [],
+ "source": [
+ "EPOCHS = 1"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "4GrDp5_WWGZv"
+ },
+ "outputs": [],
+ "source": [
+ "model.compile(optimizer='adam', loss='mse')\n",
+ "model.fit(train_tensors, prediction_tensors, epochs=EPOCHS)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "_rJYv8fFFPYb"
+ },
+ "source": [
+ "Save the model to the `MODEL_PATH` variable.\n",
+ "\n"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "W4t260o9FURP"
+ },
+ "outputs": [],
+ "source": [
+ "# Create a new directory to save the model.\n",
+ "!mkdir model\n",
+ "\n",
+ "# Save the model.\n",
+ "MODEL_PATH = './model/'\n",
+ "tf.saved_model.save(model, MODEL_PATH)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "hsJOxFTWj6JX"
+ },
+ "source": [
+ "Stage the locally saved model to a Google Cloud Storage bucket. Use
this Cloud Storage bucket to deploy the model to Vertex AI. Replace
`<BUCKET_NAME>` with the name of your Cloud Storage bucket. Replace
`<BUCKET_DIRECTORY>` with the path to your Cloud Storage bucket."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "WQp1e_JgllBW"
+ },
+ "outputs": [],
+ "source": [
+ "GCS_BUCKET = '<BUCKET_NAME>'\n",
+ "GCS_BUCKET_DIRECTORY = '<BUCKET_DIRECTORY>'"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "yiXRXV89e8_Y"
+ },
+ "outputs": [],
+ "source": [
+ "# Stage to the Cloud Storage bucket.\n",
+ "import glob\n",
+ "from google.cloud import storage\n",
+ "client = storage.Client(project=PROJECT_ID)\n",
+ "bucket = client.bucket(GCS_BUCKET)\n",
+ "\n",
+ "def upload_model_to_gcs(model_path, bucket, gcs_model_dir):\n",
+ " for file in glob.glob(model_path + '/**', recursive=True):\n",
+ " if os.path.isfile(file):\n",
+ " path = os.path.join(gcs_model_dir, file[1 +
len(model_path.rstrip(\"/\")):])\n",
+ " blob = bucket.blob(path)\n",
+ " blob.upload_from_filename(file)\n",
+ "\n",
+ "\n",
+ "upload_model_to_gcs(MODEL_PATH, bucket, GCS_BUCKET_DIRECTORY)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "O72h009kl_-L"
+ },
+ "source": [
+ "Upload the model saved in the Cloud Storage bucket to Vertex AI Model
Registry."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "bKN5pUD3uImj"
+ },
+ "outputs": [],
+ "source": [
+ "model_display_name = 'vertex-ai-enrichment'"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "Pp3Jca9GfpEj"
+ },
+ "outputs": [],
+ "source": [
+ "aiplatform.init(project=PROJECT_ID, location=LOCATION)\n",
+ "model = aiplatform.Model.upload(\n",
+ " display_name = model_display_name,\n",
+ " description='Model used in the vertex ai enrichment notebook.',\n",
+ " artifact_uri=\"gs://\" + GCS_BUCKET + \"/\" +
GCS_BUCKET_DIRECTORY,\n",
+ "
serving_container_image_uri='us-docker.pkg.dev/vertex-ai/prediction/tf2-cpu.2-13:latest',\n",
+ ")"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "ms_KqSIbZkLP"
+ },
+ "source": [
+ "Create an endpoint on Vertex AI."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";
+ },
+ "id": "YKKzRrN6czni",
+ "outputId": "bfd954c0-8267-476d-dd0c-15e612ae0cc1"
+ },
+ "outputs": [],
+ "source": [
+ "endpoint = aiplatform.Endpoint.create(display_name =
model_display_name,\n",
+ " project = PROJECT_ID,\n",
+ " location = LOCATION)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "WgSpy0J3oBFP"
+ },
+ "source": [
+ "Deploy the model to the Vertex AI endpoint.\n",
+ "\n",
+ "**Note:** This step is a Long Running Operation (LRO). Depending on
the size of the model, it might take more than five minutes to complete."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "FLQtMVQjnsls"
+ },
+ "outputs": [],
+ "source": [
+ "deployed_model_display_name = 'vertexai-enrichment-notebook'\n",
+ "model.deploy(endpoint = endpoint,\n",
+ " deployed_model_display_name =
deployed_model_display_name,\n",
+ " machine_type = 'n1-standard-2')"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";
+ },
+ "id": "3JjIwzZouAi5",
+ "outputId": "ffb1fb74-365a-426b-d60d-d3910c116e10"
+ },
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "8125472293125095424\n"
+ ]
+ }
+ ],
+ "source": [
+ "model_endpoint_id =
aiplatform.Endpoint.list(filter=f'display_name=\"{deployed_model_display_name}\"')[0].name\n",
+ "print(model_endpoint_id)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "ouMQZ4sC4zuO"
+ },
+ "source": [
+ "### Set up the Vertex AI Feature Store for online serving\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "B1Bk7XP7190z"
+ },
+ "source": [
+ "Set up the feature data in BigQuery."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";,
+ "height": 206
+ },
+ "id": "4Qkysu_g19c_",
+ "outputId": "187ee1e8-07c9-457a-abbe-fab724d997ce"
+ },
+ "outputs": [
+ {
+ "data": {
+ "application/vnd.google.colaboratory.intrinsic+json": {
+ "summary": "{\n \"name\": \"data\",\n \"rows\": 100000,\n
\"fields\": [\n {\n \"column\": \"user_id\",\n \"properties\": {\n
\"dtype\": \"string\",\n \"num_unique_values\": 100000,\n
\"samples\": [\n \"66192\",\n \"73109\",\n
\"49397\"\n ],\n \"semantic_type\": \"\",\n
\"description\": \"\"\n }\n },\n {\n \"column\": \"age\",\n
\"properties\": {\n \"dtype\": \"In [...]
+ "type": "dataframe",
+ "variable_name": "data"
+ },
+ "text/html": [
+ "\n",
+ " <div id=\"df-8aa66ce8-dfee-43c9-bbea-f2b02e8a007e\"
class=\"colab-df-container\">\n",
+ " <div>\n",
+ "<style scoped>\n",
+ " .dataframe tbody tr th:only-of-type {\n",
+ " vertical-align: middle;\n",
+ " }\n",
+ "\n",
+ " .dataframe tbody tr th {\n",
+ " vertical-align: top;\n",
+ " }\n",
+ "\n",
+ " .dataframe thead th {\n",
+ " text-align: right;\n",
+ " }\n",
+ "</style>\n",
+ "<table border=\"1\" class=\"dataframe\">\n",
+ " <thead>\n",
+ " <tr style=\"text-align: right;\">\n",
+ " <th></th>\n",
+ " <th>user_id</th>\n",
+ " <th>age</th>\n",
+ " <th>gender</th>\n",
+ " <th>state</th>\n",
+ " <th>country</th>\n",
+ " </tr>\n",
+ " </thead>\n",
+ " <tbody>\n",
+ " <tr>\n",
+ " <th>0</th>\n",
+ " <td>7723</td>\n",
+ " <td>12</td>\n",
+ " <td>0</td>\n",
+ " <td>0</td>\n",
+ " <td>0</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>1</th>\n",
+ " <td>93041</td>\n",
+ " <td>12</td>\n",
+ " <td>0</td>\n",
+ " <td>1</td>\n",
+ " <td>1</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>2</th>\n",
+ " <td>45741</td>\n",
+ " <td>12</td>\n",
+ " <td>1</td>\n",
+ " <td>1</td>\n",
+ " <td>1</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>3</th>\n",
+ " <td>16718</td>\n",
+ " <td>12</td>\n",
+ " <td>0</td>\n",
+ " <td>1</td>\n",
+ " <td>1</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>4</th>\n",
+ " <td>70137</td>\n",
+ " <td>12</td>\n",
+ " <td>1</td>\n",
+ " <td>1</td>\n",
+ " <td>1</td>\n",
+ " </tr>\n",
+ " </tbody>\n",
+ "</table>\n",
+ "</div>\n",
+ " <div class=\"colab-df-buttons\">\n",
+ "\n",
+ " <div class=\"colab-df-container\">\n",
+ " <button class=\"colab-df-convert\"
onclick=\"convertToInteractive('df-8aa66ce8-dfee-43c9-bbea-f2b02e8a007e')\"\n",
+ " title=\"Convert this dataframe to an interactive
table.\"\n",
+ " style=\"display:none;\">\n",
+ "\n",
+ " <svg xmlns=\"http://www.w3.org/2000/svg\"; height=\"24px\"
viewBox=\"0 -960 960 960\">\n",
+ " <path
d=\"M120-120v-720h720v720H120Zm60-500h600v-160H180v160Zm220
220h160v-160H400v160Zm0 220h160v-160H400v160ZM180-400h160v-160H180v160Zm440
0h160v-160H620v160ZM180-180h160v-160H180v160Zm440 0h160v-160H620v160Z\"/>\n",
+ " </svg>\n",
+ " </button>\n",
+ "\n",
+ " <style>\n",
+ " .colab-df-container {\n",
+ " display:flex;\n",
+ " gap: 12px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-convert {\n",
+ " background-color: #E8F0FE;\n",
+ " border: none;\n",
+ " border-radius: 50%;\n",
+ " cursor: pointer;\n",
+ " display: none;\n",
+ " fill: #1967D2;\n",
+ " height: 32px;\n",
+ " padding: 0 0 0 0;\n",
+ " width: 32px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-convert:hover {\n",
+ " background-color: #E2EBFA;\n",
+ " box-shadow: 0px 1px 2px rgba(60, 64, 67, 0.3), 0px 1px
3px 1px rgba(60, 64, 67, 0.15);\n",
+ " fill: #174EA6;\n",
+ " }\n",
+ "\n",
+ " .colab-df-buttons div {\n",
+ " margin-bottom: 4px;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-convert {\n",
+ " background-color: #3B4455;\n",
+ " fill: #D2E3FC;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-convert:hover {\n",
+ " background-color: #434B5C;\n",
+ " box-shadow: 0px 1px 3px 1px rgba(0, 0, 0, 0.15);\n",
+ " filter: drop-shadow(0px 1px 2px rgba(0, 0, 0, 0.3));\n",
+ " fill: #FFFFFF;\n",
+ " }\n",
+ " </style>\n",
+ "\n",
+ " <script>\n",
+ " const buttonEl =\n",
+ "
document.querySelector('#df-8aa66ce8-dfee-43c9-bbea-f2b02e8a007e
button.colab-df-convert');\n",
+ " buttonEl.style.display =\n",
+ " google.colab.kernel.accessAllowed ? 'block' :
'none';\n",
+ "\n",
+ " async function convertToInteractive(key) {\n",
+ " const element =
document.querySelector('#df-8aa66ce8-dfee-43c9-bbea-f2b02e8a007e');\n",
+ " const dataTable =\n",
+ " await
google.colab.kernel.invokeFunction('convertToInteractive',\n",
+ " [key],
{});\n",
+ " if (!dataTable) return;\n",
+ "\n",
+ " const docLinkHtml = 'Like what you see? Visit the '
+\n",
+ " '<a target=\"_blank\"
href=https://colab.research.google.com/notebooks/data_table.ipynb>data table
notebook</a>'\n",
+ " + ' to learn more about interactive tables.';\n",
+ " element.innerHTML = '';\n",
+ " dataTable['output_type'] = 'display_data';\n",
+ " await google.colab.output.renderOutput(dataTable,
element);\n",
+ " const docLink = document.createElement('div');\n",
+ " docLink.innerHTML = docLinkHtml;\n",
+ " element.appendChild(docLink);\n",
+ " }\n",
+ " </script>\n",
+ " </div>\n",
+ "\n",
+ "\n",
+ "<div id=\"df-cdc48cec-6814-456d-af79-c67163baa6ec\">\n",
+ " <button class=\"colab-df-quickchart\"
onclick=\"quickchart('df-cdc48cec-6814-456d-af79-c67163baa6ec')\"\n",
+ " title=\"Suggest charts\"\n",
+ " style=\"display:none;\">\n",
+ "\n",
+ "<svg xmlns=\"http://www.w3.org/2000/svg\";
height=\"24px\"viewBox=\"0 0 24 24\"\n",
+ " width=\"24px\">\n",
+ " <g>\n",
+ " <path d=\"M19 3H5c-1.1 0-2 .9-2 2v14c0 1.1.9 2 2
2h14c1.1 0 2-.9 2-2V5c0-1.1-.9-2-2-2zM9 17H7v-7h2v7zm4 0h-2V7h2v10zm4
0h-2v-4h2v4z\"/>\n",
+ " </g>\n",
+ "</svg>\n",
+ " </button>\n",
+ "\n",
+ "<style>\n",
+ " .colab-df-quickchart {\n",
+ " --bg-color: #E8F0FE;\n",
+ " --fill-color: #1967D2;\n",
+ " --hover-bg-color: #E2EBFA;\n",
+ " --hover-fill-color: #174EA6;\n",
+ " --disabled-fill-color: #AAA;\n",
+ " --disabled-bg-color: #DDD;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-quickchart {\n",
+ " --bg-color: #3B4455;\n",
+ " --fill-color: #D2E3FC;\n",
+ " --hover-bg-color: #434B5C;\n",
+ " --hover-fill-color: #FFFFFF;\n",
+ " --disabled-bg-color: #3B4455;\n",
+ " --disabled-fill-color: #666;\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart {\n",
+ " background-color: var(--bg-color);\n",
+ " border: none;\n",
+ " border-radius: 50%;\n",
+ " cursor: pointer;\n",
+ " display: none;\n",
+ " fill: var(--fill-color);\n",
+ " height: 32px;\n",
+ " padding: 0;\n",
+ " width: 32px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart:hover {\n",
+ " background-color: var(--hover-bg-color);\n",
+ " box-shadow: 0 1px 2px rgba(60, 64, 67, 0.3), 0 1px 3px 1px
rgba(60, 64, 67, 0.15);\n",
+ " fill: var(--button-hover-fill-color);\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart-complete:disabled,\n",
+ " .colab-df-quickchart-complete:disabled:hover {\n",
+ " background-color: var(--disabled-bg-color);\n",
+ " fill: var(--disabled-fill-color);\n",
+ " box-shadow: none;\n",
+ " }\n",
+ "\n",
+ " .colab-df-spinner {\n",
+ " border: 2px solid var(--fill-color);\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " animation:\n",
+ " spin 1s steps(1) infinite;\n",
+ " }\n",
+ "\n",
+ " @keyframes spin {\n",
+ " 0% {\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " border-left-color: var(--fill-color);\n",
+ " }\n",
+ " 20% {\n",
+ " border-color: transparent;\n",
+ " border-left-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " }\n",
+ " 30% {\n",
+ " border-color: transparent;\n",
+ " border-left-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " border-right-color: var(--fill-color);\n",
+ " }\n",
+ " 40% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " }\n",
+ " 60% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " }\n",
+ " 80% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " }\n",
+ " 90% {\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " }\n",
+ " }\n",
+ "</style>\n",
+ "\n",
+ " <script>\n",
+ " async function quickchart(key) {\n",
+ " const quickchartButtonEl =\n",
+ " document.querySelector('#' + key + ' button');\n",
+ " quickchartButtonEl.disabled = true; // To prevent
multiple clicks.\n",
+ " quickchartButtonEl.classList.add('colab-df-spinner');\n",
+ " try {\n",
+ " const charts = await
google.colab.kernel.invokeFunction(\n",
+ " 'suggestCharts', [key], {});\n",
+ " } catch (error) {\n",
+ " console.error('Error during call to suggestCharts:',
error);\n",
+ " }\n",
+ "
quickchartButtonEl.classList.remove('colab-df-spinner');\n",
+ "
quickchartButtonEl.classList.add('colab-df-quickchart-complete');\n",
+ " }\n",
+ " (() => {\n",
+ " let quickchartButtonEl =\n",
+ "
document.querySelector('#df-cdc48cec-6814-456d-af79-c67163baa6ec button');\n",
+ " quickchartButtonEl.style.display =\n",
+ " google.colab.kernel.accessAllowed ? 'block' :
'none';\n",
+ " })();\n",
+ " </script>\n",
+ "</div>\n",
+ "\n",
+ " </div>\n",
+ " </div>\n"
+ ],
+ "text/plain": [
+ " user_id age gender state country\n",
+ "0 7723 12 0 0 0\n",
+ "1 93041 12 0 1 1\n",
+ "2 45741 12 1 1 1\n",
+ "3 16718 12 0 1 1\n",
+ "4 70137 12 1 1 1"
+ ]
+ },
+ "execution_count": 8,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "feature_store_query = \"\"\"\n",
+ "SELECT cast(id as string) AS user_id,\n",
+ " age,\n",
+ " lower(gender) as gender,\n",
+ " lower(state) as state,\n",
+ " lower(country) as country,\n",
+ "FROM `bigquery-public-data.thelook_ecommerce.users`\n",
+ "\"\"\"\n",
+ "\n",
+ "# Fetch feature values from BigQuery.\n",
+ "client = bigquery.Client(project=PROJECT_ID)\n",
+ "data = client.query(feature_store_query).result().to_dataframe()\n",
+ "\n",
+ "# Convert feature values to the string type. This step helps when
creating tensors\n",
+ "# of these values for inference that requires the same data type.\n",
+ "data['gender'] = pd.factorize(data['gender'])[0]\n",
+ "data['gender'] = data['gender'].astype(str)\n",
+ "data['state'] = pd.factorize(data['state'])[0]\n",
+ "data['state'] = data['state'].astype(str)\n",
+ "data['country'] = pd.factorize(data['country'])[0]\n",
+ "data['country'] = data['country'].astype(str)\n",
+ "data.head()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "Mm-HCUaa3ROZ"
+ },
+ "source": [
+ "Create a BigQuery dataset to use as the source for the Vertex AI
Feature Store."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";
+ },
+ "id": "vye3UBGZ3Q8n",
+ "outputId": "437597af-837d-483e-8c1e-ebbe0eca81e0"
+ },
+ "outputs": [],
+ "source": [
+ "dataset_id = \"vertexai_enrichment\"\n",
+ "dataset = bigquery.Dataset(f\"{PROJECT_ID}.{dataset_id}\")\n",
+ "dataset.location = \"US\"\n",
+ "dataset = client.create_dataset(\n",
+ " dataset, exists_ok=True, timeout=30\n",
+ ")\n",
+ "\n",
+ "print(\"Created dataset - %s.%s\" % (dataset, dataset_id))"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "7lKiprPX4AZy"
+ },
+ "source": [
+ "Create a BigQuery view with the precomputed feature values."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "xqaLPTxb4DDF"
+ },
+ "outputs": [],
+ "source": [
+ "view_id = \"users_view\"\n",
+ "view_reference = \"%s.%s.%s\" % (PROJECT_ID, dataset_id, view_id)\n",
+ "view = bigquery.Table(view_reference)\n",
+ "view = client.load_table_from_dataframe(data, view_reference)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "eQLkSg3p7WAm"
+ },
+ "source": [
+ "Initialize clients for Vertex AI to create and set up an online
store."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "GF_eIl-wVvRy"
+ },
+ "outputs": [],
+ "source": [
+ "API_ENDPOINT = f\"{LOCATION}-aiplatform.googleapis.com\"\n",
+ "\n",
+ "admin_client = FeatureOnlineStoreAdminServiceClient(\n",
+ " client_options={\"api_endpoint\": API_ENDPOINT}\n",
+ ")\n",
+ "registry_client = FeatureRegistryServiceClient(\n",
+ " client_options={\"api_endpoint\": API_ENDPOINT}\n",
+ ")"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "d9Mbk6m9Vgdo"
+ },
+ "source": [
+ "Create an online store instances on Vertex AI."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";
+ },
+ "id": "Zj-xEu_hWY7f",
+ "outputId": "7f4ed1d9-c0c4-4c3c-f199-1e340d2cff11"
+ },
+ "outputs": [],
+ "source": [
+ "feature_store_name = \"vertexai_enrichment\"\n",
+ "\n",
+ "online_store_config = feature_online_store_pb2.FeatureOnlineStore(\n",
+ " bigtable=feature_online_store_pb2.FeatureOnlineStore.Bigtable(\n",
+ "
auto_scaling=feature_online_store_pb2.FeatureOnlineStore.Bigtable.AutoScaling(\n",
+ " min_node_count=1, max_node_count=1,
cpu_utilization_target=80\n",
+ " )\n",
+ " )\n",
+ ")\n",
+ "\n",
+ "create_store_lro = admin_client.create_feature_online_store(\n",
+ "
feature_online_store_admin_service_pb2.CreateFeatureOnlineStoreRequest(\n",
+ " parent=f\"projects/{PROJECT_ID}/locations/{LOCATION}\",\n",
+ " feature_online_store_id=feature_store_name,\n",
+ " feature_online_store=online_store_config,\n",
+ " )\n",
+ ")\n",
+ "\n",
+ "create_store_lro.result()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "DAHjWlqXXLU_"
+ },
+ "source": [
+ "For the store instances created previously, use BigQuery as the data
source to create feature views."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";
+ },
+ "id": "IhUERuRGXNaN",
+ "outputId": "84facd77-5be4-4c99-90b5-d8ccb4c5d702"
+ },
+ "outputs": [],
+ "source": [
+ "feature_view_name = \"users\"\n",
+ "\n",
+ "bigquery_source = feature_view_pb2.FeatureView.BigQuerySource(\n",
+ " uri=f\"bq://{view_reference}\",
entity_id_columns=[\"user_id\"]\n",
+ ")\n",
+ "\n",
+ "create_view_lro = admin_client.create_feature_view(\n",
+ "
feature_online_store_admin_service_pb2.CreateFeatureViewRequest(\n",
+ "
parent=f\"projects/{PROJECT_ID}/locations/{LOCATION}/featureOnlineStores/{feature_store_name}\",\n",
+ " feature_view_id=feature_view_name,\n",
+ " feature_view=feature_view_pb2.FeatureView(\n",
+ " big_query_source=bigquery_source,\n",
+ " ),\n",
+ " )\n",
+ ")\n",
+ "\n",
+ "create_view_lro.result()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "qbf4l8eBX6NG"
+ },
+ "source": [
+ "Pull feature values from BigQuery into the feature store."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "gdpsLCmMX7fX"
+ },
+ "outputs": [],
+ "source": [
+ "sync_response = admin_client.sync_feature_view(\n",
+ "
feature_view=f\"projects/{PROJECT_ID}/locations/{LOCATION}/featureOnlineStores/{feature_store_name}/featureViews/{feature_view_name}\"\n",
+ ")"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "Lav6JTW4YKhR"
+ },
+ "outputs": [],
+ "source": [
+ "while True:\n",
+ " feature_view_sync = admin_client.get_feature_view_sync(\n",
+ " name=sync_response.feature_view_sync\n",
+ " )\n",
+ " if feature_view_sync.run_time.end_time.seconds > 0:\n",
+ " if feature_view_sync.final_status.code == 0\n",
+ " print(\"feature view sync completed for %s\" %
feature_view_sync.name)\n",
+ " else:\n",
+ " print(\"feature view sync failed for %s\" %
feature_view_sync.name)\n",
+ " break\n",
+ " time.sleep(10)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "T3MMx7oJYPeC"
+ },
+ "source": [
+ "Confirm the sync creation."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";
+ },
+ "id": "ucSQRUfUYRFX",
+ "outputId": "d2160812-9874-40bb-f464-f797eafb9999"
+ },
+ "outputs": [],
+ "source": [
+ "admin_client.list_feature_view_syncs(\n",
+ "
parent=f\"projects/{PROJECT_ID}/locations/{LOCATION}/featureOnlineStores/{feature_store_name}/featureViews/{feature_view_name}\"\n",
+ ")"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "pHODouJDwc60"
+ },
+ "source": [
+ "### Publish messages to Pub/Sub\n",
+ "\n",
+ "Use the Pub/Sub Python client to publish messages.\n"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "QKCuwDioxw-f"
+ },
+ "outputs": [],
+ "source": [
+ "# Replace <TOPIC_NAME> with the name of your Pub/Sub topic.\n",
+ "TOPIC = \"<TOPIC_NAME> \"\n",
+ "\n",
+ "# Replace <SUBSCRIPTION_NAME> with the subscription path for your
topic.\n",
+ "SUBSCRIPTION = \"<SUBSCRIPTION_NAME>\""
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "R0QYsOYFb_EU"
+ },
+ "source": [
+ "Retrieve sample data from a public dataset in BigQuery. Convert it
into Python dictionaries, and then send it to Pub/Sub."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";,
+ "height": 206
+ },
+ "id": "Kn7wmiKib-Wx",
+ "outputId": "9680fbcc-dcb5-4158-90ae-69a9f3c776d0"
+ },
+ "outputs": [
+ {
+ "data": {
+ "application/vnd.google.colaboratory.intrinsic+json": {
+ "summary": "{\n \"name\": \"data\",\n \"rows\": 5,\n
\"fields\": [\n {\n \"column\": \"user_id\",\n \"properties\": {\n
\"dtype\": \"string\",\n \"num_unique_values\": 5,\n
\"samples\": [\n \"62544\",\n \"16569\",\n
\"17228\"\n ],\n \"semantic_type\": \"\",\n
\"description\": \"\"\n }\n },\n {\n \"column\":
\"product_id\",\n \"properties\": {\n \"dtype\": \"Int64 [...]
+ "type": "dataframe",
+ "variable_name": "data"
+ },
+ "text/html": [
+ "\n",
+ " <div id=\"df-c0bd3a49-0e2d-4cfe-97cb-25a78b4402db\"
class=\"colab-df-container\">\n",
+ " <div>\n",
+ "<style scoped>\n",
+ " .dataframe tbody tr th:only-of-type {\n",
+ " vertical-align: middle;\n",
+ " }\n",
+ "\n",
+ " .dataframe tbody tr th {\n",
+ " vertical-align: top;\n",
+ " }\n",
+ "\n",
+ " .dataframe thead th {\n",
+ " text-align: right;\n",
+ " }\n",
+ "</style>\n",
+ "<table border=\"1\" class=\"dataframe\">\n",
+ " <thead>\n",
+ " <tr style=\"text-align: right;\">\n",
+ " <th></th>\n",
+ " <th>user_id</th>\n",
+ " <th>product_id</th>\n",
+ " <th>sale_price</th>\n",
+ " </tr>\n",
+ " </thead>\n",
+ " <tbody>\n",
+ " <tr>\n",
+ " <th>0</th>\n",
+ " <td>25005</td>\n",
+ " <td>14235</td>\n",
+ " <td>0.02</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>1</th>\n",
+ " <td>62544</td>\n",
+ " <td>14235</td>\n",
+ " <td>0.02</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>2</th>\n",
+ " <td>17228</td>\n",
+ " <td>14235</td>\n",
+ " <td>0.02</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>3</th>\n",
+ " <td>54015</td>\n",
+ " <td>14235</td>\n",
+ " <td>0.02</td>\n",
+ " </tr>\n",
+ " <tr>\n",
+ " <th>4</th>\n",
+ " <td>16569</td>\n",
+ " <td>14235</td>\n",
+ " <td>0.02</td>\n",
+ " </tr>\n",
+ " </tbody>\n",
+ "</table>\n",
+ "</div>\n",
+ " <div class=\"colab-df-buttons\">\n",
+ "\n",
+ " <div class=\"colab-df-container\">\n",
+ " <button class=\"colab-df-convert\"
onclick=\"convertToInteractive('df-c0bd3a49-0e2d-4cfe-97cb-25a78b4402db')\"\n",
+ " title=\"Convert this dataframe to an interactive
table.\"\n",
+ " style=\"display:none;\">\n",
+ "\n",
+ " <svg xmlns=\"http://www.w3.org/2000/svg\"; height=\"24px\"
viewBox=\"0 -960 960 960\">\n",
+ " <path
d=\"M120-120v-720h720v720H120Zm60-500h600v-160H180v160Zm220
220h160v-160H400v160Zm0 220h160v-160H400v160ZM180-400h160v-160H180v160Zm440
0h160v-160H620v160ZM180-180h160v-160H180v160Zm440 0h160v-160H620v160Z\"/>\n",
+ " </svg>\n",
+ " </button>\n",
+ "\n",
+ " <style>\n",
+ " .colab-df-container {\n",
+ " display:flex;\n",
+ " gap: 12px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-convert {\n",
+ " background-color: #E8F0FE;\n",
+ " border: none;\n",
+ " border-radius: 50%;\n",
+ " cursor: pointer;\n",
+ " display: none;\n",
+ " fill: #1967D2;\n",
+ " height: 32px;\n",
+ " padding: 0 0 0 0;\n",
+ " width: 32px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-convert:hover {\n",
+ " background-color: #E2EBFA;\n",
+ " box-shadow: 0px 1px 2px rgba(60, 64, 67, 0.3), 0px 1px
3px 1px rgba(60, 64, 67, 0.15);\n",
+ " fill: #174EA6;\n",
+ " }\n",
+ "\n",
+ " .colab-df-buttons div {\n",
+ " margin-bottom: 4px;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-convert {\n",
+ " background-color: #3B4455;\n",
+ " fill: #D2E3FC;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-convert:hover {\n",
+ " background-color: #434B5C;\n",
+ " box-shadow: 0px 1px 3px 1px rgba(0, 0, 0, 0.15);\n",
+ " filter: drop-shadow(0px 1px 2px rgba(0, 0, 0, 0.3));\n",
+ " fill: #FFFFFF;\n",
+ " }\n",
+ " </style>\n",
+ "\n",
+ " <script>\n",
+ " const buttonEl =\n",
+ "
document.querySelector('#df-c0bd3a49-0e2d-4cfe-97cb-25a78b4402db
button.colab-df-convert');\n",
+ " buttonEl.style.display =\n",
+ " google.colab.kernel.accessAllowed ? 'block' :
'none';\n",
+ "\n",
+ " async function convertToInteractive(key) {\n",
+ " const element =
document.querySelector('#df-c0bd3a49-0e2d-4cfe-97cb-25a78b4402db');\n",
+ " const dataTable =\n",
+ " await
google.colab.kernel.invokeFunction('convertToInteractive',\n",
+ " [key],
{});\n",
+ " if (!dataTable) return;\n",
+ "\n",
+ " const docLinkHtml = 'Like what you see? Visit the '
+\n",
+ " '<a target=\"_blank\"
href=https://colab.research.google.com/notebooks/data_table.ipynb>data table
notebook</a>'\n",
+ " + ' to learn more about interactive tables.';\n",
+ " element.innerHTML = '';\n",
+ " dataTable['output_type'] = 'display_data';\n",
+ " await google.colab.output.renderOutput(dataTable,
element);\n",
+ " const docLink = document.createElement('div');\n",
+ " docLink.innerHTML = docLinkHtml;\n",
+ " element.appendChild(docLink);\n",
+ " }\n",
+ " </script>\n",
+ " </div>\n",
+ "\n",
+ "\n",
+ "<div id=\"df-d22dfcc1-aa4a-402f-915a-84aa081a58a9\">\n",
+ " <button class=\"colab-df-quickchart\"
onclick=\"quickchart('df-d22dfcc1-aa4a-402f-915a-84aa081a58a9')\"\n",
+ " title=\"Suggest charts\"\n",
+ " style=\"display:none;\">\n",
+ "\n",
+ "<svg xmlns=\"http://www.w3.org/2000/svg\";
height=\"24px\"viewBox=\"0 0 24 24\"\n",
+ " width=\"24px\">\n",
+ " <g>\n",
+ " <path d=\"M19 3H5c-1.1 0-2 .9-2 2v14c0 1.1.9 2 2
2h14c1.1 0 2-.9 2-2V5c0-1.1-.9-2-2-2zM9 17H7v-7h2v7zm4 0h-2V7h2v10zm4
0h-2v-4h2v4z\"/>\n",
+ " </g>\n",
+ "</svg>\n",
+ " </button>\n",
+ "\n",
+ "<style>\n",
+ " .colab-df-quickchart {\n",
+ " --bg-color: #E8F0FE;\n",
+ " --fill-color: #1967D2;\n",
+ " --hover-bg-color: #E2EBFA;\n",
+ " --hover-fill-color: #174EA6;\n",
+ " --disabled-fill-color: #AAA;\n",
+ " --disabled-bg-color: #DDD;\n",
+ " }\n",
+ "\n",
+ " [theme=dark] .colab-df-quickchart {\n",
+ " --bg-color: #3B4455;\n",
+ " --fill-color: #D2E3FC;\n",
+ " --hover-bg-color: #434B5C;\n",
+ " --hover-fill-color: #FFFFFF;\n",
+ " --disabled-bg-color: #3B4455;\n",
+ " --disabled-fill-color: #666;\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart {\n",
+ " background-color: var(--bg-color);\n",
+ " border: none;\n",
+ " border-radius: 50%;\n",
+ " cursor: pointer;\n",
+ " display: none;\n",
+ " fill: var(--fill-color);\n",
+ " height: 32px;\n",
+ " padding: 0;\n",
+ " width: 32px;\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart:hover {\n",
+ " background-color: var(--hover-bg-color);\n",
+ " box-shadow: 0 1px 2px rgba(60, 64, 67, 0.3), 0 1px 3px 1px
rgba(60, 64, 67, 0.15);\n",
+ " fill: var(--button-hover-fill-color);\n",
+ " }\n",
+ "\n",
+ " .colab-df-quickchart-complete:disabled,\n",
+ " .colab-df-quickchart-complete:disabled:hover {\n",
+ " background-color: var(--disabled-bg-color);\n",
+ " fill: var(--disabled-fill-color);\n",
+ " box-shadow: none;\n",
+ " }\n",
+ "\n",
+ " .colab-df-spinner {\n",
+ " border: 2px solid var(--fill-color);\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " animation:\n",
+ " spin 1s steps(1) infinite;\n",
+ " }\n",
+ "\n",
+ " @keyframes spin {\n",
+ " 0% {\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " border-left-color: var(--fill-color);\n",
+ " }\n",
+ " 20% {\n",
+ " border-color: transparent;\n",
+ " border-left-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " }\n",
+ " 30% {\n",
+ " border-color: transparent;\n",
+ " border-left-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " border-right-color: var(--fill-color);\n",
+ " }\n",
+ " 40% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " border-top-color: var(--fill-color);\n",
+ " }\n",
+ " 60% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " }\n",
+ " 80% {\n",
+ " border-color: transparent;\n",
+ " border-right-color: var(--fill-color);\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " }\n",
+ " 90% {\n",
+ " border-color: transparent;\n",
+ " border-bottom-color: var(--fill-color);\n",
+ " }\n",
+ " }\n",
+ "</style>\n",
+ "\n",
+ " <script>\n",
+ " async function quickchart(key) {\n",
+ " const quickchartButtonEl =\n",
+ " document.querySelector('#' + key + ' button');\n",
+ " quickchartButtonEl.disabled = true; // To prevent
multiple clicks.\n",
+ " quickchartButtonEl.classList.add('colab-df-spinner');\n",
+ " try {\n",
+ " const charts = await
google.colab.kernel.invokeFunction(\n",
+ " 'suggestCharts', [key], {});\n",
+ " } catch (error) {\n",
+ " console.error('Error during call to suggestCharts:',
error);\n",
+ " }\n",
+ "
quickchartButtonEl.classList.remove('colab-df-spinner');\n",
+ "
quickchartButtonEl.classList.add('colab-df-quickchart-complete');\n",
+ " }\n",
+ " (() => {\n",
+ " let quickchartButtonEl =\n",
+ "
document.querySelector('#df-d22dfcc1-aa4a-402f-915a-84aa081a58a9 button');\n",
+ " quickchartButtonEl.style.display =\n",
+ " google.colab.kernel.accessAllowed ? 'block' :
'none';\n",
+ " })();\n",
+ " </script>\n",
+ "</div>\n",
+ "\n",
+ " </div>\n",
+ " </div>\n"
+ ],
+ "text/plain": [
+ " user_id product_id sale_price\n",
+ "0 25005 14235 0.02\n",
+ "1 62544 14235 0.02\n",
+ "2 17228 14235 0.02\n",
+ "3 54015 14235 0.02\n",
+ "4 16569 14235 0.02"
+ ]
+ },
+ "execution_count": 18,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "read_query = \"\"\"\n",
+ "SELECT cast(user_id as string) AS user_id,\n",
+ " product_id,\n",
+ " sale_price,\n",
+ "FROM `bigquery-public-data.thelook_ecommerce.order_items`\n",
+ "LIMIT 5;\n",
+ "\"\"\"\n",
+ "\n",
+ "client = bigquery.Client(project=PROJECT_ID)\n",
+ "data = client.query(read_query).result().to_dataframe()\n",
+ "data.head()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "MaCJwaPexPKZ"
+ },
+ "outputs": [],
+ "source": [
+ "messages = data.to_dict(orient='records')\n",
+ "\n",
+ "publisher = pubsub_v1.PublisherClient()\n",
+ "topic_name = publisher.topic_path(PROJECT_ID, TOPIC)\n",
+ "subscription_path = publisher.subscription_path(PROJECT_ID,
SUBSCRIPTION)\n",
+ "for message in messages:\n",
+ " data = json.dumps(message).encode('utf-8')\n",
+ " publish_future = publisher.publish(topic_name, data)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "zPSFEMm02omi"
+ },
+ "source": [
+ "## Use the Vertex AI Feature Store enrichment handler\n",
+ "\n",
+ "The
[`VertexAIFeatureStoreEnrichmentHandler`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.enrichment_handlers.vertex_ai_feature_store.html#apache_beam.transforms.enrichment_handlers.vertex_ai_feature_store.VertexAIFeatureStoreEnrichmentHandler)
is a built-in handler included in the Apache Beam SDK versions 2.55.0 and
later."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "K41xhvmA5yQk"
+ },
+ "source": [
+ "Configure the `VertexAIFeatureStoreEnrichmentHandler` with the
following required parameters:\n",
+ "\n",
+ "* `project`: the Google Cloud project ID for the feature store\n",
+ "* `location`: the region of the feature store, for example
`us-central1`\n",
+ "* `api_endpoint`: the public endpoint of the feature store\n",
+ "* `feature_store_name`: the name of the Vertex AI Feature Store\n",
+ "* `feature_view_name`: the name of the feature view within the
Vertex AI Feature Store\n",
+ "* `row_key`: The field name in the input row containing the entity
ID for the feature store. This value is used to extract the entity ID from each
element. The entity ID is used to fetch feature values for that specific
element in the enrichment transform.\n",
+ "\n",
+ "Optionally, to provide more configuration values to connect with the
Vertex AI client, the `VertexAIFeatureStoreEnrichmentHandler` accepts a keyword
argument (kwargs). For more information, see
[`FeatureOnlineStoreServiceClient`](https://cloud.google.com/php/docs/reference/cloud-ai-platform/latest/V1.FeatureOnlineStoreServiceClient).\n",
+ "\n",
+ "**Note:** When exceptions occur, by default, the logging severity is
set to warning
([`ExceptionLevel.WARN`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.enrichment_handlers.utils.html#apache_beam.transforms.enrichment_handlers.utils.ExceptionLevel.WARN)).
To configure the severity to raise exceptions, set `exception_level` to
[`ExceptionLevel.RAISE`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.enrichment_handlers.utils.html#apa
[...]
+ "\n",
+ "The `VertexAIFeatureStoreEnrichmentHandler` handler returns the
latest feature values from the feature store."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "3dB26jhI45gd"
+ },
+ "outputs": [],
+ "source": [
+ "row_key = 'user_id'"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "cr1j_DHK4gA4"
+ },
+ "outputs": [],
+ "source": [
+ "vertex_ai_handler =
VertexAIFeatureStoreEnrichmentHandler(project=PROJECT_ID,\n",
+ " location=LOCATION,\n",
+ " api_endpoint =
API_ENDPOINT,\n",
+ "
feature_store_name=feature_store_name,\n",
+ "
feature_view_name=feature_view_name,\n",
+ " row_key=row_key)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "-Lvo8O2V-0Ey"
+ },
+ "source": [
+ "## Use the enrichment transform\n",
+ "\n",
+ "To use the [enrichment
transform](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.enrichment.html#apache_beam.transforms.enrichment.Enrichment),
the
[`EnrichmentHandler`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.enrichment.html#apache_beam.transforms.enrichment.EnrichmentSourceHandler)
parameter is required. You can also use a configuration parameter to specify a
`lambda` for a join function, a timeout, a throttler, and a repeat [...]
+ "\n",
+ "\n",
+ "* `join_fn`: A lambda function that takes dictionaries as input and
returns an enriched row (`Callable[[Dict[str, Any], Dict[str, Any]],
beam.Row]`). The enriched row specifies how to join the data fetched from the
API. Defaults to a
[cross-join](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.enrichment.html#apache_beam.transforms.enrichment.cross_join).\n",
+ "* `timeout`: The number of seconds to wait for the request to be
completed by the API before timing out. Defaults to 30 seconds.\n",
+ "* `throttler`: Specifies the throttling mechanism. The only
supported option is default client-side adaptive throttling.\n",
+ "* `repeater`: Specifies the retry strategy when errors like
`TooManyRequests` and `TimeoutException` occur. Defaults to
[`ExponentialBackOffRepeater`](https://beam.apache.org/releases/pydoc/current/apache_beam.io.requestresponse.html#apache_beam.io.requestresponse.ExponentialBackOffRepeater).\n",
+ "\n",
+ "\n",
+ "To use the Redis cache, apply the `with_redis_cache` hook to the
enrichment transform. The coders for encoding and decoding the input and output
for the cache are optional and are internally inferred."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "xJTCfSmiV1kv"
+ },
+ "source": [
+ "The following example demonstrates the code needed to add this
transform to your pipeline.\n",
+ "\n",
+ "\n",
+ "```\n",
+ "with beam.Pipeline() as p:\n",
+ " output = (p\n",
+ " ...\n",
+ " | \"Enrich with Vertex AI\" >>
Enrichment(vertex_ai_handler)\n",
+ " | \"RunInference\" >> RunInference(model_handler)\n",
+ " ...\n",
+ " )\n",
+ "```\n",
+ "\n",
+ "\n",
+ "\n",
+ "\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "F-xjiP_pHWZr"
+ },
+ "source": [
+ "To make a prediction, use the following fields: `product_id`,
`quantity`, `price`, `customer_id`, and `customer_location`. Retrieve the value
of the `customer_location` field from Bigtable.\n",
+ "\n",
+ "The enrichment transform performs a
[`cross_join`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.enrichment.html#apache_beam.transforms.enrichment.cross_join)
by default."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "CX9Cqybu6scV"
+ },
+ "source": [
+ "## Use the `VertexAIModelHandlerJSON` interface to run inference\n",
+ "\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "zy5Jl7_gLklX"
+ },
+ "source": [
+ "Because the enrichment transform outputs data in the format
`beam.Row`, in order to align it with the
[`VertexAIModelHandlerJSON`](https://beam.apache.org/releases/pydoc/current/apache_beam.ml.inference.vertex_ai_inference.html#apache_beam.ml.inference.vertex_ai_inference.VertexAIModelHandlerJSON)
interface, convert the output into a list of `tensorflow.tensor`. Some
enriched fields are of `string` type. For tensor creation, all values must be
of the same type. Therefore, conver [...]
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "KBKoB06nL4LF"
+ },
+ "outputs": [],
+ "source": [
+ "def convert_row_to_tensor(element: beam.Row):\n",
+ " element_dict = element._asdict()\n",
+ " row = list(element_dict.values())\n",
+ " for i, r in enumerate(row):\n",
+ " if isinstance(r, str):\n",
+ " row[i] = int(r)\n",
+ " return tf.convert_to_tensor(row,
dtype=tf.float32).numpy().tolist()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "-tGHyB_vL3rJ"
+ },
+ "source": [
+ "Initialize the model handler with the preprocessing function."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "VqUUEwcU-r2e"
+ },
+ "outputs": [],
+ "source": [
+ "model_handler =
VertexAIModelHandlerJSON(endpoint_id=model_endpoint_id,\n",
+ " project=PROJECT_ID,\n",
+ " location=LOCATION,\n",
+ "
).with_preprocess_fn(convert_row_to_tensor)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "vNHI4gVgNec2"
+ },
+ "source": [
+ "Define a `DoFn` to format the output."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "rkN-_Yf4Nlwy"
+ },
+ "outputs": [],
+ "source": [
+ "class PostProcessor(beam.DoFn):\n",
+ " def process(self, element, *args, **kwargs):\n",
+ " print('Customer %d who bought product %d is recommended to buy
product %d' % (element.example[0], element.example[1],
math.ceil(element.inference[0])))"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "0a1zerXycQ0z"
+ },
+ "source": [
+ "## Run the pipeline\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "WrwY0_gV_IDK"
+ },
+ "source": [
+ "Configure the pipeline to run in streaming mode."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "t0425sYBsYtB"
+ },
+ "outputs": [],
+ "source": [
+ "options = pipeline_options.PipelineOptions()\n",
+ "options.view_as(pipeline_options.StandardOptions).streaming = True #
Streaming mode is set to True"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "DBNijQDY_dRe"
+ },
+ "source": [
+ "Pub/Sub sends the data in bytes. Convert the data to `beam.Row`
objects by using a `DoFn`."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "sRw9iL8pKP5O"
+ },
+ "outputs": [],
+ "source": [
+ "class DecodeBytes(beam.DoFn):\n",
+ " \"\"\"\n",
+ " The DecodeBytes `DoFn` converts the data read from Pub/Sub to
`beam.Row`.\n",
+ " First, decode the encoded string. Convert the output to\n",
+ " a `dict` with `json.loads()`, which is used to create a
`beam.Row`.\n",
+ " \"\"\"\n",
+ " def process(self, element, *args, **kwargs):\n",
+ " element_dict = json.loads(element.decode('utf-8'))\n",
+ " yield beam.Row(**element_dict)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "xofUJym-_GuB"
+ },
+ "source": [
+ "Use the following code to run the pipeline.\n",
+ "\n",
+ "**Note:** Because this pipeline is a streaming pipeline, you need to
manually stop the cell. If you don't stop the cell, the pipeline continues to
run."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";,
+ "height": 671
+ },
+ "id": "St07XoibcQSb",
+ "outputId": "0ca70756-6a69-4d63-9ab7-8814ae6adf05"
+ },
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "Customer 25005 who bought product 14235 is recommended to buy
product 8944\n",
+ "Customer 62544 who bought product 14235 is recommended to buy
product 23313\n",
+ "Customer 17228 who bought product 14235 is recommended to buy
product 6600\n",
+ "Customer 54015 who bought product 14235 is recommended to buy
product 19682\n",
+ "Customer 16569 who bought product 14235 is recommended to buy
product 6441\n"
+ ]
+ }
+ ],
+ "source": [
+ "with beam.Pipeline(options=options) as p:\n",
+ " _ = (p\n",
+ " | \"Read from Pub/Sub\" >>
beam.io.ReadFromPubSub(subscription=subscription_path)\n",
+ " | \"ConvertToRow\" >> beam.ParDo(DecodeBytes())\n",
+ " | \"Enrichment\" >> Enrichment(vertex_ai_handler)\n",
+ " | \"RunInference\" >> RunInference(model_handler)\n",
+ " | \"Format Output\" >> beam.ParDo(PostProcessor())\n",
+ " )"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {
+ "id": "yDjkq2VI7fuM"
+ },
+ "source": [
+ "## Clean up resources"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/";
+ },
+ "id": "UiPb_kzv7pCu",
+ "outputId": "7902c30f-4db0-431b-9dd8-b647b3cb34da"
+ },
+ "outputs": [
+ {
+ "data": {
+ "text/plain": [
+ "<google.api_core.operation.Operation at 0x7b0e1a2843d0>"
+ ]
+ },
+ "execution_count": 8,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "# Delete feature views.\n",
+ "admin_client.delete_feature_view(\n",
+ "
name=f\"projects/{PROJECT_ID}/locations/{LOCATION}/featureOnlineStores/{feature_store_name}/featureViews/{feature_view_name}\"\n",
+ ")\n",
+ "\n",
+ "# Delete online store instance.\n",
+ "admin_client.delete_feature_online_store(\n",
+ "
name=f\"projects/{PROJECT_ID}/locations/{LOCATION}/featureOnlineStores/{feature_store_name}\",\n",
+ " force=True,\n",
+ ")"
+ ]
+ }
+ ],
+ "metadata": {
+ "colab": {
+ "provenance": []
+ },
+ "kernelspec": {
+ "display_name": "Python 3",
+ "name": "python3"
+ },
+ "language_info": {
+ "name": "python"
+ }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 0
+}
![https://raw.githubusercontent.com/google/or-tools/main/tools/colab_32px.png\"](https://raw.githubusercontent.com/google/or-tools/main/tools/colab_32px.png\"){kind=link}
![https://raw.githubusercontent.com/google/or-tools/main/tools/github_32px.png\"](https://raw.githubusercontent.com/google/or-tools/main/tools/github_32px.png\"){kind=link}