Eseguire il training di modelli con visualizzazioni delle funzionalità

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.

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

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
)

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à.

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",
    )

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 informazioni dettagliate sulla configurazione dell'inserimento, il backfill e la deduplicazione, vedere Inserimento 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:

Tipo di colonna Raggiunge il modello interno?
Output espliciti delle funzionalità (ColumnSelection, aggregazione) Sì
RequestSource colonne dichiarate come caratteristiche Sì
Colonne di entità (chiavi di ricerca) No (a meno che non sia dichiarato in modo esplicito come funzionalità)
Colonne serie temporali No (a meno che non sia dichiarato in modo esplicito come funzionalità)
  • Last updated on