Riferimento API di Feature Views

Importante

Questa funzionalità è in Anteprima Pubblica. Gli amministratori dell'area di lavoro possono controllare l'accesso a questa funzionalità dalla pagina Anteprime . Vedere Gestire le anteprime di Azure Databricks.

Controllo di accesso

Le funzionalità sono oggetti del catalogo Unity gestibili. L'accesso CREATE FEATUREa una funzionalità è controllato dai privilegi del catalogo unity , READ FEATUREe MANAGE . Per le descrizioni complete, vedere Informazioni di riferimento sui privilegi del catalogo Unity.

  • CREATE FEATURE - Obbligatorio per creare una funzionalità in uno schema. create_feature e register_feature richiedono CREATE FEATURE nello schema padre. Seguendo il principio dei privilegi minimi, concedere CREATE FEATURE a livello di schema; è anche possibile concederlo a un catalogo per consentire la creazione di funzionalità in qualsiasi schema in tale catalogo.
  • READ FEATURE - Obbligatorio per leggere una funzionalità e i relativi dati. get_feature, create_training_sete la lettura dei dati delle funzionalità materializzati per il training o la gestione richiedono READ FEATURE la funzionalità. READ FEATURE concesso per uno schema o un catalogo si applica a tutte le funzionalità correnti e future contenute.
  • MANAGE — Necessario per gestire il ciclo di vita e le concessioni di una funzionalità. L'eliminazione di una funzionalità con delete_featuree la materializzazione di una funzionalità con materialize_features o delete_materialized_feature, richiedono MANAGE nella funzionalità .

Tutte le operazioni sulle funzionalità richiedono USE CATALOG anche nel catalogo padre e USE SCHEMA nello schema padre. Per informazioni su come MANAGE e READ FEATURE si applica alla materializzazione, vedere Autorizzazioni.

API Visualizzazione funzionalità

Feature costruttore e register_feature()

L'approccio consigliato consiste nel costruire un Feature oggetto in locale e usarlo register_feature per renderlo persistente in Unity Catalog. Questo flusso di lavoro in due passaggi consente di sperimentare le funzionalità (incluso create_training_set) prima di registrarle.

Feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    entity: Optional[List[str]] = None,                    # Required for all sources except RequestSource: entity columns
    timeseries_column: Optional[str] = None,               # Required for all sources except RequestSource: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
)

FeatureEngineeringClient.register_feature() registra un oggetto costruito Feature localmente in Unity Catalog.

FeatureEngineeringClient.register_feature(
    feature: Feature,       # Required: A Feature instance (not already registered)
    catalog_name: str,      # Required: UC catalog name
    schema_name: str,       # Required: UC schema name
) -> Feature

from databricks.feature_engineering.entities import Feature, DeltaTableSource, AggregationFunction, Sum, RollingWindow
from datetime import timedelta

