Entraîner des modèles avec des vues de fonctionnalités
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.
- Pour les exigences de
CustomUDFet deFeatureViewSource, voir la 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'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 :
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 :
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 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.
Pour FeatureViewSource, enregistrez la fonctionnalité dérivée et ses fonctionnalités en amont avant d’enregistrer un modèle. Les entrées de requête requises par les dépendances transitives sont également requises au moment de l’inférence. Consultez Custom UDF dependencies pour connaître les exigences relatives aux package de modèles.
score_batch()
Effectuez une inférence par batch avec recherche automatique de caractéristiques :
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
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é.
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 requête avec CustomUDF
Pour transformer les données de requête, utilisez CustomUDF à la place de ColumnSelection. Cet exemple utilise la fonctionnalité de transformation des Logs de 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 originales transaction_amount et label, ainsi que log_transaction_amount. La UDF lit transaction_amount à partir de chaque ligne. Les colonnes de requête nécessaires pour le calcul ne peuvent pas être répertoriées dans exclude_columns.
Entraîner avec les caractéristiques FeatureViewSource
FeatureViewSource permet à un ou une CustomUDF de consommer d'autres sorties de fonctionnalités, y compris des fonctionnalités dérivées. Transmettez les sorties que vous souhaitez à create_training_set. Vous n'avez pas besoin de répertorier leurs dépendances intermédiaires.
Pour chaque ligne d'entrée, Databricks résout le graphe de dépendances complet :
- Il calcule les fonctionnalités en amont adossées à des tables à l’aide de leurs clés d’entité, de leurs horodatages et de leurs définitions de fenêtre. Il utilise des matérialisations hors ligne compatibles lorsqu’elles sont disponibles.
- Lit les colonnes de requêtes requises à partir du DataFrame d’entrée et évalue les fonctionnalités basées sur les requêtes.
- Évalue les fonctionnalités dérivées dans l’ordre des dépendances, de sorte que chaque UDF reçoive ses résultats en amont.
La caractéristique dérivée n'introduit pas d'autre fenêtre temporelle ou recherche à un moment précis. Ses caractéristiques en amont conservent leurs propres sémantiques temporelles. Le DataFrame doit contenir les colonnes d'entité, de timestamp et de requête requises par ces flux en amont, même lorsqu'une seule caractéristique dérivée finale est demandée.
Par exemple, utilisez la fonctionnalité de 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 renvoyé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 transmettez features=[margin, revenue]. Cela s'applique également aux chaînes à plusieurs niveaux : la demande de la caractéristique finale ne renvoie pas chaque sortie intermédiaire.
Vous pouvez combiner des caractéristiques basées sur des requêtes, basées sur des tables et dérivées dans la même liste features. Pour combiner leurs valeurs dans une seule UDF, représentez les valeurs de requête en tant que caractéristiques et référencez-les aux côtés des caractéristiques basées sur des tables dans un FeatureViewSource.
Pour l’expérimentation, construisez des objets Feature locaux, y compris leur graphe en amont, sans les enregistrer. Utilisez create_training_set, éventuellement avec label=None, pour inspecter les résultats. compute_features ne prend pas en charge RequestSource ou FeatureViewSource.
La limite de cinq appels d'UDF Unity Catalog par requête s'applique également aux requêtes d'entraînement. Comptez les appels d'UDF nécessaires pour l'ensemble du graphe de dépendances, et pas seulement pour les fonctionnalités demandées comme sorties. Cette limite de requête est distincte de la limite de profondeur du graphe.
Dépendances d’UDF personnalisées
Pour le calcul hors ligne, déclarez les packages Python dans la clauseENVIRONMENT de l'UDF Unity Catalog. L'installation d'un package dans le notebook seul ne l'installe pas dans l'environnement de l'UDF.
Pour le service de modèles, transmettez également les package requis explicitement à log_model. Ni l’UDF ENVIRONMENT ni une spécification de fonctionnalité nommée contenant des vues de fonctionnalités ne fournissent automatiquement ces exigences de modèle. Incluez les dépendances nécessaires aux UDF en amont ainsi que les sorties de fonctionnalités demandées.
Après avoir entraîné un modèle scikit-learn sur transaction_training_set.load_df(), enregistrez-le dans les Logs avec ce même jeu d’entraînement. Incluez NumPy et un package 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 endpoints de service doivent automatiquement intégrer databricks-feature-lookup version 1.15.0 ou ultérieure, qui prend en charge le calcul d'UDF Unity Catalog à la demande pour les fonctionnalités RequestSource. Maintenez la cohérence des versions de vos packages UDF entre les environnements hors ligne et de service afin d'éviter des différences dans les valeurs calculées. Pour les endpoints Feature Serving sans modèle, déclarez plutôt les packages sur create_feature_spec. Voir Ajouter des dépendances Python.
Entraînement avec des fonctionnalités de streaming
Lorsque vous définissez un Stream, Databricks gère un pipeline d'ingestion qui écrit des données de stream dans une table Delta. create_training_set lit à partir de cette table d'ingestion et effectue des jointures temporelles (point-in-time joins) sur votre DataFrame étiqueté, tout comme les features de batch provenant d'un DeltaTableSource. Pour plus de détails sur la configuration de l'ingestion, le remplissage (backfill) et la déduplication, consultez Ingestion et remplissage.
Exemple
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.
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 ( | Oui |
| 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é) |