<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 <table align="left">
  <td>
    <a href="https
Step1:  Restart the kernel
After you install the SDK, you need to restart the notebook kernel so it can find the packages. You can restart kernel from Kernel -> Restart Kernel, or running the following
Step2:  Before you begin
Select a GPU runtime
Make sure you're running this notebook in a GPU runtime if you have that option. In Colab, select "Runtime --> Change runtime type > GPU"
Set up your Google Cloud project
The following steps are required, regardless of your notebook environment.
Select or create a Google Cloud project. When you first create an account, you get a $300 free credit towards your compute/storage costs.
Make sure that billing is enabled for your project.
Enable the Vertex AI API and Compute Engine API.
If you are running this notebook locally, you will need to install the Cloud SDK.
Enter your project ID in the cell below. Then run the cell to make sure the
Cloud SDK uses the right project for all the commands in this notebook.
Note
Step3:  Otherwise, set your project ID here.
Step4:  Authenticate your Google Cloud account
If you are using Google Cloud Notebooks, your environment is already
authenticated. Skip this step.
If you are using Colab, run the cell below and follow the instructions
when prompted to authenticate your account via oAuth.
Otherwise, follow these steps
Step5:  Prepare for output
Step 1. Create dataset for output
You need a BigQuery dataset to host the output data in us-central1. Input the name of the dataset you want to created and specify the name of the table you want to store the output later. These will be used later in the notebook.
Make sure that the table name does NOT already exist.
Step6:  Import libraries and define constants
Step7:  Terminology and Concept
Featurestore Data model
Vertex AI Feature Store organizes data with the following 3 important hierarchical concepts
Step8:  You can use GetFeaturestore or ListFeaturestores to check if the Featurestore was successfully created. The following example gets the details of the Featurestore.
Step9:  Auto scaling is available in v1beta1 since v1.11. Below is the example for the CreateFeaturestoreRequest with auto-scaling, use it with aiplatform_v1beta1.FeaturestoreServiceClient to create Featurestore
Step10:  Create Entity Type
Step11:  Feature monitoring is in preview, so you need to use v1beta1 Python. The easiest way to set this for now is using console UI. For completeness, below is example to do this using v1beta1 SDK.
You can view monitoring statistics on console UI.
Step12:  Create Feature
Step13:  Search created features
While the ListFeatures method allows you to easily view all features of a single
entity type, the SearchFeatures method searches across all featurestores
and entity types in a given location (such as us-central1). This can help you discover features that were created by someone else.
You can query based on feature properties including feature ID, entity type ID,
and feature description. You can also limit results by filtering on a specific
featurestore, feature value type, and/or labels.
Step14:  Now, narrow down the search to features that are of type DOUBLE
Step15:  Or, limit the search results to features with specific keywords in their ID and type.
Step16:  Import Feature Values
You need to import feature values before you can use them for online/offline serving. In this step, you will learn how to import feature values by calling the ImportFeatureValues API using the Python SDK.
Source Data Format and Layout
As mentioned above, BigQuery table/Avro/CSV are supported. No matter what format you are using, each imported entity must have an ID; also, each entity can optionally have a timestamp, sepecifying when the feature values are generated. This Colab uses Avro as an input, located at this public bucket. The Avro schemas are as follows
Step17:  Import feature values for Movies
Similarly, import feature values for 'movies' into the featurestore.
Step18:  Online serving
The
Online Serving APIs
lets you serve feature values for small batches of entities. It's designed for latency-sensitive service, such as online model prediction. For example, for a movie service, you might want to quickly shows movies that the current user would most likely watch by using online predictions.
Read one entity per request
The ReadFeatureValues API is used to read feature values of one entity; hence
its custom HTTP verb is readFeatureValues. By default, the API will return the  latest value of each feature, meaning the feature values with the most recent  timestamp.
To read feature values, specify the entity ID and features to read. The response
contains a header and an entity_view. Each row of data in the entity_view
contains one feature value, in the same order of features as listed in the response header.
Step19:  Read multiple entities per request
To read feature values from multiple entities, use the
StreamingReadFeatureValues API, which is almost identical to the previous
ReadFeatureValues API. Note that fetching only a small number of entities is recomended when using this API due to its latency-sensitive nature.
Step20:  Now that you have learned how to fetch imported feature values for online serving, the next step is learning how to use imported feature values for offline use cases.
Batch Serving
Batch Serving is used to fetch a large batch of feature values for high-throughput, typically for training a model or batch prediction. In this section, you will learn how to prepare for training examples by calling the BatchReadFeatureValues API.
Use case
The task is to prepare a training dataset to train a model, which predicts if a given user will watch a given movie. To achieve this, you need 2 sets of input
Step21:  After the LRO finishes, you should be able to see the result from the BigQuery console, in the dataset created earlier.
Cleaning up
To clean up all Google Cloud resources used in this project, you can delete the Google Cloud
project you used for the tutorial.
You can also keep the project but delete the featurestore
<ASSISTANT_TASK:>
Python Code:
import os
# The Google Cloud Notebook product has specific requirements
IS_GOOGLE_CLOUD_NOTEBOOK = os.path.exists("/opt/deeplearning/metadata/env_version")
# Google Cloud Notebook requires dependencies to be installed with '--user'
USER_FLAG = ""
if IS_GOOGLE_CLOUD_NOTEBOOK:
    USER_FLAG = "--user"
