Eseguire il training di modelli con visualizzazioni delle funzionalità

Le Feature Views consentono di addestrare modelli con il calcolo delle feature corretto rispetto a uno specifico momento temporale e la ricerca automatica delle feature in fase di inferenza. Per informazioni sulla definizione delle visualizzazioni delle funzionalità, vedere Visualizzazioni delle funzionalità.

Requisiti

• Le funzionalità devono essere create come Feature Views. Vedere Visualizzazioni delle funzionalità.
• Per i requisiti di CustomUDF e FeatureViewSource, consulta il riferimento dell'API Feature Views.

Metodi API

create_training_set()

Dopo aver creato visualizzazioni funzionalità, il passaggio successivo consiste nel creare dati di training per il modello. A tale scopo, passare un set di dati etichettato a create_training_set, che garantisce automaticamente un calcolo accurato dei valori di ciascuna caratteristica nel momento specifico.

Per esempio:

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

Chiamare TrainingSet.load_df per unire i dati di training originali con funzionalità calcolate in tempo dinamico.

L'argomento df deve soddisfare i requisiti seguenti:

• Deve contenere tutte le colonne di entità a cui fanno riferimento le definizioni di funzionalità.
• Deve contenere la colonna timeseries a cui fanno riferimento le definizioni delle funzionalità.
• Deve contenere tutte le colonne dichiarate in qualsiasi RequestSource schema. I tipi vengono convalidati rispetto allo schema dichiarato. Le mancate corrispondenze generano un errore (nessun cast implicito).
• Deve contenere le colonne delle etichette.
• L'insieme dei nomi delle colonne di entità, nomi delle colonne delle serie temporali e nomi delle colonne delle funzionalità richieste devono essere univoci a livello globale in tutte le origini.

Correttezza limitata nel tempo: Per l'aggregazione e le ColumnSelection funzionalità basate su una sorgente tabellare, le funzionalità vengono calcolate usando solo i dati di origine disponibili prima del timestamp di ogni riga, per evitare che dati futuri influiscano sull'addestramento del modello. Per RequestSource le funzionalità, il valore viene ricavato direttamente dalla riga etichettata del DataFrame.

log_model()

Usare MLflow per registrare un modello con metadati delle funzionalità per il rilevamento della derivazione e la ricerca automatica delle funzionalità durante l'inferenza:

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
    extra_pip_requirements: Optional[List[str]] = None, # Optional: Additional serving dependencies
)

Il flavor parametro specifica il modulo MLflow model flavor da usare, ad esempio mlflow.sklearn o mlflow.xgboost.

Con i modelli registrati con un TrainingSet, il tracciamento automatico della discendenza delle caratteristiche usate per l'addestramento avviene automaticamente. Quando il set di training include RequestSource caratteristiche, le RequestSource colonne vengono aggiunte alla firma del modello MLflow in qualità di input richiesti. In questo modo, lo schema API dell'endpoint di servizio riflette i campi che i chiamanti devono fornire in fase di inferenza. Per informazioni dettagliate, vedere Eseguire il training dei modelli con le tabelle delle funzionalità.

Per FeatureViewSource, registra la caratteristica derivata e le sue caratteristiche a monte prima di registrare un modello. Anche gli input di richiesta necessari alle dipendenze transitive sono richiesti al momento dell'inferenza. Vedi Dipendenze UDF personalizzate per i requisiti del pacchetto modello.

score_batch()

Eseguire l'inferenza batch con la ricerca automatica delle funzionalità:

FeatureEngineeringClient.score_batch(
    model_uri: str,                           # URI of logged model
    df: DataFrame,                            # DataFrame with entity keys and timestamps
) -> DataFrame

score_batch utilizza i metadati delle funzionalità archiviate con il modello per calcolare automaticamente le funzionalità corrette al punto nel tempo per l'inferenza, garantendo la coerenza con l'addestramento. Per informazioni dettagliate, vedere Eseguire il training dei modelli con le tabelle delle funzionalità.

Esempio di flusso di lavoro

import mlflow
from databricks.feature_engineering import FeatureEngineeringClient
from sklearn.ensemble import RandomForestClassifier

fe = FeatureEngineeringClient()

# Assume features are registered in UC
# labeled_df should have columns "user_id", "transaction_time", and "is_fraud"

# 1. Create training set using Feature Views
training_set = fe.create_training_set(
    df=labeled_df,
    features=features,
    label="is_fraud",
)

# 2. Load training data with computed features
training_df = training_set.load_df()
X = training_df.drop("is_fraud").toPandas()
y = training_df.select("is_fraud").toPandas().values.ravel()

