Usare tables online per la gestione delle funzionalità in tempo reale

Un table online è una copia di sola lettura di un Delta Table, archiviata in formato orientato alle righe e ottimizzata per l'accesso online. I tables online sono completamente serverless tables che ridimensionano automaticamente la capacità di velocità effettiva con il carico delle richieste e forniscono un accesso a bassa latenza e velocità effettiva elevata ai dati di qualsiasi scala. Le tables online sono progettate per lavorare con le applicazioni Mosaic AI Model Serving, Feature Serving e di generazione aumentata (RAG) where vengono usate per ricerche rapide dei dati.

È possibile utilizzare tables anche online nelle query utilizzando Lakehouse Federation. Quando si utilizza la federazione di Lakehouse, è necessario impiegare un serverless data warehouse SQL per accedere online a tables. Sono supportate solo le operazioni di lettura (SELECT). Questa funzionalità è destinata solo a scopi interattivi o di debug e non deve essere usata per carichi di lavoro di produzione o cruciali.

La creazione di un table online tramite l'interfaccia utente di Databricks è un processo in un unico passaggio. Esegui select il Delta table in Explorer Catalog e crea online selecttable. È anche possibile usare l'API REST o Databricks SDK per creare e gestire tablesonline. Vedere Lavorare online con tables tramite le API.

Requisiti

• L'area di lavoro deve essere abilitata per Unity Catalog. Seguire la documentazione per creare un metastore di Unity Catalog, abilitarlo in uno spazio di lavoro e creare un Catalog.
• Un modello deve essere registrato in Unity Catalog per accedere online a tables.

Lavorare con tables online tramite l'interfaccia utente

Questa sezione descrive come creare ed eliminare tablesonline e come controllare lo stato e attivare gli aggiornamenti dei tablesonline.

Creare un table online usando l'interfaccia utente

È possibile creare un table online usando Catalog Explorer. Per altre informazioni sulle autorizzazioni necessarie, vedere Autorizzazioni.

1. Per creare un tableonline, Delta table deve avere una chiave primaria di origine. Se il Delta table da usare non ha una chiave primaria, crearne una seguendo queste istruzioni: Usare il Delta table esistente in Unity Catalog come una funzionalità table.

2. In Catalog Explorer, accedi alla sorgente table che desideri sync a un tableonline. Dal menu CreaselectOnline table.

   online

3. Usare i selettori nella finestra di dialogo per configurare tableonline.

   

   Nome: nome da usare per il table online in Unity Catalog.

   Chiave primaria: Column(i) nella sorgente table da usare come chiave primaria (o chiavi) nel tableonline.

   Chiave Timeseries: (facoltativo). Column nella table di origine da usare come chiave serie temporale. Qualora specificato, il table online include solo la riga con il valore della chiave della serie temporale più recente per ogni chiave primaria.

   Sync modalità: specifica come la pipeline di sincronizzazione aggiorna l'tableonline. Select uno tra istantanea, attivatoo continuo.

   Criterio	Descrizione
   Snapshot	La pipeline viene eseguita una volta per creare uno snapshot della sorgente table e copiarlo su tableonline. Le modifiche successive al table di origine vengono automaticamente riflesse nella table online catturando una nuova istantanea dell'origine e creando una nuova copia. Il contenuto del table online viene aggiornato in modo atomico.
   Attivato	La pipeline viene eseguita una volta per creare una copia snapshot iniziale della sorgente table nell'online table. A differenza della modalità snapshot sync, quando l'table online viene aggiornata, vengono recuperate e applicate solo le modifiche apportate dall'ultima esecuzione della pipeline all'tableonline. L'incremento refresh può essere attivato manualmente o automaticamente in base a una pianificazione.
   Continuo	La pipeline viene eseguita continuamente. Le modifiche successive al table di origine vengono applicate in modo incrementale al table online in modalità di streaming in tempo reale. Non è necessario alcun refresh manuale.

1. Fare clic su Fine al termine dell’operazione. Viene visualizzata la pagina table online.
2. Il nuovo table online viene creato con catalog, schemae il nome specificato nella finestra di dialogo di creazione. In Catalog Explorer il table online è indicato dall'icona table online .