# Step 1: Construct the feature locally
feature = Feature(
    source=DeltaTableSource(catalog_name="main", schema_name="store", table_name="transactions"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

# Step 2: Register in Unity Catalog
fe = FeatureEngineeringClient()
registered_feature = fe.register_feature(
    feature=feature,
    catalog_name="main",
    schema_name="store",
)

create_feature()

FeatureEngineeringClient.create_feature() convalida, costruisce e registra immediatamente una funzionalità nel catalogo unity in un unico passaggio. Usare questa opzione quando non è necessario provare prima di tutto con la funzionalità in locale.

FeatureEngineeringClient.create_feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    catalog_name: str,                                     # Required: The catalog name for the feature
    schema_name: str,                                      # Required: The schema name for the feature
    entity: Optional[List[str]] = None,                    # Required for all sources except RequestSource: entity columns
    timeseries_column: Optional[str] = None,               # Required for all sources except RequestSource: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
) -> Feature

Parametri:

  • source: origine dati usata nel calcolo delle funzionalità (DeltaTableSource, StreamSourceo RequestSource).
  • function: che AggregationFunction aggrega l'operatore (ad esempio, Sum(input="amount")), la colonna di input e l'intervallo di tempo insieme. Oppure ColumnSelection("column_name") per le funzionalità pass-through.
  • catalog_name: nome del catalogo Unity per la funzionalità.
  • schema_name: nome dello schema del catalogo Unity per la funzionalità.
  • entity: elenco di nomi di colonna che definiscono le chiavi di aggregazione o di ricerca (chiavi primarie). Obbligatorio per tutti i tipi di origine ad eccezione RequestSourcedi . Ad esempio, ["user_id"] aggrega o cerca per utente.
  • timeseries_column: colonna timestamp utilizzata per l'aggregazione dell'intervallo di tempo o per la selezione di valori più recenti. Obbligatorio per tutti i tipi di origine ad eccezione RequestSourcedi .
  • name: nome della funzionalità facoltativo. Se omesso, generato automaticamente dalla colonna di input, dalla funzione e dalla finestra (ad esempio, amount_avg_rolling_7d).
  • description: Descrizione facoltativa della funzionalità.

Restituisce: Un'istanza di Feature convalidata

Genera: ValueError se fallisce una convalida

delete_feature()

Elimina una funzionalità dal catalogo unity in base al nome completo.

FeatureEngineeringClient.delete_feature(
    full_name: str,  # Required: '<catalog>.<schema>.<feature_name>'
) -> None

fe.delete_feature(full_name="main.store.amount_sum_rolling_7d")

Prima di eliminare una funzionalità, rimuovere o aggiornare i modelli o le specifiche di funzionalità che vi fanno riferimento. Se la funzionalità è stata materializzata, eliminare prima la caratteristica materializzata. Vedere Come eliminare una funzionalità materializzata.

Nomi generati automaticamente

Quando name viene omesso, viene generato automaticamente un nome. I nomi generati seguono il modello : {column}_{function}_{window}. Per esempio:

  • price_avg_rolling_1h (prezzo medio di 1 ora)
  • transaction_count_rolling_30d_1d (conteggio di 30 giorni della transazione con ritardo 1d dal timestamp dell'evento)

Funzioni supportate

Funzione di aggregazione

Note

Le funzioni di aggregazione vengono disposte in un insieme AggregationFunction a un intervallo di tempo, come descritto nelle finestre temporali. Ogni funzione accetta un input parametro che specifica la colonna di origine da aggregare.

Function Description Esempio di caso d'uso
Sum(input="column") Totale dei valori Utilizzo giornaliero delle app per utente in minuti
Avg(input="column") Media dei valori Importo medio delle transazioni
Count(input="column") Numero di record Numero di accessi per utente
Min(input="column") Valore minimo Frequenza cardiaca più bassa registrata da un dispositivo indossabile
Max(input="column") Valore massimo Quantità massima di transazioni per sessione
StddevPop(input="column") Deviazione standard della popolazione Variabilità giornaliera dell'importo delle transazioni in tutti i clienti
StddevSamp(input="column") Deviazione standard di esempio Variabilità dei tassi di click-through delle campagne pubblicitarie
VarPop(input="column") Varianza della popolazione Distribuzione delle letture dei sensori per i dispositivi IoT in una fabbrica
VarSamp(input="column") Varianza campione Distribuzione delle classificazioni dei film su un gruppo campionato
ApproxCountDistinct(input="column", relativeSD=0.05) Conteggio univoco approssimativo Conteggio distinto degli articoli acquistati
ApproxPercentile(input="column", percentile=0.95, accuracy=100) Percentile approssimativo Latenza di risposta p95
First(input="column") Primo valore Timestamp del primo accesso
Last(input="column") Ultimo valore Importo dell'acquisto più recente

Note

First e Last includono valori Null per impostazione predefinita. Per ignorare i valori Null, aggiungere un oggetto filter_condition che esclude in modo esplicito le colonne di input null.

ColumnSelection (pass-through)

ColumnSelection seleziona una singola colonna da un'origine senza applicare alcuna aggregazione. Viene eseguito il wrapping direttamente nel function parametro (non all'interno AggregationFunctiondi ). Il tipo restituito viene dedotto dallo schema di origine.

Function Description Esempio di caso d'uso
ColumnSelection("col") Valore più recente di una colonna (nessuna aggregazione) Categoria fornitore più recente, pass-through di un campo richiesta

ColumnSelection può essere usato con qualsiasi origine dati:

  • DeltaTableSource: restituisce il valore più recente per ogni chiave di entità tramite un join temporizzato (nessuna aggregazione della finestra di lookback).
  • StreamSource: restituisce il valore più recente per ogni chiave di entità dal flusso (nessuna aggregazione della finestra di lookback).
  • RequestSource: passa attraverso il valore fornito in fase di inferenza (o estratto dal dataframe etichettato in fase di training).
from databricks.feature_engineering.entities import (
    ColumnSelection, DeltaTableSource, Feature, FieldDefinition,
    RequestSource, ScalarDataType,
)

delta_source = DeltaTableSource(
    catalog_name="main", schema_name="feature_store", table_name="transactions",
)

request_source = RequestSource(
    schema=[
        FieldDefinition(name="session_duration", data_type=ScalarDataType.DOUBLE),
    ]
)

# ColumnSelection from a Delta table
latest_amount = Feature(
    source=delta_source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    name="latest_transaction_amount",
)

# ColumnSelection from a RequestSource
session_feature = Feature(
    source=request_source,
    function=ColumnSelection("session_duration"),
    name="session_duration",
)

Esempio: funzionalità di aggregazione e selezione di colonne

L'esempio seguente mostra le funzionalità definite sulla stessa origine dati.

from databricks.feature_engineering.entities import (
    AggregationFunction, Feature, Sum, Avg, ApproxCountDistinct,
    ColumnSelection, RollingWindow,
)
from datetime import timedelta

window = RollingWindow(window_duration=timedelta(days=7))

sum_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(Sum(input="amount"), window),
)

avg_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(Avg(input="amount"), window),
)

distinct_count = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(ApproxCountDistinct(input="product_id", relativeSD=0.01), window),
)

# Column selection (no aggregation, no time window)
latest_amount = Feature(
    source=source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="event_time",
    name="latest_amount",
)

Funzionalità con condizioni di filtro