# 3. Train model
model = RandomForestClassifier().fit(X, y)

# 4. Log model with feature metadata
with mlflow.start_run():
    fe.log_model(
        model=model,
        artifact_path="fraud_model",
        flavor=mlflow.sklearn,
        training_set=training_set,
        registered_model_name="main.ecommerce.fraud_model",
    )

# 5. Batch scoring with automatic feature lookup
# inference_df must contain the same entity and timeseries columns
# used during training. Features are automatically computed.
predictions = fe.score_batch(
    model_uri="models:/main.ecommerce.fraud_model/1",
    df=inference_df,
)
predictions.display()

Training con le funzionalità RequestSource

Quando il modello richiede dati forniti in fase di inferenza (ad esempio i dettagli delle transazioni da una chiamata API), usare RequestSource le funzionalità insieme alle funzionalità supportate da tabelle. Durante il training, le RequestSource colonne vengono estratte dal dataframe etichettato.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    DeltaTableSource, Feature, FieldDefinition, RequestSource,
    ScalarDataType, ColumnSelection,
)

fe = FeatureEngineeringClient()

# RequestSource provides transaction data at inference time
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),
    ]
)

delta_source = DeltaTableSource(
    catalog_name="catalog",
    schema_name="schema",
    table_name="vendor_data",
)

# A column selection feature from the request source (pass-through)
latest_transaction_amount = Feature(
    source=request_source,
    function=ColumnSelection("transaction_amount"),
    name="latest_transaction_amount",
)

# A lookup feature from a delta table
vendor_category = Feature(
    source=delta_source,
    function=ColumnSelection("vendor_category"),
    entity=["vendor_id"],
    timeseries_column="transaction_time",
    name="vendor_category",
)

# labels_df must contain: transaction_id, transaction_time, vendor_id,
# transaction_amount, and the label column.
ts = fe.create_training_set(
    df=labels_df,
    features=[latest_transaction_amount, vendor_category],
    label="is_fraud",
    exclude_columns=["card_id"],
)

import mlflow
from sklearn.ensemble import RandomForestClassifier

with mlflow.start_run():
    training_df = ts.load_df().toPandas()
    X = training_df.drop(columns=["is_fraud"])
    y = training_df["is_fraud"]
    model = RandomForestClassifier().fit(X, y)

    # log_model() adds RequestSource columns to the MLflow model signature
    fe.log_model(
        model=model,
        artifact_path="fraud_model",
        flavor=mlflow.sklearn,
        training_set=ts,
        registered_model_name="catalog.schema.fraud_model",
    )

Trasforma i valori della richiesta con CustomUDF

Per trasformare i dati delle richieste, si usa CustomUDF invece di ColumnSelection. Questo esempio utilizza la funzione di trasformazione logaritmica delle transazioni:

log_transaction_amount = fe.get_feature(
    full_name="main.ecommerce.log_transaction_amount"
)
request_df = spark.createDataFrame(
    [(0.0, 0), (99.0, 1)],
    "transaction_amount DOUBLE, label INT",
)

transaction_training_set = fe.create_training_set(
    df=request_df,
    features=[log_transaction_amount],
    label="label",
)
transaction_training_set.load_df().show()

Il risultato comprende le colonne originali transaction_amount e label, più log_transaction_amount. L'UDF legge transaction_amount da ogni riga. Le colonne di richiesta richieste per il calcolo non possono essere elencate in exclude_columns.

Allena con le funzionalità di FeatureViewSource

FeatureViewSource permette a un CustomUDF di utilizzare gli output di altre funzionalità, comprese le funzionalità derivate. Trasmetti gli output desiderati a create_training_set. Non è necessario elencare le loro dipendenze intermedie.

Per ogni riga di input, Azure Databricks risolve l'intero grafo di dipendenza:

1. Calcola le funzionalità a monte supportate da tabelle utilizzando le loro chiavi di entità, timestamp e definizioni di finestre. Utilizza materializzazioni offline compatibili quando disponibili.
2. Legge le colonne richieste richieste dal DataFrame di input e valuta le funzionalità supportate dalle richieste.
3. Valuta le caratteristiche derivate in ordine di dipendenze, così che ogni UDF riceva i suoi risultati a monte.

La caratteristica derivata non introduce un'altra finestra temporale o ricerca puntuale nel tempo. Le sue caratteristiche a monte mantengono la loro semantica temporale. Il DataFrame deve contenere l'entità, il timestamp e le colonne di richiesta necessarie a chi è a monte, anche quando viene richiesta solo la caratteristica derivata finale.

Ad esempio, usa la funzione margin registrata, che combina revenue_sum_7d e cost_sum_7d:

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()
margin = fe.get_feature(full_name="main.ecommerce.margin")

