Esercitazione: Distribuire ed eseguire query su un endpoint di gestione delle funzionalità

Questo articolo illustra come distribuire ed eseguire query su un endpoint di gestione delle funzionalità in un processo dettagliato. Questo articolo usa Databricks SDK. Alcuni passaggi possono essere completati anche usando l'API REST o l'interfaccia utente di Databricks e includono riferimenti alla documentazione per tali metodi.

In questo esempio è presente una tabella delle città con le relative posizioni (latitudine e longitudine) e un'app consigliata che tiene conto della distanza corrente dell'utente da tali città. Poiché la posizione dell'utente cambia costantemente, la distanza tra l'utente e ogni città deve essere calcolata al momento dell'inferenza. Questa esercitazione illustra come eseguire questi calcoli con bassa latenza usando le tabelle online di Databricks e databricks Feature Serving. Per il set completo di codice di esempio, vedere il notebook di esempio.

Passaggio 1: Creare la tabella di origine

La tabella di origine contiene valori di funzionalità precompilate e può essere qualsiasi tabella Delta nel catalogo Unity con una chiave primaria. In questo esempio la tabella contiene un elenco di città con la loro latitudine e longitudine. La chiave primaria è destination_id. Di seguito sono riportati i dati di esempio.

name	destination_id (pk)	latitude	longitude
Nashville, Tennessee	0	36.162663	-86.7816
Honolulu, Hawaii	1	21.309885	-157.85814
Las Vegas, Nevada	2	36.171562	-115.1391
New York, New York	3	40.712776	-74.005974

Passaggio 2. Creare una tabella online

Una tabella online è una copia di sola lettura di una tabella Delta ottimizzata per l'accesso online. Per altre informazioni, vedere Usare tabelle online per la gestione delle funzionalità in tempo reale.

Per creare una tabella online, è possibile usare l'interfaccia utente Creare una tabella online usando l'interfaccia utente, l'API REST o Databricks SDK, come nell'esempio seguente:

from pprint import pprint
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.catalog import *
import mlflow

workspace = WorkspaceClient()

# Create an online table
feature_table_name = f"main.on_demand_demo.location_features"
online_table_name=f"main.on_demand_demo.location_features_online"

spec = OnlineTableSpec(
 primary_key_columns=["destination_id"],
 source_table_full_name = feature_table_name,
 run_triggered=OnlineTableSpecTriggeredSchedulingPolicy.from_dict({'triggered': 'true'}),
 perform_full_copy=True)

# ignore "already exists" error
try:
 online_table_pipeline = workspace.online_tables.create(name=online_table_name, spec=spec)
except Exception as e:
 if "already exists" in str(e):
   pass
 else:
   raise e

pprint(workspace.online_tables.get(online_table_name))

Passaggio 3. Creare una funzione nel catalogo unity

In questo esempio, la funzione calcola la distanza tra la destinazione (la cui posizione non cambia) e l'utente (la cui posizione cambia frequentemente e non è nota fino al momento dell'inferenza).

# 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
$$""")

Passaggio 4. Creare una specifica di funzionalità in Unity Catalog

La specifica di funzionalità specifica le funzionalità che l'endpoint gestisce e le relative chiavi di ricerca. Specifica inoltre le funzioni necessarie da applicare alle funzionalità recuperate con le relative associazioni. Per informazioni dettagliate, vedere Creare un featureSpec.

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

Passaggio 5. Creare un endpoint di gestione delle funzionalità

Per creare un endpoint di gestione delle funzionalità, è possibile usare l'interfaccia utente Creare un endpoint, l'API REST o Databricks SDK, come illustrato di seguito.

L'endpoint di gestione delle funzionalità accetta come parametro l'oggetto feature_spec creato nel passaggio 4.

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)

Passaggio 6. Eseguire query sull'endpoint di gestione delle funzionalità

Quando si esegue una query sull'endpoint, si specifica la chiave primaria e, facoltativamente, i dati di contesto usati dalla funzione. In questo esempio, la funzione accetta come input la posizione corrente dell'utente (latitudine e longitudine). Poiché la posizione dell'utente cambia costantemente, deve essere fornita alla funzione in fase di inferenza come funzionalità di contesto.

È anche possibile eseguire query sull'endpoint usando l'interfaccia utente Eseguire query su un endpoint usando l'interfaccia utente o l'API REST.

Per semplicità, questo esempio calcola solo la distanza tra due città. Uno scenario più realistico potrebbe calcolare la distanza dell'utente da ogni posizione nella tabella delle funzionalità per determinare quali città consigliare.

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)

Notebook di esempio

Per una illustrazione completa dei passaggi, vedere questo notebook:

Notebook di esempio di Funzionalità di gestione delle funzionalità con tabelle online

Ottenere il notebook

Informazioni aggiuntive

Per informazioni dettagliate sull'uso dell'API Python di progettazione delle funzionalità, vedere la documentazione di riferimento.