Il filter_condition parametro consente di filtrare le righe dalla tabella di origine prima di calcolare le aggregazioni. Questa funzione funge da clausola SQL WHERE applicata prima del raggruppamento e dell'aggregazione dei dati.

Note

filter_condition filtra le righe prima dell'aggregazione, ad esempio una clausola SQL WHERE applicata prima GROUP BYdi . Non modifica la granularità, che viene sempre definita dalla entity definizione della funzionalità.

I filtri sono utili quando si lavora con tabelle di origine di grandi dimensioni che includono un superset di dati necessari per il calcolo delle funzionalità e ridurre al minimo la necessità di creare viste separate sopra queste tabelle.

from databricks.feature_engineering.entities import AggregationFunction, Sum, Count, RollingWindow, DeltaTableSource
from datetime import timedelta

# Source with filter applied at the source level
high_value_transactions = DeltaTableSource(
    catalog_name="main",
    schema_name="ecommerce",
    table_name="transactions",
    filter_condition="amount > 100",  # Only transactions over $100
)

high_value_sales = Feature(
    source=high_value_transactions,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=30))),
)

# Multiple conditions
completed_orders_source = DeltaTableSource(
    catalog_name="main",
    schema_name="ecommerce",
    table_name="orders",
    filter_condition="status = 'completed' AND payment_method = 'credit_card'",
)

completed_orders = Feature(
    source=completed_orders_source,
    entity=["user_id"],
    timeseries_column="order_time",
    function=AggregationFunction(Count(input="order_id"), RollingWindow(window_duration=timedelta(days=7))),
)

# Filter on a StreamSource
from databricks.feature_engineering.entities import StreamSource

purchase_stream = StreamSource(
    full_name="main.ecommerce.transactions_stream",
    filter_condition="value.event_type = 'purchase'",
)

purchase_total = Feature(
    source=purchase_stream,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(Sum(input="value.amount"), RollingWindow(window_duration=timedelta(hours=1))),
)

Origini dati

DeltaTableSource

DeltaTableSource è un oggetto Python temporaneo usato per definire il modo in cui le funzionalità vengono calcolate da una tabella di origine. Non crea una nuova tabella. Specifica la configurazione per la lettura dei dati e l'aggregazione delle funzionalità.

DeltaTableSource(
    catalog_name: str,                              # Required: Catalog name
    schema_name: str,                               # Required: Schema name
    table_name: str,                                # Required: Table name
    filter_condition: Optional[str] = None,         # Optional: SQL WHERE clause to filter source data
    transformation_sql: Optional[str] = None,       # Optional: SQL SELECT expression for column transformations
    dataframe_schema: Optional[str] = None,         # Required if transformation_sql is set: schema of the resulting DataFrame
)

Parametri:

  • catalog_name, schema_name, table_name: identificare la tabella Delta di origine nel catalogo unity.
  • filter_condition: clausola SQL WHERE applicata prima dell'aggregazione. Esempio: "status = 'completed'".
  • transformation_sql: espressione SQL SELECT applicata alla tabella di origine. Usare questa opzione per rinominare colonne, tipi di cast o calcolare colonne derivate prima dell'aggregazione. Se omesso, vengono selezionate tutte le colonne (*). Esempio: "user_id, CAST(amount AS DOUBLE) AS amount, event_time".
  • dataframe_schema: schema del dataframe risultante dopo le trasformazioni, in formato JSON Spark StructType (da df.schema.json()). Obbligatorio se transformation_sql specificato. Questo indica al sistema i nomi e i tipi di colonna risultanti dalla trasformazione.

filter_condition transformation_sql Quando e sono impostati, la query risultante è : SELECT {transformation_sql} FROM {table} WHERE {filter_condition}.

Note

L'oggetto timeseries_column (specificato nella definizione della funzionalità, non in DeltaTableSource) deve essere di tipo TimestampType o DateType. I tipi integer possono funzionare, ma causano perdite di precisione per le aggregazioni dell'intervallo di tempo.

Esempio: uso di transformation_sql per le trasformazioni delle colonne

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="raw_events",
    transformation_sql="user_id, CAST(price_cents AS DOUBLE) / 100 AS price, event_time",
    filter_condition="event_type = 'purchase'",
    dataframe_schema=spark.sql(
        "SELECT user_id, CAST(price_cents AS DOUBLE) / 100 AS price, event_time FROM main.analytics.raw_events LIMIT 0"
    ).schema.json(),
)

Esempio: derivazione e transformation_sqldataframe_schema da un dataframe PySpark

È possibile scrivere la trasformazione come query PySpark, quindi estrarre lo schema dal dataframe risultante:

df = spark.sql(f"""
  SELECT user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time
  FROM main.analytics.events
  WHERE event_date >= date_sub(current_date(), 7)
  LIMIT 0
""")

# Use df.schema.json() as the dataframe_schema
source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    transformation_sql="user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time",
    filter_condition="event_date >= date_sub(current_date(), 7)",
    dataframe_schema=df.schema.json(),
)

Note