Get stato e attivare gli aggiornamenti tramite l'interfaccia utente

Per controllare lo stato del tableonline, fare clic sul nome del table nel Catalog per aprirlo. La pagina online viene visualizzata con la scheda panoramica aperta. La sezione inserimento dati mostra lo stato dell'ultimo update. Per attivare un update, fare clic su Sync ora. La sezione di inserimento dati include anche un collegamento alla pipeline di Delta Live Tables che aggiorna table.

Pianificare gli aggiornamenti periodici

Per tables online con Snapshot o modalità attivata Triggeredsync, è possibile pianificare aggiornamenti periodici automatici. La pianificazione update viene gestita dalla pipeline Delta Live Tables che aggiorna il table.

1. In Esplora Catalog passare all'tableonline.
2. Nella sezione Inserimento dati fare clic sul collegamento alla pipeline.
3. Nell'angolo superiore destro, fare clic su Schedulee aggiungere un nuovo programma o update programmi esistenti.

Eliminare un table online attraverso l'interfaccia utente

Dalla pagina online tableselectElimina dal menu kebab .

Lavorare con tables online utilizzando le API

È anche possibile usare Databricks SDK o l'API REST per creare e gestire tablesonline.

Per informazioni di riferimento, vedere la documentazione di riferimento per Databricks SDK per Python o l’API REST.

Requisiti

Databricks SDK versione 0.20 o successiva.

Creare un table online usando le API

SDK Databricks - Python

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

w = WorkspaceClient(host='https://xxx.databricks.com', token='xxx')

# Create an online table
spec = OnlineTableSpec(
  primary_key_columns=["pk_col"],
  source_table_full_name="main.default.source_table",
  run_triggered=OnlineTableSpecTriggeredSchedulingPolicy.from_dict({'triggered': 'true'})
)

online_table = OnlineTable(
  name="main.default.my_online_table",  # Fully qualified table name
  spec=spec  # Online table specification
)

w.online_tables.create_and_wait(table=online_table)

REST API

curl --request POST "https://xxx.databricks.com/api/2.0/online-tables" \
--header "Authorization: Bearer xxx" \
--data '{
    "name": "main.default.my_online_table",
    "spec": {
        "run_triggered": {},
        "source_table_full_name": "main.default.source_table",
        "primary_key_columns": ["a"]
    }
  }'

L'table online avvia automaticamente la sincronizzazione dopo che è stato creato.

Get stato e attivazione refresh usando le API

È possibile visualizzare lo stato e le specifiche del table online seguendo l'esempio riportato di seguito. Se il table online non è continuo e si vuole attivare un refresh manuale dei relativi dati, è possibile usare l'API della pipeline per farlo.

Usare l'ID pipeline associato al table online nella specifica di table online e avviare una nuova update nella pipeline per attivare il refresh. Equivale a fare clic su Sync ora nell'interfaccia utente table online in Catalog Explorer.

SDK Databricks - Python

pprint(w.online_tables.get('main.default.my_online_table'))

# Sample response
OnlineTable(name='main.default.my_online_table',
    spec=OnlineTableSpec(perform_full_copy=None,
        pipeline_id='some-pipeline-id',
        primary_key_columns=['pk_col'],
        run_continuously=None,
        run_triggered={},
        source_table_full_name='main.default.source_table',
        timeseries_key=None),
    status=OnlineTableStatus(continuous_update_status=None,
        detailed_state=OnlineTableState.PROVISIONING,
        failed_status=None,
        message='Online Table creation is '
            'pending. Check latest status in '
            'Delta Live Tables: '
            'https://xxx.databricks.com/pipelines/some-pipeline-id',
        provisioning_status=None,
        triggered_update_status=None))

# Trigger an online table refresh by calling the pipeline API. To discard all existing data
# in the online table before refreshing, set "full_refresh" to "True". This is useful if your
# online table sync is stuck due to, for example, the source table being deleted and recreated
# with the same name while the sync was running.
w.pipelines.start_update(pipeline_id='some-pipeline-id', full_refresh=True)

REST API

curl --request GET \
  "https://xxx.databricks.com/api/2.0/online-tables/main.default.my_online_table" \
  --header "Authorization: Bearer xxx"

# Sample response
{
  "name": "main.default.my_online_table",
  "spec": {
    "run_triggered": {},
    "source_table_full_name": "main.default.source_table",
    "primary_key_columns": ["pk_col"],
    "pipeline_id": "some-pipeline-id"
  },
  "status": {
    "detailed_state": "PROVISIONING",
    "message": "Online Table creation is pending. Check latest status in Delta Live Tables: https://xxx.databricks.com#joblist/pipelines/some-pipeline-id"
  }
}

# Trigger an online table refresh by calling the pipeline API. To discard all existing data
# in the online table before refreshing, set "full_refresh" to "True". This is useful if your
# online table sync is stuck due to, for example, the source table being deleted and recreated
# with the same name while the sync was running.
curl --request POST "https://xxx.databricks.com/api/2.0/pipelines/some-pipeline-id/updates" \
  --header "Authorization: Bearer xxx" \
  --data '{
    "full_refresh": true
  }'

Eliminare un table online tramite le API

SDK Databricks - Python

w.online_tables.delete('main.default.my_online_table')

REST API

curl --request DELETE \
  "https://xxx.databricks.com/api/2.0/online-tables/main.default.my_online_table" \
  --header "Authorization: Bearer xxx"

L'eliminazione del table online interrompe la sincronizzazione dei dati in corso e rilascia tutte le relative risorse.

Gestire i dati table online usando un endpoint di gestione delle funzionalità

Per i modelli e le applicazioni ospitati all'esterno di Databricks, è possibile creare un endpoint di gestione delle funzionalità per la gestione delle funzionalità da tablesonline. L’endpoint rende disponibili le funzionalità a bassa latenza usando un'API REST.

1. Creazione di funzioni specifiche

   Quando si crea una specifica di funzionalità, si specifica il delta di origine table. In questo modo, la specifica di funzionalità può essere usata sia in scenari offline che online. Per le ricerche online, l'endpoint di servizio usa automaticamente l'table online per eseguire ricerche di funzionalità con bassa latenza.

   L'table Delta di origine e l'table online devono usare la stessa chiave primaria.

   La specifica della funzionalità può essere visualizzata nella scheda funzione in Catalog Explorer.

   from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup
   
   fe = FeatureEngineeringClient()
   fe.create_feature_spec(
     name="catalog.default.user_preferences_spec",
     features=[
       FeatureLookup(
         table_name="user_preferences",
         lookup_key="user_id"
       )
     ]
   )
   

2. Creare un endpoint di gestione delle funzionalità.

   Questo passaggio presuppone che sia stata creata una table online denominata user_preferences_online_table che sincronizza i dati del Delta tableuser_preferences. Usare la specifica di funzionalità per creare un endpoint di gestione delle funzionalità. L'endpoint rende disponibili i dati tramite un'API REST usando l'tableonline associato.

   Nota

   L'utente che esegue questa operazione deve essere il proprietario sia del table offline che del tableonline.

   SDK Databricks - Python

   from databricks.sdk import WorkspaceClient
   from databricks.sdk.service.serving import EndpointCoreConfigInput, ServedEntityInput
   
   workspace = WorkspaceClient()
   
   # Create endpoint
   endpoint_name = "fse-location"
   
   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"
         )
       ]
     )
   )
   

   API Python

   from databricks.feature_engineering.entities.feature_serving_endpoint import (
     ServedEntity,
     EndpointCoreConfig,
   )
   
   fe.create_feature_serving_endpoint(
     name="user-preferences",
     config=EndpointCoreConfig(
       served_entities=ServedEntity(
         feature_spec_name="catalog.default.user_preferences_spec",
         workload_size="Small",
         scale_to_zero_enabled=True
       )
     )
   )
   

3. Get dati dall'endpoint di gestione delle funzionalità.

   Per accedere all'endpoint API, inviare una richiesta HTTP GET all'URL dell'endpoint. L’esempio seguente illustra come eseguire questa operazione con Python. Per altri linguaggi e strumenti, vedere Funzionalità di gestione delle funzionalità.

   # Set up credentials
   export DATABRICKS_TOKEN=...
   

   url = "https://{workspace_url}/serving-endpoints/user-preferences/invocations"
   
   headers = {'Authorization': f'Bearer {DATABRICKS_TOKEN}', 'Content-Type': 'application/json'}
   
   data = {
     "dataframe_records": [{"user_id": user_id}]
   }
   data_json = json.dumps(data, allow_nan=True)
   
   response = requests.request(method='POST', headers=headers, url=url, data=data_json)
   if response.status_code != 200:
     raise Exception(f'Request failed with status {response.status_code}, {response.text}')
   
   print(response.json()['outputs'][0]['hotel_preference'])
   

Usare tables online con applicazioni RAG

Le applicazioni RAG sono un caso d'uso comune per tablesonline. È possibile creare un table online per i dati strutturati necessari all'applicazione RAG e ospitarli in un endpoint di gestione delle funzionalità. L'applicazione RAG usa l'endpoint di gestione delle funzionalità per cercare i dati pertinenti dal tableonline.

Attieniti alla procedura seguente:

1. Creare un endpoint di gestione delle funzionalità.
2. Creare uno strumento usando LangChain o qualsiasi pacchetto simile che usa l’endpoint per cercare i dati pertinenti.
3. Usare lo strumento in un agente LangChain o in un agente simile per recuperare i dati pertinenti.
4. Creare un endpoint di gestione del modello per ospitare l’applicazione.

Per istruzioni dettagliate e un notebook di esempio, vedere Esempio di progettazione delle funzionalità: applicazione RAG strutturata.

Esempi di notebook

Il notebook seguente illustra come pubblicare funzionalità in tables online per la gestione in tempo reale e la ricerca automatica delle funzionalità.

Demo del notebook online tables

Get notebook

Utilizzare tables con Mosaic AI Model Serving online

È possibile usare tables online per cercare le funzionalità di Mosaic AI Model Serving. Quando si sync una funzionalità table a un tableonline, i modelli addestrati utilizzando funzionalità di quella funzionalità table cercano automaticamente la funzionalità values dal sistema online table durante l'inferenza. Non è necessaria alcuna configurazione aggiuntiva.

1. Usare un FeatureLookup oggetto per eseguire il training del modello.

   Per il training del modello, usare le funzionalità della funzionalità offline table nel training del modello set, come illustrato nell'esempio seguente:

   training_set = fe.create_training_set(
     df=id_rt_feature_labels,
     label='quality',
     feature_lookups=[
         FeatureLookup(
             table_name="user_preferences",
             lookup_key="user_id"
         )
     ],
     exclude_columns=['user_id'],
   )
   

2. Usare il modello con Mosaic AI Model Serving. Il modello cerca automaticamente le caratteristiche del tableonline. Per informazioni dettagliate, vedere Ricerca automatica delle funzionalità con Databricks Model Serving .

Autorizzazioni utenti

Per creare un tableonline, è necessario disporre delle autorizzazioni seguenti:

• SELECT privilegio su la sorgente table.
• USE_CATALOG privilegio su il catalogdi destinazione.
• USE_SCHEMA e CREATE_TABLE privilegio sulla destinazione schema.

Per gestire la pipeline di sincronizzazione dei dati di un tableonline, è necessario essere il proprietario del table online oppure avere il privilegio di REFRESH sul tableonline. Gli utenti che non dispongono di privilegi di USE_CATALOG e USE_SCHEMA per il catalog non vedranno il table online in Catalog Explorer.

Il metastore Catalog Unity deve avere il Modello di Privilegio versione 1.0.

Autorizzazione dell’endpoint

Un'entità servizio univoca viene creata automaticamente per un endpoint di servizio funzionalità o di servizio modello, con autorizzazioni limitate necessarie per eseguire query sui dati da tablesonline. Questa entità servizio consente agli endpoint di accedere ai dati indipendentemente dall'utente che ha creato la risorsa e garantisce che l’endpoint possa continuare a funzionare se l’autore lascia l'area di lavoro.

La durata di questa entità servizio è la durata dell’endpoint. I log di controllo possono indicare i record generati dal sistema per il proprietario dell'Unity Catalogcatalog che concede i privilegi necessari a questo principale del servizio.

Limiti

