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

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

• Les Features doivent être créées sous forme de Feature Views. Consultez Vues des fonctionnalités.
• Pour les CustomUDF prérequis et FeatureViewSource de Feature Views, consultez la documentation de référence de l’API Feature Views.

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

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.

Pour FeatureViewSource, enregistrez la caractéristique dérivée et ses caractéristiques en amont avant d’enregistrer un modèle. Les entrées de requête nécessaires aux dépendances transitives sont également requises au moment de l’inférence. Voir Dépendances UDF personnalisées pour les exigences du paquet modèle.

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

Transformer les valeurs de la requête avec CustomUDF

Pour transformer les données de requête, utilisez CustomUDF au lieu de ColumnSelection. Cet exemple utilise la fonction de transformation logarithmique des transactions :

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

Le résultat comprend les colonnes d’origine transaction_amount et label, ainsi que log_transaction_amount. L’UDF lit transaction_amount dans chaque ligne. Les colonnes de requête requises pour le calcul ne peuvent pas être listées dans exclude_columns.

Entraînez-vous avec les fonctionnalités de FeatureViewSource

FeatureViewSource permet à un CustomUDF de consommer les sorties d’autres fonctionnalités, y compris les fonctionnalités dérivées. Transmettez les résultats souhaités à create_training_set. Vous n’avez pas besoin d’énumérer leurs dépendances intermédiaires.

Pour chaque ligne d’entrée, Azure Databricks résout le graphe de dépendance complet :

1. Calcule les fonctionnalités en amont sauvegardées par table à l’aide de leurs clés d’entité, horodatages et définitions de fenêtres. Il utilise des matérialisations hors ligne compatibles lorsque cela est disponible.
2. Lit les colonnes de requête requises du DataFrame d’entrée et évalue les fonctionnalités soutenues par requête.
3. Évalue les caractéristiques dérivées dans l’ordre des dépendances, afin que chaque UDF reçoive ses résultats en amont.

La caractéristique dérivée n’introduit pas une autre fenêtre temporelle ni une recherche à un moment donné. Ses caractéristiques en amont conservent leur propre sémantique temporelle. Le DataFrame doit contenir l’entité, l’horodatage et les colonnes de requête nécessaires à ceux en amont, même lorsque seule la dernière caractéristique dérivée est demandée.

Par exemple, utilisez la fonction marge enregistrée, qui combine revenue_sum_7d et 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()

Le résultat contient label et margin. Les caractéristiques de revenus et de coûts sont calculées mais ne sont pas retournées sous forme de colonnes supplémentaires. Pour inclure les revenus dans les données d’entraînement, récupérez-les avec revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") et passez features=[margin, revenue]. Cela s’applique également aux chaînes multi-niveaux : demander la dernière fonctionnalité ne retourne pas chaque sortie intermédiaire.

Vous pouvez combiner des fonctionnalités basées sur des requêtes, des fonctionnalités basées sur des tables et des fonctionnalités dérivées dans la même liste features. Pour combiner leurs valeurs dans un seul UDF, représentez les valeurs de requête comme des caractéristiques et référencez-les aux côtés de fonctionnalités à base de tableau dans un FeatureViewSourcefichier .

Pour l’expérimentation, construisez des objets locaux Feature , y compris leur graphe en amont, sans les enregistrer. Utilisez create_training_set, optionnellement avec label=None, pour inspecter les résultats. compute_features ne supporte RequestSource pas ni FeatureViewSource.

Dépendances UDF personnalisées

Pour le calcul hors ligne, déclarez les packages Python dans la ENVIRONMENT clause UDF du catalogue Unity. Installer un package dans le seul notebook ne l’installe pas dans l’environnement UDF.

Pour le service de modèles, passez également explicitement les paquets requis à log_model. Ni la ENVIRONMENT de l’UDF ni la spécification de caractéristique nommée contenant les Feature Views ne répondent automatiquement à ces exigences de modèle. Incluez les dépendances nécessaires aux UDF situées en amont ainsi que les sorties des fonctionnalités demandées.

Après avoir entraîné un modèle scikit-learn sur transaction_training_set.load_df(), enregistrez-le avec ce même ensemble d’entraînement. Incluez NumPy et un paquet de recherche compatible :

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

Les points de terminaison de mise en service doivent automatiquement détecter databricks-feature-lookup la version 1.15.0 ou ultérieure, qui prend en charge le calcul UDF Unity Catalog à la demande pour les fonctionnalités RequestSource. Gardez les versions de vos packages UDF cohérentes entre les environnements hors ligne et de service afin d’éviter les différences de valeurs calculées. Pour les points de terminaison Feature Serving sans modèle, déclarez plutôt les paquets sur create_feature_spec . Voir Ajouter des dépendances Python.

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 de détails sur la configuration d’ingestion, le remblai et la déduplication, voir Ingestion et remblai.

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.