# labeled_df contains customer_id, event_time, and label.
training_set = fe.create_training_set(
    df=labeled_df,
    features=[margin],
    label="label",
    exclude_columns=["customer_id", "event_time"],
)
training_df = training_set.load_df()

Il risultato contiene label e margin. Le caratteristiche di ricavo e costo vengono calcolate ma non vengono restituite come colonne aggiuntive. Per includere i ricavi nei dati di addestramento, recuperali con revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") e passa features=[margin, revenue]. Questo vale anche per le catene multilivello: richiedere la caratteristica finale non restituisce ogni output intermedio.

Puoi combinare funzionalità supportate da richieste, con supporto da tabella e derivate nello stesso features elenco. Per combinare i loro valori all'interno di un unico UDF, rappresenta i valori delle richieste come caratteristiche e riferiscili insieme a funzionalità supportate da tabella in un FeatureViewSource.

A scopo di sperimentazione, creare oggetti locali Feature, compreso il relativo grafo upstream, senza registrarli. Usare create_training_set, opzionalmente con label=None, per ispezionare i risultati. compute_features non supporta RequestSource né FeatureViewSource.

Dipendenze personalizzate UDF

Per il calcolo offline, dichiarare pacchetti Python nella clausolaENVIRONMENT UDF del Unity Catalog. Installare un pacchetto solo nel notebook non lo installa nell'ambiente UDF.

Per il servizio dei modelli, passa anche esplicitamente i pacchetti richiesti a log_model. Né l'UDF ENVIRONMENT né una specifica di funzionalità denominata che contiene Feature View forniscono automaticamente questi requisiti del modello. Includi le dipendenze necessarie per le UDF a monte, nonché gli output della funzionalità richiesta.

Dopo aver addestrato un modello scikit-learn su transaction_training_set.load_df(), registralo usando lo stesso set di addestramento. Includi NumPy e un pacchetto di ricerca compatibile:

import mlflow

fe.log_model(
    model=model,
    artifact_path="transaction_model",
    flavor=mlflow.sklearn,
    training_set=transaction_training_set,
    registered_model_name="main.ecommerce.transaction_model",
    extra_pip_requirements=[
        "numpy==1.26.4",
        "databricks-feature-lookup>=1.15.0",
    ],
)

Gli endpoint di serving dovrebbero rilevare automaticamente la versione 1.15.0 o successiva di databricks-feature-lookup, che supporta il calcolo on-demand delle UDF di Unity Catalog per le funzionalità RequestSource. Mantieni le versioni dei tuoi pacchetti UDF coerenti tra ambienti offline e di servizio per evitare differenze nei valori calcolati. Per gli endpoint Feature Serving senza modello, dichiarare i pacchetti su create_feature_spec invece. Vedi Aggiungi dipendenze di Python.

Training con funzionalità di streaming

Quando si definisce un flusso, Databricks gestisce una pipeline di inserimento che scrive i dati di flusso in una tabella Delta. create_training_set legge da questa tabella di inserimento ed esegue join temporizzato sul dataframe etichettato, proprio come le funzionalità batch di un oggetto DeltaTableSource. Per dettagli sulla configurazione dell'ingestione, sul backfill e sulla deduplicazione, consulta Ingestione e backfill.

Example

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

fe = FeatureEngineeringClient()

# Define a streaming feature
stream_source = StreamSource(full_name="my_catalog.my_schema.my_stream")

streaming_feature = Feature(
    name="user_purchase_sum",
    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)),
    ),
)

# Create training set — reads from the ingestion table
# labeled_df must contain "user_id", "event_time", and label columns.
# Entity and timeseries columns use leaf node names (not value. prefixes).
training_set = fe.create_training_set(
    df=labeled_df,
    features=[streaming_feature],
    label="is_fraud",
)

training_df = training_set.load_df()

Combinazione di funzionalità batch e streaming

Le funzionalità batch e di streaming possono essere usate insieme nello stesso set di training e nello stesso modello. Al momento dell’erogazione, le caratteristiche batch vengono recuperate dagli archivi offline o online, mentre le caratteristiche di streaming vengono recuperate dagli archivi online.

training_set = fe.create_training_set(
    df=labeled_df,
    features=[batch_feature, streaming_feature],
    label="is_fraud",
)

Il modello registrato con log_model() recupera le feature dallo store online e configura la firma del modello per entrambi i tipi di sorgente.

Che cosa raggiunge il modello non elaborato al momento della gestione

Il wrapper del modello di Feature Store filtra le colonne prima di passarle al modello non elaborato: