Entraîner des modèles avec des vues de fonctionnalités

info

Aperçu

Cette fonctionnalité est en Aperçu public. Les administrateurs du Workspace peuvent contrôler l'accès à cette fonctionnalité à partir de la page Previews . Consultez Gérer les aperçus Databricks.

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

Exigences

• Les fonctionnalités doivent être créées en tant que vues de fonctionnalités. Consultez Vues de fonctionnalités.

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'entraînement pour votre modèle. Pour ce faire, transmettez un dataset étiqueté à create_training_set, ce qui garantit automatiquement un calcul précis à un moment donné de chaque valeur de fonctionnalité.

Par exemple :

Python

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 les données d'entraînement d'origine avec des fonctionnalités calculées dynamiquement et ponctuelles.

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érie temporelle référencée par les définitions de fonctionnalités.
• Doit contenir toutes les colonnes déclarées dans n'importe quel schéma RequestSource. Les types sont validés par rapport au schéma déclaré. Les incohérences entraînent une erreur (pas de conversion implicite).
• Devrait contenir la/les colonne(s) d'étiquettes.
• L'ensemble des noms de colonne d'entité, des noms de colonne de série chronologique et des noms de colonne de fonctionnalité de requête doit être unique au niveau mondial sur toutes les sources.

**Exactitude ponctuelle :** Pour ColumnSelection les fonctionnalités d'agrégation et basées sur une source de table, les fonctionnalités sont calculées en utilisant uniquement les données sources disponibles avant le Timestamp de chaque ligne, afin d'éviter les fuites de données futures dans l'entraînement du modèle. Pour les fonctionnalités RequestSource, la valeur est prise directement de la ligne du DataFrame étiquetée.

log_model()

Utilisez MLflow pour enregistrer un modèle avec des métadonnées de fonctionnalités pour le suivi de la lignée et la recherche automatique de fonctionnalités pendant l'inférence :

Python

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 parameter spécifie le module de saveur de modèle MLflow à utiliser, tel que mlflow.sklearn ou mlflow.xgboost.

Les modèles enregistrés avec un TrainingSet suivent automatiquement la provenance des fonctionnalités utilisées dans l'entraînement. Lorsque l'ensemble d'entraînement comprend RequestSource fonctionnalités, les colonnes RequestSource sont ajoutées à la signature du modèle MLflow en tant qu'entrées requises. Ceci garantit que le schéma d'API de l'Endpoint de diffusion reflète les champs que les appelants doivent fournir au moment de l'inférence. Pour plus de détails, consultez Entraîner des modèles avec des tables de fonctionnalités.

score_batch()

Effectuez une inférence par batch avec recherche automatique de caractéristiques :

Python

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 de fonctionnalité stockées avec le modèle pour compute automatiquement des fonctionnalités correctes au moment de l'inférence, assurant ainsi la cohérence avec l'entraînement. Pour plus de détails, consultez Entraîner des modèles avec des tables de fonctionnalités.

Exemple de workflow

Python

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

Entraînement avec les fonctionnalités RequestSource

Lorsque votre modèle nécessite des données fournies au moment de l'inférence (telles que les détails de transaction provenant d'un appel d'API), utilisez les fonctionnalités RequestSource en même temps que les fonctionnalités basées sur des tables. Pendant l'entraînement, RequestSource colonnes sont extraites du DataFrame étiqueté.

Python

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

Entraînement avec des fonctionnalités de streaming

Lorsque vous définissez un Stream, Databricks gère un pipeline d'ingestion qui écrit les données de stream dans une table Delta. create_training_set lit à partir de cette table d'ingestion et effectue des jointures à un instant T sur votre DataFrame étiqueté, tout comme les fonctionnalités de batch d'un DeltaTableSource. Pour plus de détails sur la configuration de l'ingestion, le remplissage et la déduplication, consultez Ingestion et remplissage.

Exemple

Python

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

Mixage des fonctionnalités de batch et de streaming

Les fonctionnalités batch et streaming peuvent être utilisées ensemble dans le même ensemble d'entraînement et le même modèle. Au moment de la diffusion, les fonctionnalités batch sont recherchées dans les magasins hors ligne ou en ligne, et les fonctionnalités streaming sont recherchées dans les magasins en ligne.

Python

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

Le modèle journalisé avec log_model() effectue des recherches de fonctionnalités à partir du 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 de modèle du Magasin de fonctionnalités 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 comme fonctionnalités	Oui
Colonnes d'entité (clés de recherche)	Non (à moins que ce ne soit explicitement déclaré comme une fonctionnalité)
Colonnes de séries chronologiques	Non (à moins que ce ne soit explicitement déclaré comme une fonctionnalité)

Type de colonne	Atteint le modèle interne ?
Sorties de fonctionnalités explicites (ColumnSelection, agrégation)	Oui
RequestSource colonnes déclarées comme fonctionnalités	Oui
Colonnes d'entité (clés de recherche)	Non (à moins que ce ne soit explicitement déclaré comme une fonctionnalité)
Colonnes de séries chronologiques	Non (à moins que ce ne soit explicitement déclaré comme une fonctionnalité)