Exemple : Déployer et interroger un endpoint de Feature Serving

Exemple étape par étape montrant comment déployer et query un Endpoint de Feature Serving à l'aide du Databricks SDK. Vous pouvez également utiliser l'interface utilisateur de Databricks ou l'API REST pour créer et interroger des endpoints de Feature Serving. Les Link vers cette documentation sont inclus.

Dans cet exemple, vous avez une table de villes avec leurs emplacements (latitude et longitude) et une application de recommandation qui prend en compte la distance actuelle de l'utilisateur par rapport à ces villes. Étant donné que l'emplacement de l'utilisateur change constamment, la distance entre l'utilisateur et chaque ville doit être calculée au moment de l'inférence. Ce tutoriel illustre comment effectuer ces calculs avec une faible latence en utilisant le Magasin de fonctionnalités en ligne Databricks et le Feature Serving Databricks. Pour l'ensemble complet du code d'exemple, consultez le notebook d'exemple.

Étape 1. Créer la table source

La table source contient des valeurs de fonctionnalités précalculées et peut être n'importe quelle table Delta dans Unity Catalog avec une clé primaire. Dans cet exemple, la table contient une liste de villes avec leur latitude et leur longitude. La clé primaire est destination_id. L'échantillon de données est affiché ci-dessous.

Nom	ID de destination (pk)	latitude	Longitude
Nashville, Tennessee	0	36,162663	-86,7816
Honolulu, Hawaï	1	21,309885	-157,85814
Las Vegas, Nevada	2	36,171562	-115,1391
New York, New York	3	40,712776	-74,005974

Nom	ID de destination (pk)	latitude	Longitude
Nashville, Tennessee	0	36,162663	-86,7816
Honolulu, Hawaï	1	21,309885	-157,85814
Las Vegas, Nevada	2	36,171562	-115,1391
New York, New York	3	40,712776	-74,005974

Étape 2. Créer le magasin de fonctionnalités en ligne.

Pour plus de détails sur les magasins de fonctionnalités en ligne Databricks, consultez Magasins de fonctionnalités en ligne Databricks.

Python

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

feature_table_name = f"{catalog_name}.{schema_name}.location_features"
function_name = f"{catalog_name}.{schema_name}.distance"


# Create the feature table
fe.create_table(
  name = feature_table_name,
  primary_keys="destination_id",
  df = destination_location_df,
  description = "Destination location features."
)


# Enable Change Data Feed to enable CONTINOUS and TRIGGERED publish modes

spark.sql(f"ALTER TABLE {feature_table_name} SET TBLPROPERTIES (delta.enableChangeDataFeed = 'true')")

# Create an online store with specified capacity
online_store_name = f"{username}-online-store"

fe.create_online_store(
    name=online_store_name,
    capacity="CU_2"  # Valid options: "CU_1", "CU_2", "CU_4", "CU_8"
)

# Wait until the state is AVAILABLE
online_store = fe.get_online_store(name=online_store_name)
online_store.state

# Publish the table

published_table = fe.publish_table(
    online_store=online_store,
    source_table_name=feature_table_name,
    online_table_name=online_table_name
)

Étape 3. Créez une fonction dans Unity Catalog

