Entraîner des modèles à l’aide de Feature Views

Important

Cette fonctionnalité est disponible en préversion publique. Les administrateurs d’espace de travail peuvent contrôler l’accès à cette fonctionnalité à partir de la page Aperçus . Consultez Gérer les préversions d’Azure Databricks.

Les vues de fonctionnalités vous permettent d’entraîner des modèles avec un calcul de fonctionnalité correct dans le temps et une recherche automatique des fonctionnalités lors de l’inférence. Pour plus d’informations sur la définition des vues de fonctionnalités, consultez Vues de fonctionnalité.

Requirements

Méthodes d’API

create_training_set()

Après avoir créé des vues de fonctionnalités, l’étape suivante consiste à créer des données d’apprentissage pour votre modèle. Pour ce faire, transmettez un jeu de données étiqueté à create_training_set, ce qui garantit automatiquement un calcul précis dans le temps de chaque valeur de fonctionnalité.

Par exemple:

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

Appelez TrainingSet.load_df pour joindre des données d'entraînement d'origine avec des fonctionnalités calculées dynamiquement à un moment donné.

L’argument df doit répondre aux exigences suivantes :

  • Doit contenir toutes les colonnes d’entité référencées par les définitions de fonctionnalités.
  • Doit contenir la colonne de séries temporelles référencée par les définitions de caractéristiques.
  • Doit contenir toutes les colonnes déclarées dans n’importe quel RequestSource schéma. Les types sont validés par rapport au schéma déclaré. Les incompatibilités déclenchent une erreur (aucune conversion implicite).
  • Doit contenir des colonnes d’étiquettes.
  • L’ensemble de noms de colonnes d’entité, de noms de colonnes timeseries et de noms de colonnes de fonctionnalité de requête doit être globalement unique sur toutes les sources.

Exactitude temporelle : Pour l’agrégation et ColumnSelection les caractéristiques soutenues par une source de données tabulaire, les caractéristiques sont calculées en utilisant uniquement les données sources disponibles avant le moment marqué par l’horodatage de chaque ligne, afin d’éviter les fuites de données futures lors de l’entraînement du modèle. Pour les caractéristiques RequestSource, la valeur est extraite directement de la ligne étiquetée du DataFrame.

log_model()

Utilisez MLflow pour consigner un modèle avec des métadonnées de fonctionnalité pour le suivi de la traçabilité et la recherche automatique des fonctionnalités pendant l’inférence :

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
)

Le flavor paramètre spécifie le module de saveur du modèle MLflow à utiliser, tel que mlflow.sklearn ou mlflow.xgboost.

Les modèles enregistrés avec un TrainingSet suivent automatiquement la traçabilité des variables utilisées lors de l’apprentissage. Lorsque le jeu d’apprentissage inclut des caractéristiques RequestSource, les colonnes RequestSource sont ajoutées à la signature du modèle MLflow en tant qu’entrées requises. Cela garantit que le schéma d’API du point de terminaison de service reflète les champs que les appelants doivent fournir au moment de l’inférence. Pour en savoir plus, consultez Entraîner des modèles avec des tables de caractéristiques.

score_batch()

Effectuez une inférence par lots avec la recherche automatique des fonctionnalités :

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

score_batch utilise les métadonnées des caractéristiques stockées avec le modèle pour calculer automatiquement les caractéristiques exactes à un moment donné pour l'inférence, ce qui garantit la cohérence avec l'entraînement. Pour en savoir plus, consultez Entraîner des modèles avec des tables de caractéristiques.

Exemple de flux de travail

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()

Formation avec les fonctionnalités RequestSource

Lorsque votre modèle nécessite des données fournies au moment de l’inférence (par exemple, les détails de transaction à partir d’un appel d’API), utilisez les caractéristiques RequestSource en même temps que les caractéristiques de table. Pendant l’entraînement, les colonnes RequestSource sont extraites du DataFrame étiqueté.

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

Formation avec fonctionnalités de diffusion en continu

Lorsque vous définissez un flux, Databricks gère un pipeline d’ingestion qui écrit des données de flux dans une table Delta. create_training_set lit depuis cette table d’ingestion et effectue des jointures « point-in-time » avec votre DataFrame étiqueté, tout comme les variables batch d’un DeltaTableSource. Pour plus d’informations sur la configuration de l’ingestion, la réintégration des données et la déduplication, consultez Ingestion et réintégration des données.

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()

Combinaison de fonctionnalités de traitement par lots et de diffusion en continu

Les fonctionnalités batch et de diffusion en continu peuvent être utilisées ensemble dans le même jeu d’entraînement et le même modèle. Au moment du service, les fonctionnalités batch sont recherchées à partir de magasins hors connexion ou en ligne, et les fonctionnalités de diffusion sont recherchées à partir de magasins en ligne.

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

Le modèle enregistré avec log_model() effectue des recherches de variables dans le magasin en ligne et configure la signature du modèle pour les deux types de source.

Ce qui atteint le modèle brut au moment du service

Le wrapper du modèle Feature Store filtre les colonnes avant de les transmettre au modèle brut.

Type de colonne Atteint le modèle interne ?
Sorties de fonctionnalités explicites (ColumnSelection, agrégation) Oui
RequestSource colonnes déclarées en tant que fonctionnalités Oui
Colonnes d’entité (clés de recherche) Non (sauf si elle est déclarée explicitement en tant que fonctionnalité)
Colonnes de la série chronologique Non (sauf si elle est déclarée explicitement en tant que fonctionnalité)
  • Last updated on