transformation_sql supporta solo espressioni a livello di riga (ridenominazione di colonne, cast, aritmetica). Le funzioni di aggregazione come COUNT(*) o SUM() non sono supportate. Usare AggregationFunction invece nella definizione di funzionalità.

DeltaTableSource.from_sql()

Per praticità, è possibile creare un oggetto DeltaTableSource da una query SQL. Il metodo analizza la query per estrarre automaticamente il nome della tabella, transformation_sql, e filter_condition.

DeltaTableSource.from_sql(
    sql: str,                           # Required: SQL SELECT query
    spark: SparkSession,                # Required: active SparkSession (for schema inference)
) -> DeltaTableSource

Sono supportate solo query semplici SELECT ... FROM ... [WHERE ...] . L'istruzione SQL complessa (JOINs, sottoquery, CTEs, UNIONs) viene rifiutata. Per le query complesse, costruire DeltaTableSource direttamente con transformation_sql e filter_condition.

from databricks.feature_engineering.entities import (
    AggregationFunction,
    DeltaTableSource,
    Feature,
    Sum,
    TumblingWindow,
)

source = DeltaTableSource.from_sql(
    spark=spark,
    sql=f"SELECT customer_id, event_ts, amount * 2 AS doubled_amount, amount FROM {CATALOG}.{SCHEMA}.{TABLE}",
)

feature = Feature(
    source=source,
    function=AggregationFunction(Sum(input="doubled_amount"), time_window=TumblingWindow(window_duration=timedelta(days=7))),
    entity=["customer_id"], timeseries_column="event_ts",
)

Eseguire l'iterazione con to_dataframe()

Usare source.to_dataframe() per visualizzare in anteprima i dati che verranno usati per il calcolo delle funzionalità. Ciò è utile per eseguire filter_condition l'iterazione e transformation_sql fino a quando non producono i risultati previsti.

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    filter_condition="event_type = 'purchase'",
)

# Preview the filtered source data
source.to_dataframe().display()

Informazioni sulle entità

Le colonne di entità definiscono il livello di aggregazione per le funzionalità. Vengono specificati nella Feature definizione, non in DeltaTableSource. Le entità determinano:

  • Modalità di raggruppamento dei dati: le funzionalità vengono aggregate per ogni combinazione univoca di valori di entità (simile a GROUP BY in SQL)
  • La struttura della chiave primaria: ogni combinazione di entità univoca restituisce una riga di funzionalità calcolate

Esempio: Funzionalità a livello di cliente

Il codice seguente aggrega le funzionalità a livello di cliente (una riga per cliente):

from databricks.feature_engineering.entities import DeltaTableSource

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="user_events",
)