! pip install {USER_FLAG} --upgrade google-cloud-aiplatform
Explanation: <table align="left">
  <td>
    <a href="https://colab.research.google.com/github/GoogleCloudPlatform/vertex-ai-samples/blob/master/notebooks/official/feature_store/gapic-feature-store.ipynb"">
      <img src="https://cloud.google.com/ml-engine/images/colab-logo-32px.png" alt="Colab logo"> Run in Colab
    </a>
  </td>
  <td>
    <a href="https://github.com/GoogleCloudPlatform/vertex-ai-samples/blob/master/notebooks/official/feature_store/gapic-feature-store.ipynb">
      <img src="https://cloud.google.com/ml-engine/images/github-logo-32px.png" alt="GitHub logo">
      View on GitHub
    </a>
  </td>
</table>
Overview
This Colab introduces Vertex AI Feature Store, a managed cloud service for machine learning engineers and data scientists to store, serve, manage and share machine learning features at a large scale.
This Colab assumes that you understand basic Google Cloud concepts such as Project, Storage and Vertex AI. Some machine learning knowledge is also helpful but not required.
Dataset
This Colab uses a movie recommendation dataset as an example throughout all the sessions. The task is to train a model to predict if a user is going to watch a movie and serve this model online. 
Objective
In this notebook, you will learn how to:
* How to import your features into Vertex AI Feature Store.
* How to serve online prediction requests using the imported features.
* How to access imported features in offline jobs, such as training jobs.
Costs
This tutorial uses billable components of Google Cloud:
Vertex AI
Cloud Storage
Cloud Bigtable
Learn about Vertex AI
pricing and Cloud Storage
pricing, and use the Pricing
Calculator
to generate a cost estimate based on your projected usage.
Set up your local development environment
If you are using Colab or Google Cloud Notebooks, your environment already meets
all the requirements to run this notebook. You can skip this step.
Otherwise, make sure your environment meets this notebook's requirements.
You need the following:
The Google Cloud SDK
Git
Python 3
virtualenv
Jupyter notebook running in a virtual environment with Python 3
The Google Cloud guide to Setting up a Python development
environment and the Jupyter
installation guide provide detailed instructions
for meeting these requirements. The following steps provide a condensed set of
instructions:
Install and initialize the Cloud SDK.
Install Python 3.
Install
   virtualenv
   and create a virtual environment that uses Python 3. Activate the virtual environment.
To install Jupyter, run pip install jupyter on the
command-line in a terminal shell.
To launch Jupyter, run jupyter notebook on the command-line in a terminal shell.
Open this notebook in the Jupyter Notebook Dashboard.
Install additional packages
For this Colab, you need the Vertex SDK for Python.
End of explanation
# Automatically restart kernel after installs
import os
if not os.getenv("IS_TESTING"):
    # Automatically restart kernel after installs
    import IPython
    app = IPython.Application.instance()
    app.kernel.do_shutdown(True)
Explanation: Restart the kernel
After you install the SDK, you need to restart the notebook kernel so it can find the packages. You can restart kernel from Kernel -> Restart Kernel, or running the following:
End of explanation
import os
PROJECT_ID = ""
# Get your Google Cloud project ID from gcloud
if not os.getenv("IS_TESTING"):
    shell_output = !gcloud config list --format 'value(core.project)' 2>/dev/null
    PROJECT_ID = shell_output[0]
    print("Project ID: ", PROJECT_ID)
Explanation: Before you begin
Select a GPU runtime
Make sure you're running this notebook in a GPU runtime if you have that option. In Colab, select "Runtime --> Change runtime type > GPU"
Set up your Google Cloud project
The following steps are required, regardless of your notebook environment.
Select or create a Google Cloud project. When you first create an account, you get a $300 free credit towards your compute/storage costs.
Make sure that billing is enabled for your project.
Enable the Vertex AI API and Compute Engine API.
If you are running this notebook locally, you will need to install the Cloud SDK.
Enter your project ID in the cell below. Then run the cell to make sure the
Cloud SDK uses the right project for all the commands in this notebook.
Note: Jupyter runs lines prefixed with ! as shell commands, and it interpolates Python variables prefixed with $ into these commands.
Set your project ID
If you don't know your project ID, you may be able to get your project ID using gcloud.
End of explanation
if PROJECT_ID == "" or PROJECT_ID is None:
    PROJECT_ID = "python-docs-samples-tests"  # @param {type:"string"}
Explanation: Otherwise, set your project ID here.
End of explanation
import os
import sys
# If you are running this notebook in Colab, run this cell and follow the
# instructions to authenticate your GCP account. This provides access to your
# Cloud Storage bucket and lets you submit training jobs and prediction
# requests.
# The Google Cloud Notebook product has specific requirements
IS_GOOGLE_CLOUD_NOTEBOOK = os.path.exists("/opt/deeplearning/metadata/env_version")
# If on Google Cloud Notebooks, then don't execute this code
if not IS_GOOGLE_CLOUD_NOTEBOOK:
    if "google.colab" in sys.modules:
        from google.colab import auth as google_auth
        google_auth.authenticate_user()
    # If you are running this notebook locally, replace the string below with the
    # path to your service account key and run this cell to authenticate your GCP
    # account.
    elif not os.getenv("IS_TESTING"):
        %env GOOGLE_APPLICATION_CREDENTIALS ''
Explanation: Authenticate your Google Cloud account
If you are using Google Cloud Notebooks, your environment is already
authenticated. Skip this step.
If you are using Colab, run the cell below and follow the instructions
when prompted to authenticate your account via oAuth.
Otherwise, follow these steps:
In the Cloud Console, go to the Create service account key
   page.
Click Create service account.
In the Service account name field, enter a name, and
   click Create.
In the Grant this service account access to project section, click the Role drop-down list. Type "Vertex AI"
into the filter box, and select
   Vertex AI Administrator. Type "Storage Object Admin" into the filter box, and select Storage Object Admin.
Click Create. A JSON file that contains your key downloads to your
local environment.
Enter the path to your service account key as the
GOOGLE_APPLICATION_CREDENTIALS variable in the cell below and run the cell.
End of explanation
from datetime import datetime
from google.cloud import bigquery
# Output dataset
DESTINATION_DATA_SET = "movie_predictions"  # @param {type:"string"}
TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")
DESTINATION_DATA_SET = "{prefix}_{timestamp}".format(
    prefix=DESTINATION_DATA_SET, timestamp=TIMESTAMP
)
# Output table. Make sure that the table does NOT already exist; the BatchReadFeatureValues API cannot overwrite an existing table
DESTINATION_TABLE_NAME = "training_data"  # @param {type:"string"}
DESTINATION_PATTERN = "bq://{project}.{dataset}.{table}"
DESTINATION_TABLE_URI = DESTINATION_PATTERN.format(
    project=PROJECT_ID, dataset=DESTINATION_DATA_SET, table=DESTINATION_TABLE_NAME
)
# Create dataset
REGION = "us-central1"  # @param {type:"string"}
client = bigquery.Client(project=PROJECT_ID)
dataset_id = "{}.{}".format(client.project, DESTINATION_DATA_SET)
dataset = bigquery.Dataset(dataset_id)
dataset.location = REGION
dataset = client.create_dataset(dataset)
print("Created dataset {}.{}".format(client.project, dataset.dataset_id))
Explanation: Prepare for output
Step 1. Create dataset for output
You need a BigQuery dataset to host the output data in us-central1. Input the name of the dataset you want to created and specify the name of the table you want to store the output later. These will be used later in the notebook.
Make sure that the table name does NOT already exist.
End of explanation
# Other than project ID and featurestore ID and endpoints needs to be set
API_ENDPOINT = "us-central1-aiplatform.googleapis.com"  # @param {type:"string"}
INPUT_CSV_FILE = "gs://cloud-samples-data-us-central1/vertex-ai/feature-store/datasets/movie_prediction.csv"
from google.cloud.aiplatform_v1 import (FeaturestoreOnlineServingServiceClient,
                                        FeaturestoreServiceClient)
from google.cloud.aiplatform_v1.types import FeatureSelector, IdMatcher
from google.cloud.aiplatform_v1.types import entity_type as entity_type_pb2
from google.cloud.aiplatform_v1.types import feature as feature_pb2
from google.cloud.aiplatform_v1.types import featurestore as featurestore_pb2
from google.cloud.aiplatform_v1.types import \
    featurestore_online_service as featurestore_online_service_pb2
from google.cloud.aiplatform_v1.types import \
    featurestore_service as featurestore_service_pb2
from google.cloud.aiplatform_v1.types import io as io_pb2
from google.protobuf.duration_pb2 import Duration
# Create admin_client for CRUD and data_client for reading feature values.
admin_client = FeaturestoreServiceClient(client_options={"api_endpoint": API_ENDPOINT})
data_client = FeaturestoreOnlineServingServiceClient(
    client_options={"api_endpoint": API_ENDPOINT}
)
# Represents featurestore resource path.
BASE_RESOURCE_PATH = admin_client.common_location_path(PROJECT_ID, REGION)
Explanation: Import libraries and define constants
End of explanation
FEATURESTORE_ID = "movie_prediction"
try:
    create_lro = admin_client.create_featurestore(
        featurestore_service_pb2.CreateFeaturestoreRequest(
            parent=BASE_RESOURCE_PATH,
            featurestore_id=FEATURESTORE_ID,
            featurestore=featurestore_pb2.Featurestore(
                online_serving_config=featurestore_pb2.Featurestore.OnlineServingConfig(
                    fixed_node_count=1
                ),
            ),
        )
    )
    # Wait for LRO to finish and get the LRO result.
    print(create_lro.result())
except Exception as e:
    print(e)
Explanation: Terminology and Concept
Featurestore Data model
Vertex AI Feature Store organizes data with the following 3 important hierarchical concepts:
Featurestore -&gt; EntityType -&gt; Feature
* Featurestore: the place to store your features
* EntityType: under a Featurestore, an EntityType describes an object to be modeled, real one or virtual one.
* Feature: under an EntityType, a feature describes an attribute of the EntityType
In the movie prediction example, you will create a featurestore called movie_prediction. This store has 2 entity types: Users and Movies. The Users entity type has the age, gender, and like genres features. The Movies entity type has the genres and average rating features.
Create Featurestore and Define Schemas
Create Featurestore
The method to create a featurestore returns a
long-running operation (LRO). An LRO starts an asynchronous job. LROs are returned for other API
methods too, such as updating or deleting a featurestore. Calling
create_fs_lro.result() waits for the LRO to complete.
End of explanation
admin_client.get_featurestore(
    name=admin_client.featurestore_path(PROJECT_ID, REGION, FEATURESTORE_ID)
)
Explanation: You can use GetFeaturestore or ListFeaturestores to check if the Featurestore was successfully created. The following example gets the details of the Featurestore.
End of explanation
from google.cloud.aiplatform_v1beta1.types import \
    featurestore as v1beta1_featurestore_pb2
from google.cloud.aiplatform_v1beta1.types import \
    featurestore_service as v1beta1_featurestore_service_pb2
create_featurestore_request = v1beta1_featurestore_service_pb2.CreateFeaturestoreRequest(
    parent=BASE_RESOURCE_PATH,
    featurestore_id=FEATURESTORE_ID,
    featurestore=v1beta1_featurestore_pb2.Featurestore(
        online_serving_config=v1beta1_featurestore_pb2.Featurestore.OnlineServingConfig(
            scaling=v1beta1_featurestore_pb2.Featurestore.OnlineServingConfig.Scaling(
                min_node_count=1, max_node_count=5
            )
        ),
    ),
)
Explanation: Auto scaling is available in v1beta1 since v1.11. Below is the example for the CreateFeaturestoreRequest with auto-scaling, use it with aiplatform_v1beta1.FeaturestoreServiceClient to create Featurestore:
End of explanation
try:
    users_entity_type_lro = admin_client.create_entity_type(
        featurestore_service_pb2.CreateEntityTypeRequest(
            parent=admin_client.featurestore_path(PROJECT_ID, REGION, FEATURESTORE_ID),
            entity_type_id="users",
            entity_type=entity_type_pb2.EntityType(
                description="Users entity",
            ),
        )
    )
    # Similarly, wait for EntityType creation operation.
    print(users_entity_type_lro.result())
except Exception as e:
    print(e)
# Create movies entity type without a monitoring configuration.
try:
    movies_entity_type_lro = admin_client.create_entity_type(
        featurestore_service_pb2.CreateEntityTypeRequest(
            parent=admin_client.featurestore_path(PROJECT_ID, REGION, FEATURESTORE_ID),
            entity_type_id="movies",
            entity_type=entity_type_pb2.EntityType(description="Movies entity"),
        )
    )
    # Similarly, wait for EntityType creation operation.
    print(movies_entity_type_lro.result())
except Exception as e:
    print(e)
Explanation: Create Entity Type
End of explanation
from google.cloud.aiplatform_v1beta1 import \
    FeaturestoreServiceClient as v1beta1_FeaturestoreServiceClient
from google.cloud.aiplatform_v1beta1.types import \
    entity_type as v1beta1_entity_type_pb2
from google.cloud.aiplatform_v1beta1.types import \
    featurestore_monitoring as v1beta1_featurestore_monitoring_pb2
from google.cloud.aiplatform_v1beta1.types import \
    featurestore_service as v1beta1_featurestore_service_pb2
v1beta1_admin_client = v1beta1_FeaturestoreServiceClient(
    client_options={"api_endpoint": API_ENDPOINT}
)
# Enable monitoring for users entity type.
# All Features belonging to this EntityType will by default inherit the monitoring config.
v1beta1_admin_client.update_entity_type(
    v1beta1_featurestore_service_pb2.UpdateEntityTypeRequest(
        entity_type=v1beta1_entity_type_pb2.EntityType(
            name=admin_client.entity_type_path(
                PROJECT_ID, REGION, FEATURESTORE_ID, "users"
            ),
            monitoring_config=v1beta1_featurestore_monitoring_pb2.FeaturestoreMonitoringConfig(
                snapshot_analysis=v1beta1_featurestore_monitoring_pb2.FeaturestoreMonitoringConfig.SnapshotAnalysis(
                    monitoring_interval=Duration(seconds=86400),  # 1 day
                ),
            ),
        ),
    )
)
Explanation: Feature monitoring is in preview, so you need to use v1beta1 Python. The easiest way to set this for now is using console UI. For completeness, below is example to do this using v1beta1 SDK.
You can view monitoring statistics on console UI.
End of explanation
# Create features for the 'users' entity.
try:
    admin_client.batch_create_features(
        parent=admin_client.entity_type_path(
            PROJECT_ID, REGION, FEATURESTORE_ID, "users"
        ),
        requests=[
            featurestore_service_pb2.CreateFeatureRequest(
                feature=feature_pb2.Feature(
                    value_type=feature_pb2.Feature.ValueType.INT64,
                    description="User age",
                ),
                feature_id="age",
            ),
            featurestore_service_pb2.CreateFeatureRequest(
                feature=feature_pb2.Feature(
                    value_type=feature_pb2.Feature.ValueType.STRING,
                    description="User gender",
                ),
                feature_id="gender",
            ),
            featurestore_service_pb2.CreateFeatureRequest(
                feature=feature_pb2.Feature(
                    value_type=feature_pb2.Feature.ValueType.STRING_ARRAY,
                    description="An array of genres that this user liked",
                ),
                feature_id="liked_genres",
            ),
        ],
    ).result()
except Exception as e:
    print(e)
# Create features for movies type.
try:
    admin_client.batch_create_features(
        parent=admin_client.entity_type_path(
            PROJECT_ID, REGION, FEATURESTORE_ID, "movies"
        ),
        requests=[
            featurestore_service_pb2.CreateFeatureRequest(
                feature=feature_pb2.Feature(
                    value_type=feature_pb2.Feature.ValueType.STRING,
                    description="The title of the movie",
                ),
                feature_id="title",
            ),
            featurestore_service_pb2.CreateFeatureRequest(
                feature=feature_pb2.Feature(
                    value_type=feature_pb2.Feature.ValueType.STRING,
                    description="The genres of the movie",
                ),
                feature_id="genres",
            ),
            featurestore_service_pb2.CreateFeatureRequest(
                feature=feature_pb2.Feature(
                    value_type=feature_pb2.Feature.ValueType.DOUBLE,
                    description="The average rating for the movie, range is [1.0-5.0]",
                ),
                feature_id="average_rating",
            ),
        ],
    ).result()
except Exception as e:
    print(e)
Explanation: Create Feature
End of explanation
# Search for all features across all featurestores.
list(admin_client.search_features(location=BASE_RESOURCE_PATH))
Explanation: Search created features
While the ListFeatures method allows you to easily view all features of a single
entity type, the SearchFeatures method searches across all featurestores
and entity types in a given location (such as us-central1). This can help you discover features that were created by someone else.
You can query based on feature properties including feature ID, entity type ID,
and feature description. You can also limit results by filtering on a specific
featurestore, feature value type, and/or labels.
End of explanation
# Search for all features with value type `DOUBLE`
list(
    admin_client.search_features(
        featurestore_service_pb2.SearchFeaturesRequest(
            location=BASE_RESOURCE_PATH, query="value_type=DOUBLE"
        )
    )
)
Explanation: Now, narrow down the search to features that are of type DOUBLE
End of explanation
# Filter on feature value type and keywords.
list(
    admin_client.search_features(
        featurestore_service_pb2.SearchFeaturesRequest(
            location=BASE_RESOURCE_PATH, query="feature_id:title AND value_type=STRING"
        )
    )
)
Explanation: Or, limit the search results to features with specific keywords in their ID and type.
End of explanation
import_users_request = featurestore_service_pb2.ImportFeatureValuesRequest(
    entity_type=admin_client.entity_type_path(
        PROJECT_ID, REGION, FEATURESTORE_ID, "users"
    ),
    avro_source=io_pb2.AvroSource(
        # Source
        gcs_source=io_pb2.GcsSource(
            uris=[
                "gs://cloud-samples-data-us-central1/vertex-ai/feature-store/datasets/users.avro"
            ]
        )
    ),
    entity_id_field="user_id",
    feature_specs=[
        # Features
        featurestore_service_pb2.ImportFeatureValuesRequest.FeatureSpec(id="age"),
        featurestore_service_pb2.ImportFeatureValuesRequest.FeatureSpec(id="gender"),
        featurestore_service_pb2.ImportFeatureValuesRequest.FeatureSpec(
            id="liked_genres"
        ),
    ],
    feature_time_field="update_time",
    worker_count=1,
)
# Start to import, will take a couple of minutes
ingestion_lro = admin_client.import_feature_values(import_users_request)
# Polls for the LRO status and prints when the LRO has completed
ingestion_lro.result()
Explanation: Import Feature Values
You need to import feature values before you can use them for online/offline serving. In this step, you will learn how to import feature values by calling the ImportFeatureValues API using the Python SDK.
Source Data Format and Layout
As mentioned above, BigQuery table/Avro/CSV are supported. No matter what format you are using, each imported entity must have an ID; also, each entity can optionally have a timestamp, sepecifying when the feature values are generated. This Colab uses Avro as an input, located at this public bucket. The Avro schemas are as follows:
For the Users entity:
schema = {
  "type": "record",
  "name": "User",
  "fields": [
      {
       "name":"user_id",
       "type":["null","string"]
      },
      {
       "name":"age",
       "type":["null","long"]
      },
      {
       "name":"gender",
       "type":["null","string"]
      },
      {
       "name":"liked_genres",
       "type":{"type":"array","items":"string"}
      },
      {
       "name":"update_time",
       "type":["null",{"type":"long","logicalType":"timestamp-micros"}]
      },
  ]
 }
For the Movies entity
schema = {
 "type": "record",
 "name": "Movie",
 "fields": [
     {
      "name":"movie_id",
      "type":["null","string"]
     },
     {
      "name":"average_rating",
      "type":["null","double"]
     },
     {
      "name":"title",
      "type":["null","string"]
     },
     {
      "name":"genres",
      "type":["null","string"]
     },
     {
      "name":"update_time",
      "type":["null",{"type":"long","logicalType":"timestamp-micros"}]
     },
 ]
}
Import feature values for Users
When importing, specify the following in your request:
Data source format: BigQuery Table/Avro/CSV
Data source URL
Destination: featurestore/entity types/features to be imported
End of explanation
import_movie_request = featurestore_service_pb2.ImportFeatureValuesRequest(
    entity_type=admin_client.entity_type_path(
        PROJECT_ID, REGION, FEATURESTORE_ID, "movies"
    ),
    avro_source=io_pb2.AvroSource(
        gcs_source=io_pb2.GcsSource(
            uris=[
                "gs://cloud-samples-data-us-central1/vertex-ai/feature-store/datasets/movies.avro"
            ]
        )
    ),
    entity_id_field="movie_id",
    feature_specs=[
        featurestore_service_pb2.ImportFeatureValuesRequest.FeatureSpec(id="title"),
        featurestore_service_pb2.ImportFeatureValuesRequest.FeatureSpec(id="genres"),
        featurestore_service_pb2.ImportFeatureValuesRequest.FeatureSpec(
            id="average_rating"
        ),
    ],
    feature_time_field="update_time",
    worker_count=1,
)
# Start to import, will take a couple of minutes
ingestion_lro = admin_client.import_feature_values(import_movie_request)
# Polls for the LRO status and prints when the LRO has completed
ingestion_lro.result()
Explanation: Import feature values for Movies
Similarly, import feature values for 'movies' into the featurestore.
End of explanation
# Fetch the following 3 features.
feature_selector = FeatureSelector(
    id_matcher=IdMatcher(ids=["age", "gender", "liked_genres"])
)
data_client.read_feature_values(
    featurestore_online_service_pb2.ReadFeatureValuesRequest(
        # Fetch from the following feature store/entity type
        entity_type=admin_client.entity_type_path(
            PROJECT_ID, REGION, FEATURESTORE_ID, "users"
        ),
        # Fetch the user features whose ID is "alice"
        entity_id="alice",
        feature_selector=feature_selector,
    )
)
Explanation: Online serving
The
Online Serving APIs
lets you serve feature values for small batches of entities. It's designed for latency-sensitive service, such as online model prediction. For example, for a movie service, you might want to quickly shows movies that the current user would most likely watch by using online predictions.
Read one entity per request
The ReadFeatureValues API is used to read feature values of one entity; hence
its custom HTTP verb is readFeatureValues. By default, the API will return the  latest value of each feature, meaning the feature values with the most recent  timestamp.
To read feature values, specify the entity ID and features to read. The response
contains a header and an entity_view. Each row of data in the entity_view
contains one feature value, in the same order of features as listed in the response header.
End of explanation
# Read the same set of features as above, but for multiple entities.
response_stream = data_client.streaming_read_feature_values(
    featurestore_online_service_pb2.StreamingReadFeatureValuesRequest(
        entity_type=admin_client.entity_type_path(
            PROJECT_ID, REGION, FEATURESTORE_ID, "users"
        ),
        entity_ids=["alice", "bob"],
        feature_selector=feature_selector,
    )
)
# Iterate and process response. Note the first one is always the header only.
for response in response_stream:
    print(response)
Explanation: Read multiple entities per request
To read feature values from multiple entities, use the
StreamingReadFeatureValues API, which is almost identical to the previous
ReadFeatureValues API. Note that fetching only a small number of entities is recomended when using this API due to its latency-sensitive nature.
End of explanation
batch_serving_request = featurestore_service_pb2.BatchReadFeatureValuesRequest(
    # featurestore info
    featurestore=admin_client.featurestore_path(PROJECT_ID, REGION, FEATURESTORE_ID),
    # URL for the label data, i.e., Table 1.
    csv_read_instances=io_pb2.CsvSource(
        gcs_source=io_pb2.GcsSource(uris=[INPUT_CSV_FILE])
    ),
    destination=featurestore_service_pb2.FeatureValueDestination(
        bigquery_destination=io_pb2.BigQueryDestination(
            # Output to BigQuery table created earlier
            output_uri=DESTINATION_TABLE_URI
        )
    ),
    entity_type_specs=[
        featurestore_service_pb2.BatchReadFeatureValuesRequest.EntityTypeSpec(
            # Read the 'age', 'gender' and 'liked_genres' features from the 'users' entity
            entity_type_id="users",
            feature_selector=FeatureSelector(
                id_matcher=IdMatcher(
                    ids=[
                        # features, use "*" if you want to select all features within this entity type
                        "age",
                        "gender",
                        "liked_genres",
                    ]
                )
            ),
        ),
        featurestore_service_pb2.BatchReadFeatureValuesRequest.EntityTypeSpec(
            # Read the 'average_rating' and 'genres' feature values of the 'movies' entity
            entity_type_id="movies",
            feature_selector=FeatureSelector(
                id_matcher=IdMatcher(ids=["average_rating", "genres"])
            ),
        ),
    ],
)
# Execute the batch read
batch_serving_lro = admin_client.batch_read_feature_values(batch_serving_request)
# This long runing operation will poll until the batch read finishes.
batch_serving_lro.result()
Explanation: Now that you have learned how to fetch imported feature values for online serving, the next step is learning how to use imported feature values for offline use cases.
Batch Serving
Batch Serving is used to fetch a large batch of feature values for high-throughput, typically for training a model or batch prediction. In this section, you will learn how to prepare for training examples by calling the BatchReadFeatureValues API.
Use case
The task is to prepare a training dataset to train a model, which predicts if a given user will watch a given movie. To achieve this, you need 2 sets of input:
Features: you already imported into the featurestore.
Labels: the groud-truth data recorded that user X has watched movie Y.
To be more specific, the ground-truth observation is described in Table 1 and the desired training dataset is described in Table 2. Each row in Table 2 is a result of joining the imported feature values from Vertex AI Feature Store according to the entity IDs and timestamps in Table 1. In this example,  the age, gender and liked_genres features from users and
the genres and average_rating features from movies are chosen to train the model. Note that only positive examples are shown in these 2 tables, i.e., you can imagine there is a label column whose values are all True.
BatchReadFeatureValues API takes Table 1 as
input, joins all required feature values from the featurestore, and returns Table 2 for training.
<h4 align="center">Table 1. Ground-truth Data</h4>
users | movies | timestamp          
----- | -------- | -------------------- 
alice  | Cinema Paradiso     | 2019-11-01T00:00:00Z 
bob  | The Shining     | 2019-11-15T18:09:43Z 
...   | ...      | ...     
<h4 align="center">Table 2. Expected Training Data Generated by Batch Read API (Positive Samples)</h4>
timestamp            | entity_type_users | age | gender | liked_genres | entity_type_movies | genres | average_rating
-------------------- | ----------------- | --------------- | ---------------- | -------------------- | -------- | --------- | ----- 
2019-11-01T00:00:00Z | bob              | 35        | M                | [Action, Crime]                 | The Shining | Horror | 4.8 
2019-11-01T00:00:00Z | alice             | 55        | F                | [Drama, Comedy]                 | Cinema Paradiso | Romance | 4.5 
... | ... | ... | ... | ... | ... | ... | ... 
Why timestamp?
Note that there is a timestamp column in Table 2. This indicates the time when the ground-truth was observed. This is to avoid data inconsistency.
For example, the 1st row of Table 2 indicates that user alice watched movie Cinema Paradiso on 2019-11-01T00:00:00Z. The featurestore keeps feature values for all timestamps but fetches feature values only at the given timestamp during batch serving. On 2019-11-01 alice might be 54 years old, but now alice might be 56; featurestore returns age=54 as alice's age, instead of age=56, because that is the value of the feature at the observation time. Similarly, other features might be time-variant as well, such as liked_genres.
Batch Read Feature Values
Assemble the request which specify the following info:
Where is the label data, i.e., Table 1.
Which features are read, i.e., the column names in Table 2.
The output is stored in a BigQuery table.
End of explanation
admin_client.delete_featurestore(
    request=featurestore_service_pb2.DeleteFeaturestoreRequest(
        name=admin_client.featurestore_path(PROJECT_ID, REGION, FEATURESTORE_ID),
        force=True,
    )
).result()
client.delete_dataset(
    DESTINATION_DATA_SET, delete_contents=True, not_found_ok=True
)  # Make an API request.
print("Deleted dataset '{}'.".format(DESTINATION_DATA_SET))
Explanation: After the LRO finishes, you should be able to see the result from the BigQuery console, in the dataset created earlier.
Cleaning up
To clean up all Google Cloud resources used in this project, you can delete the Google Cloud
project you used for the tutorial.
You can also keep the project but delete the featurestore:
End of explanation
<END_TASK>