Dans cet exemple, la fonction calcule la distance entre la destination (dont l'emplacement ne change pas) et l'utilisateur (dont l'emplacement change fréquemment et n'est pas connu avant le moment de l'inférence).

Python

# Define the function. This function calculates the distance between two locations.
function_name = f"main.on_demand_demo.distance"

spark.sql(f"""
CREATE OR REPLACE FUNCTION {function_name}(latitude DOUBLE, longitude DOUBLE, user_latitude DOUBLE, user_longitude DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON AS
$$
import math
lat1 = math.radians(latitude)
lon1 = math.radians(longitude)
lat2 = math.radians(user_latitude)
lon2 = math.radians(user_longitude)

# Earth's radius in kilometers
radius = 6371

# Haversine formula
dlat = lat2 - lat1
dlon = lon2 - lon1
a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
distance = radius * c

return distance
$$""")

Étape 4. Créez une spécification de fonctionnalité dans Unity Catalog.

La spécification de fonctionnalités spécifie les fonctionnalités que l'Endpoint sert et leurs clés de recherche. Il spécifie également toutes les fonctions requises à appliquer aux fonctionnalités récupérées avec leurs liaisons. Pour plus de détails, consultez Créer un FeatureSpec.

Python

from databricks.feature_engineering import FeatureLookup, FeatureFunction, FeatureEngineeringClient

fe = FeatureEngineeringClient()

features=[
 FeatureLookup(
   table_name=feature_table_name,
   lookup_key="destination_id"
 ),
 FeatureFunction(
   udf_name=function_name,
   output_name="distance",
   input_bindings={
     "latitude": "latitude",
     "longitude": "longitude",
     "user_latitude": "user_latitude",
     "user_longitude": "user_longitude"
   },
 ),
]

feature_spec_name = f"main.on_demand_demo.travel_spec"

# The following code ignores errors raised if a feature_spec with the specified name already exists.
try:
 fe.create_feature_spec(name=feature_spec_name, features=features, exclude_columns=None)
except Exception as e:
 if "already exists" in str(e):
   pass
 else:
   raise e

Étape 5. Créez un endpoint de service de Feature Serving

Pour créer un Endpoint de Feature Serving, vous pouvez utiliser l'interface utilisateur Créer un Endpoint, l'API REST ou le SDK Databricks, comme indiqué ici.

Le endpoint de Feature Serving prend le feature_spec que vous avez créé à l'étape 4 comme parameter.

Python

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import EndpointCoreConfigInput, ServedEntityInput

# Create endpoint
endpoint_name = "fse-location"

try:
 status = workspace.serving_endpoints.create_and_wait(
   name=endpoint_name,
   config = EndpointCoreConfigInput(
     served_entities=[
       ServedEntityInput(
         entity_name=feature_spec_name,
         scale_to_zero_enabled=True,
         workload_size="Small"
       )
     ]
   )
 )
 print(status)

# Get the status of the endpoint
status = workspace.serving_endpoints.get(name=endpoint_name)
print(status)

Étape 6. Interroger l’Endpoint Feature Serving

Lorsque vous interrogez l’endpoint, vous fournissez la clé primaire et, en option, toutes les données contextuelles utilisées par la fonction. Dans cet exemple, la fonction prend en entrée la localisation actuelle de l'utilisateur (latitude et longitude). Étant donné que la localisation de l’utilisateur change constamment, elle doit être fournie à la fonction au moment de l’inférence en tant que fonctionnalité contextuelle.

Vous pouvez également query l'Endpoint en utilisant l'interface utilisateur Query an Endpoint using the UI ou la REST API.

Par simplicité, cet exemple calcule uniquement la distance par rapport à deux villes. Un scénario plus réaliste pourrait calculer la distance de l'utilisateur à partir de chaque emplacement dans la table de fonctionnalités pour déterminer quelles villes recommander.

Python

import mlflow.deployments

client = mlflow.deployments.get_deploy_client("databricks")
response = client.predict(
   endpoint=endpoint_name,
   inputs={
       "dataframe_records": [
           {"destination_id": 1, "user_latitude": 37, "user_longitude": -122},
           {"destination_id": 2, "user_latitude": 37, "user_longitude": -122},
       ]
   },
)

pprint(response)

Exemple de Notebook

Référez-vous à ce Notebook pour une illustration complète des étapes :

Notebook d'exemple de Feature Serving avec magasin en ligne.

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Informations supplémentaires

Pour plus de détails sur l'utilisation de l'API Python d'ingénierie des fonctionnalités, consultez la documentation de référence.