Feature(
    source=source,
    entity=["user_id"],                # Features aggregated per user
    timeseries_column="event_time",    # Timestamp for time windows
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

Esempio: funzionalità a livello di Customer Store

Per aggregare le funzionalità a un livello più dettagliato (una riga per ogni combinazione di customer-store), usare più colonne di entità:

source = DeltaTableSource(
    catalog_name="main",
    schema_name="retail",
    table_name="transactions",
)

Feature(
    source=source,
    entity=["user_id", "store_id"],  # Features aggregated per user-store pair
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

Quando sono necessarie funzionalità a diversi livelli di aggregazione (ad esempio, a livello di cliente e a livello di punto vendita del cliente), usare valori diversi entity nelle definizioni delle funzionalità. Lo stesso DeltaTableSource può essere condiviso tra le funzionalità con configurazioni di entità diverse.

StreamSource

StreamSource fa riferimento a un oggetto Stream. Stream contiene la configurazione di connessione, autenticazione, schema e inserimento per l'origine di streaming. Per Kafka, i riferimenti alle colonne nelle definizioni di funzionalità devono essere preceduti value. da o key. per indicare la parte del messaggio da leggere.

StreamSource(
    full_name: str,                       # Required: Three-part Stream name (catalog.schema.stream)
    filter_condition: Optional[str],      # Optional: SQL WHERE clause applied before aggregation
)

Parametri:

  • full_name: nome completo in tre parti di un oggetto Stream (ad esempio, "my_catalog.my_schema.my_stream").
  • filter_condition (facoltativo): clausola SQL WHERE applicata ai dati di flusso prima dell'aggregazione, usando riferimenti a colonne con prefisso punto (ad esempio, "value.event_type = 'purchase'").
from databricks.feature_engineering.entities import StreamSource

stream_source = StreamSource(
    full_name="my_catalog.my_schema.my_stream",
    filter_condition="value.event_type = 'purchase'",
)

RequestSource

RequestSource definisce uno schema per i dati forniti in fase di inferenza nel payload della richiesta anziché ricercati da una tabella pre materializzata. Durante il training, queste colonne vengono estratte dal dataframe etichettato passato a create_training_set. Durante la gestione del modello, il chiamante deve includerli nel payload della richiesta HTTP.

RequestSource viene usato con ColumnSelection (per passare direttamente un valore). Non supporta funzioni di aggregazione o finestre temporali.

Definizione dello schema

Definire lo schema come elenco di oggetti, ognuno dei quali specifica un nome di FieldDefinition colonna e un ScalarDataTypeoggetto :

from databricks.feature_engineering.entities import (
    FieldDefinition, RequestSource, ScalarDataType,
)

request_source = RequestSource(
    schema=[
        FieldDefinition(name="transaction_amount", data_type=ScalarDataType.DOUBLE),
        FieldDefinition(name="vendor_id", data_type=ScalarDataType.STRING),
        FieldDefinition(name="transaction_id", data_type=ScalarDataType.STRING),
        FieldDefinition(name="transaction_time", data_type=ScalarDataType.DATE),
    ]
)

Tipi di dati supportati

RequestSourcesupporta i tipi scalari definiti in ScalarDataType: INTEGER, FLOAT, BOOLEAN, STRINGDOUBLE, LONG. TIMESTAMPDATESHORT I tipi complessi come matrici, mappe e struct non sono supportati.

Come vengono idratati i dati della richiesta

Contesto Behavior
Formazione (create_training_set) Le colonne vengono estratte dal dataframe etichettato. I tipi vengono convalidati rispetto allo schema dichiarato. Le mancate corrispondenze generano un errore (nessun cast implicito).
Gestione (endpoint modello) Le colonne vengono estratte da dataframe_records o dataframe_split nella richiesta HTTP. I valori JSON vengono cast ai tipi dichiarati (ad esempio, numero JSON → DOUBLE).

Firma del modello

Quando un modello viene registrato usando log_model con un set di training che include RequestSource funzionalità, le RequestSource colonne vengono aggiunte alla firma del modello MLflow come input necessari. Ciò significa che lo schema API dell'endpoint di servizio riflette i campi che i chiamanti devono fornire in fase di inferenza.

API di training e inferenza

create_training_set e score_batch calcolare i valori di funzionalità corretti temporizzato su richiesta dai dati di origine. Per le funzionalità che supportano la materializzazione offline, ad esempio le aggregazioni di finestre scorrevoli nelle origini di tabelle delta, la materializzazione delle funzionalità prima in un archivio offline migliora le prestazioni di entrambe le operazioni. Quando sono disponibili funzionalità offline materializzate, le operazioni leggono i dati offline precompilate anziché ricompilare i valori delle funzionalità dall'origine. Vedere Materialze Feature Views (Materialze Feature Views ) per materializzare le funzionalità in un negozio offline.

create_training_set()

Crea un set di dati di training con il calcolo delle funzionalità corretto temporizzato. Per informazioni dettagliate, vedere Eseguire il training di modelli con visualizzazioni delle funzionalità.

FeatureEngineeringClient.create_training_set(
    df: DataFrame,                                # DataFrame with training data
    features: Optional[List[Feature]],            # List of Feature objects
    label: Union[str, List[str], None],           # Label column name(s)
    exclude_columns: Optional[List[str]] = None,  # Optional: columns to exclude
) -> TrainingSet

log_model()

Registra un modello con metadati delle funzionalità per il rilevamento della derivazione e la ricerca automatica delle funzionalità durante l'inferenza. Per informazioni dettagliate, vedere Eseguire il training di modelli con visualizzazioni delle funzionalità.

FeatureEngineeringClient.log_model(
    model,                                    # Trained model object
    artifact_path: str,                       # Path to store model artifact
    flavor: ModuleType,                       # MLflow flavor module (e.g., mlflow.sklearn)
    training_set: TrainingSet,                # TrainingSet used for training
    registered_model_name: Optional[str],     # Optional: register model in Unity Catalog
)

score_batch()

Esegue l'inferenza batch offline con la ricerca automatica delle funzionalità. Usa i metadati delle funzionalità archiviati con il modello per calcolare le funzionalità corrette temporizzato, garantendo la coerenza con il training.

FeatureEngineeringClient.score_batch(
    model_uri: str,                           # URI of logged model (e.g., "models:/catalog.schema.model/1")
    df: DataFrame,                            # DataFrame with entity keys and timestamps
) -> DataFrame

Il dataframe di input deve contenere le colonne di entità e timeeries usate durante il training. Le funzionalità vengono calcolate automaticamente dai dati di origine.

fe = FeatureEngineeringClient()

# Batch scoring with automatic feature lookup
predictions = fe.score_batch(
    model_uri="models:/main.ecommerce.fraud_model/1",
    df=inference_df,
)
predictions.display()

Intervalli di tempo

Le visualizzazioni delle funzionalità supportano tre diversi tipi di finestra per controllare il comportamento di lookback per le aggregazioni basate su intervalli di tempo: sequenza, a cascata e scorrimento.

  • Il rollback delle finestre dall'ora dell'evento. La durata e il ritardo vengono definiti in modo esplicito.
  • Le finestre scorrevoli sono finestre temporali fisse e non sovrapposte. Ogni punto dati appartiene esattamente a una finestra.
  • Le finestre scorrevoli sono finestre temporali sovrapposte, mobili, con un intervallo di spostamento configurabile.

La figura seguente mostra come funzionano.

Finestre di lookback scorrevoli in sequenza, a cascata e scorrevoli.

Finestra mobile

Note

RollingWindow era precedentemente denominato ContinuousWindow. Se si esegue la migrazione da una versione precedente dell'SDK, aggiornare le importazioni di conseguenza.

Le finestre in sequenza sono up-to-date e aggregazioni in tempo reale, in genere usate sui dati di streaming. Nelle pipeline di streaming, la finestra in sequenza genera una nuova riga solo quando il contenuto della finestra a lunghezza fissa cambia, ad esempio quando un evento entra o esce. Quando viene usata una funzionalità finestra in sequenza nelle pipeline di training, viene eseguito un calcolo accurato delle funzionalità temporizzato sui dati di origine usando la durata della finestra a lunghezza fissa immediatamente precedente al timestamp di un evento specifico. Ciò consente di evitare l'asimmetria online-offline o la perdita di dati. Caratteristiche al tempo T aggregare eventi da [T − durata, T).

class RollingWindow(TimeWindow):
    window_duration: datetime.timedelta
    delay: Optional[datetime.timedelta] = None

Nella tabella seguente sono elencati i parametri per una finestra in sequenza. L'ora di inizio e di fine della finestra si basa su questi parametri come indicato di seguito:

  • Ora di inizio: evaluation_time - window_duration - delay (inclusivo)
  • Ora di fine: evaluation_time - delay (esclusivo)
Parametro Vincoli
delay (facoltativo) Deve essere ≥ 0 (sposta la finestra indietro nel tempo dal timestamp di valutazione). Usare delay per tenere conto di qualsiasi ritardo di sistema tra il momento in cui viene creato l'evento e il timestamp dell'evento per evitare future perdite di eventi nei set di dati di training. Ad esempio, se si verifica un ritardo di un minuto tra il momento in cui vengono creati gli eventi e questi eventi vengono infine inseriti in una tabella di origine in cui viene assegnato un timestamp, il ritardo sarà timedelta(minutes=1).
window_duration Deve essere > 0 
from databricks.feature_engineering.entities import RollingWindow
from datetime import timedelta

# Look back 7 days from evaluation time
window = RollingWindow(window_duration=timedelta(days=7))

Definire una finestra in sequenza con ritardo usando il codice seguente.

# Look back 7 days, offset by 1 minute to account for data ingestion delay
window = RollingWindow(
    window_duration=timedelta(days=7),
    delay=timedelta(minutes=1)
)

Esempi di finestre in sequenza

  • window_duration=timedelta(days=7): crea una finestra di osservazione retrospettiva di 7 giorni che termina al momento attuale della valutazione. Per un evento alle 2:00 del giorno 7, sono inclusi tutti gli eventi dalle 2:00 del giorno 0 fino (ma non incluso) alle 2:00 del giorno 7.

  • window_duration=timedelta(hours=1), delay=timedelta(minutes=30): crea una finestra di lookback di 1 ora che termina 30 minuti prima del tempo di valutazione. Per un evento alle 15:00, sono inclusi tutti gli eventi dalle 13:30 fino alle 14:30, ma escluso le 14:30. Ciò è utile per tenere conto dei ritardi di inserimento dei dati.

Finestra a cascata

Per le funzionalità definite tramite finestre a scorrimento, le aggregazioni vengono calcolate su una finestra a lunghezza fissa pre-determinata che avanza in base a un intervallo di scorrimento, producendo finestre non sovrapposte che partizionano completamente il tempo. Di conseguenza, ogni evento nell'origine contribuisce esattamente a una finestra. Le funzionalità al momento t aggregano dati dalle finestre che terminano a o prima di t in modo esclusivo. Windows inizia dall'epoca Unix.

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

Nella tabella seguente sono elencati i parametri per una finestra a cascata.

Parametro Vincoli
window_duration Deve essere > 0 
from databricks.feature_engineering.entities import TumblingWindow
from datetime import timedelta

window = TumblingWindow(
    window_duration=timedelta(days=7)
)

Esempio di finestra a cascata

  • window_duration=timedelta(days=5): crea finestre a lunghezza fissa predefinita di 5 giorni ciascuno. Esempio: la finestra n. 1 si estende da Giorno 0 a Giorno 4, Window #2 si estende da Giorno 5 a Giorno 9, Finestra #3 si estende da Giorno 10 a Giorno 14 e così via. In particolare, Window #1 include tutti gli eventi con timestamp a partire dal 00:00:00.00 giorno 0 fino a (ma non incluso) qualsiasi evento con timestamp 00:00:00.00 il giorno 5. Ogni evento appartiene esattamente a una finestra.

Finestra scorrevole

Per le caratteristiche definite tramite finestre scorrevoli, le aggregazioni vengono calcolate su una finestra di lunghezza fissa pre-determinata che avanza in base a un intervallo di avanzamento, producendo finestre sovrapposte. Ogni evento nell'origine può contribuire all'aggregazione delle funzionalità per più finestre. Le funzionalità al momento t aggregano dati dalle finestre che terminano a o prima di t in modo esclusivo. Windows inizia dall'epoca Unix.

class SlidingWindow(TimeWindow):
    window_duration: datetime.timedelta
    slide_duration: datetime.timedelta

Nella tabella seguente sono elencati i parametri per una finestra scorrevole.

Parametro Vincoli
window_duration Deve essere > 0
slide_duration Deve essere > 0 e <window_duration 
from databricks.feature_engineering.entities import SlidingWindow
from datetime import timedelta

window = SlidingWindow(
    window_duration=timedelta(days=7),
    slide_duration=timedelta(days=1)
)

Esempio di finestra scorrevole

  • window_duration=timedelta(days=5), slide_duration=timedelta(days=1): crea finestre sovrapposte di 5 giorni che avanzano di 1 giorno ogni volta. Esempio: la finestra n. 1 si estende da Giorno 0 a Giorno 4, Window #2 si estende da Giorno 1 a Giorno 5, Finestra #3 si estende da Giorno 2 a Giorno 6 e così via. Ogni finestra include gli eventi dal 00:00:00.00 giorno di inizio fino al (ma non incluso) 00:00:00.00 giorno di fine. Poiché le finestre si sovrappongono, un singolo evento può appartenere a più finestre (in questo esempio ogni evento appartiene a un massimo di 5 finestre diverse).

Trigger di materializzazione

Attiva il controllo quando viene eseguita una pipeline di materializzazione. Il tipo di trigger dipende dal tipo di funzionalità.

CronSchedule

Usare CronSchedule per le funzionalità di aggregazione (AggregationFunction). La pipeline viene eseguita in base a una pianificazione fissa definita da un'espressione cron di Quarzi.

from databricks.feature_engineering.entities import CronSchedule

trigger = CronSchedule(
    quartz_cron_expression="0 0 * * * ?",  # Hourly
    timezone_id="UTC",
)

TableTrigger

Usare TableTrigger per ColumnSelection le funzionalità supportate da un oggetto DeltaTableSource. La pipeline viene eseguita ogni volta che la tabella Delta upstream riceve un nuovo commit.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

StreamingMode

Usare StreamingMode per le funzionalità supportate da un oggetto StreamSource. La pipeline viene eseguita come pipeline di streaming continuo.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    StreamSource, Feature, AggregationFunction, Sum,
    RollingWindow, OnlineStoreConfig, StreamingMode,
)
from datetime import timedelta

fe = FeatureEngineeringClient()

stream_source = StreamSource(full_name="my_catalog.my_schema.my_stream")

streaming_feature = fe.create_feature(
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(
        operator=Sum(input="value.amount"),
        time_window=RollingWindow(window_duration=timedelta(hours=1)),
    ),
    catalog_name="my_catalog",
    schema_name="my_schema",
    name="user_purchase_sum",
)

fe.materialize_features(
    features=[streaming_feature],
    online_config=OnlineStoreConfig(
        catalog_name="my_catalog",
        schema_name="my_schema",
        table_name_prefix="streaming_features_serving",
        online_store_name="feature_store_online",
    ),
    trigger=StreamingMode(),
)

Scelta di un trigger

Tipo di funzionalità Trigger Quando viene eseguito
Aggregazione (AggregationFunction) da DeltaTableSource CronSchedule In base a una pianificazione cron fissa
ColumnSelection (da DeltaTableSource) TableTrigger In ogni commit della tabella di origine
Funzionalità di StreamSource StreamingMode Streaming continuo

Non è possibile materializzare le funzionalità che richiedono tipi di trigger diversi in una singola materialize_features chiamata. In alternativa, eseguire chiamate separate.

Eseguire la migrazione delle funzionalità beta all'anteprima pubblica

L'anteprima pubblica delle visualizzazioni delle funzionalità introduce le entità funzionalità di prima classe nel catalogo unity, regolate dai CREATE FEATURE privilegi e READ FEATURE e richiede databricks-feature-engineering la versione 0.16.0 o successiva. Le funzionalità create durante la versione beta (con la versione 0.15.0) vengono archiviate come funzioni del catalogo Unity e non supportano tutte le funzionalità di anteprima pubblica. Per ottenere il supporto dell'anteprima pubblica a lungo termine, ricreare le funzionalità beta con la versione 0.16.0. Le funzionalità devono essere eliminate e ricreate, non solo ri materializzate.

Per altre informazioni sulle funzionalità, vedere Visualizzazioni delle funzionalità.

Operazioni da eseguire

  • Eseguire l'aggiornamento alla versione 0.16.0. Questa è la versione client necessaria per le funzionalità di anteprima pubblica (batch e streaming).
  • Ricreare le funzionalità. Le visualizzazioni delle funzionalità beta devono essere eliminate e ricreate, non materializzate, perché non supportano tutte le funzionalità di anteprima pubblica.
  • Eseguire la migrazione prima della chiusura della finestra. È necessario eseguire la migrazione delle funzionalità beta esistenti prima del 22 luglio 2026.

Identificare le funzionalità beta e anteprima pubblica

Le funzionalità di anteprima pubblica vengono visualizzate come oggetto Feature nel catalogo unity, ad esempio in Esplora cataloghi. Le funzionalità beta vengono visualizzate come funzione con una definizione YAML. Qualsiasi funzionalità rappresentata come funzione è una funzionalità beta di cui è necessario eseguire la migrazione.

Eseguire la migrazione delle funzionalità beta

La migrazione di una funzionalità beta include tre parti:

  • Ricreare la funzionalità come funzionalità di anteprima pubblica.
  • Ricompilare la funzionalità, quindi le tabelle offline e online vengono ricompilate sotto la nuova funzionalità.
  • Dopo aver verificato le funzionalità di cui è stata eseguita la migrazione, eliminare le funzionalità beta e le relative materializzazioni.

Ricreare le funzionalità

Usare list_beta_feature_views per trovare le funzionalità beta, Feature.clone() per creare una copia non registrata e register_feature registrare nuovamente ogni copia come funzionalità di anteprima pubblica. La clonazione cancella la registrazione, il catalogo e lo schema in modo che la funzionalità possa essere nuovamente registrata.

Per evitare conflitti di nomi, registrare le funzionalità migrate con un nome diverso o in uno schema diverso rispetto alle funzionalità beta. L'esempio seguente registra nuovamente ogni funzionalità nello schema originale con un _migrated suffisso del nome.

# Update this to the catalog whose beta Feature Views you want to migrate.
CATALOG_TO_MIGRATE = "main"

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# 1. Find every beta Feature View in the catalog. Returns Feature objects,
#    scanned across all schemas in the catalog.
beta_features = fe.list_beta_feature_views(catalog_name=CATALOG_TO_MIGRATE)

# Keep each beta feature paired with its migrated counterpart for the next steps.
migrations = []
for beta_feature in beta_features:
    catalog_name, schema_name, leaf_name = beta_feature.full_name.split(".")
    # 2. Clone the feature as an unregistered copy, renamed with a "_migrated" suffix.
    cloned = beta_feature.clone(new_name=f"{leaf_name}_migrated")
    # 3. Re-register the clone as a Public Preview feature.
    migrated = fe.register_feature(
        feature=cloned,
        catalog_name=catalog_name,
        schema_name=schema_name,
    )
    migrations.append((beta_feature, migrated))

Ri-materializzare le funzionalità di cui è stata eseguita la migrazione

Se una funzionalità beta è stata materializzata, ricompilare la controparte anteprima pubblica in modo che le tabelle offline e online vengano ricompilate sotto la nuova funzionalità. Fornire le configurazioni offline e online dell'archivio per la funzionalità migrata e ricostruire il trigger dalla materializzazione esistente della funzionalità beta.

from databricks.feature_engineering.entities import (
    CronSchedule,
    OfflineStoreConfig,
    OnlineStoreConfig,
    TableTrigger,
)

for beta_feature, migrated in migrations:
    # Inspect the beta feature's existing materializations to see what to rebuild and
    # to reconstruct the same trigger.
    trigger = None
    needs_offline = needs_online = False
    for mf in fe.list_materialized_features(feature_name=beta_feature.full_name):
        needs_online = needs_online or bool(mf.is_online)
        needs_offline = needs_offline or not mf.is_online
        # Rebuild the trigger from the materialized feature.
        if mf.cron_schedule_trigger is not None:
            trigger = CronSchedule(
                quartz_cron_expression=mf.cron_schedule_trigger.cron_expression,
                timezone_id="UTC",  # Materialized schedules run in UTC.
            )
        elif mf.table_trigger is not None:
            trigger = TableTrigger()
        elif mf.streaming_mode is not None:
            # Streaming features use StreamingMode, which can be reused as-is.
            trigger = mf.streaming_mode
    if not (needs_offline or needs_online):
        continue  # The beta feature was never materialized.

    catalog_name, schema_name, _ = migrated.full_name.split(".")
    fe.materialize_features(
        features=[migrated],
        offline_config=OfflineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
        )
        if needs_offline
        else None,
        online_config=OnlineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
            online_store_name="my_online_store",
        )
        if needs_online
        else None,
        trigger=trigger,
    )