• È supportato un solo table online per ogni tabledi origine.
• Un table online e la sua fonte table possono avere al massimo 1000 columns.
• Columns di tipi di dati ARRAY, MAP o STRUCT non possono essere usati come chiavi primarie nel tableonline.
• Se un column viene utilizzato come chiave primaria nella tableonline , tutte le righe nell'origine tablewhere l'column contiene values Null vengono ignorate.
• Le tables esterne, di sistema e interne non sono supportate come origine tables.
• Senza il feed di dati delle modifiche Delta abilitato, l'origine tables supporta solo la modalità snapshot sync.
• Delta Sharing tables sono supportati solo nella modalità snapshotsync.
• I nomi di Catalog, schemae table dei table online possono contenere solo caratteri alfanumerici e underscore e non devono iniziare con un numero. I trattini (-) non sono consentiti.
• Columns di tipo String è limitato a 64 KB.
• Column i nomi sono limitati a una lunghezza di 64 caratteri.
• La dimensione massima del video è di 2 MB.
• Le dimensioni combinate di tutti i tables online in un metastore Catalog Unity durante l'anteprima pubblica sono pari a 2 TB di dati utente non compressi.
• Il numero massimo di query al secondo (QPS) è 12.000. Contatta il team dell'account Databricks per aumentare il limit.

Risoluzione dei problemi

Non vedo l'opzione Crea online table

La causa è in genere che il table da cui si sta tentando di sync (l'origine table) non è un tipo supportato. Assicurarsi che il tipo di sicurezza della fonte table(mostrato nella scheda Catalog Explorer Dettagli) sia una delle opzioni supportate indicate di seguito.

• TABLE_EXTERNAL
• TABLE_DELTA
• TABLE_DELTA_EXTERNAL
• TABLE_DELTASHARING
• TABLE_DELTASHARING_MUTABLE
• TABLE_STREAMING_LIVE_TABLE
• TABLE_STANDARD
• TABLE_FEATURE_STORE
• TABLE_FEATURE_STORE_EXTERNAL
• TABLE_VIEW
• TABLE_VIEW_DELTASHARING
• TABLE_MATERIALIZED_VIEW

Non è possibile attivata o modalità di continua quando si crea un online

Ciò si verifica se la sorgente table non dispone di un feed di dati delle modifiche Delta abilitato o se si tratta di una vista o vista materializzata. Per usare la modalità incrementale di , abilitare il feed di dati delle modifiche nella fonte oppure utilizzare una visualizzazione non .

Il tableupdate online ha esito negativo o lo stato mostra offline

Per iniziare a risolvere l'errore, fare clic sull'ID della pipeline visualizzato nella scheda Panoramica della online in Esplora .

Nella pagina dell’interfaccia utente della pipeline visualizzata fare clic sulla voce “Failed to resolve flow’'__online_table”.

Viene visualizzata una finestra popup con i dettagli nella sezione Dettagli errore.

Le cause radice comuni includono:

• Il table di origine è stato eliminato o eliminato e ricreato con lo stesso nome, mentre il table online stava sincronizzando. Ciò è particolarmente comune con i continui tablesonline, perché sono costantemente sincronizzati.

• Non è possibile accedere alla sorgente table tramite elaborazione serverless a causa delle impostazioni del firewall. In questo caso, la sezione Dettagli errore potrebbe visualizzare il messaggio di errore "Impossibile avviare il servizio DLT nel cluster xxx...".

• Le dimensioni aggregate dei tables online superano i 2 TB (dimensioni non compresse) a livello di metastore limit. Il limit da 2 TB fa riferimento alle dimensioni non compresse dopo aver espanso il table Delta in formato orientato alle righe. Le dimensioni del table in formato riga possono essere significativamente maggiori rispetto al Delta table mostrato in Catalog Explorer, che si riferisce alle dimensioni compresse del table in un formato orientato column. La differenza può essere pari a 100x, a seconda del contenuto del table.

  Per stimare le dimensioni delle righe espanse non compresse di un Delta table, usare la query seguente da un Serverless SQL Warehouse. La query restituisce le dimensioni stimate espanse di table in byte. L'esecuzione corretta di questa query conferma anche che il Serverless Compute può accedere all'origine table.

  SELECT sum(length(to_csv(struct(*)))) FROM `source_table`;