Note

La materializzazione di ogni funzionalità nella propria materialize_features chiamata crea una pipeline separata. Per ridurre i costi di calcolo, raggruppare le funzionalità che condividono una destinazione offline e online e attivano una singola materialize_features chiamata passandole insieme in features.

Eliminare le funzionalità beta

Warning

Eliminare le funzionalità beta e le relative materializzazioni solo dopo aver verificato che le funzionalità di cui è stata eseguita la migrazione e i relativi dati materializzati siano corretti. L'eliminazione è irreversibile.

Dopo aver verificato le funzionalità di cui è stata eseguita la migrazione, eliminare le materializzazioni di ogni funzionalità beta e quindi la funzionalità beta stessa.

for beta_feature, _ in migrations:
    # Delete the beta feature's materializations first.
    mfs = list(fe.list_materialized_features(feature_name=beta_feature.full_name))
    offline_mfs = [mf for mf in mfs if not mf.is_online]
    if offline_mfs:
        # Aggregation features pair an offline and online table; deleting the offline
        # materialized feature removes its paired online table too.
        for mf in offline_mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    else:
        # Online-only features (ColumnSelection, streaming) have no offline pair; delete
        # the online materialized feature directly.
        for mf in mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    # Then delete the beta feature definition.
    fe.delete_feature(full_name=beta_feature.full_name)
  • Last updated on