Utiliser des caractéristiques pour entraîner des modèles

  • Article

Cet article explique comment entraîner des modèles à l’aide de l’ingénierie des fonctionnalités dans Unity Catalog ou dans le magasin de caractéristiques de l’espace de travail local. Vous devez d’abord créer un jeu de données d’entraînement, qui définit les fonctionnalités à utiliser et comment les associer. Ensuite, quand vous entraînez un modèle, celui-ci conserve les références aux caractéristiques.

Lorsque vous entraînez un modèle à l’aide de l’ingénierie de caractéristiques dans Unity Catalog, vous pouvez afficher la traçabilité du modèle dans l’explorateur de catalogue. Les tables et fonctions utilisées pour créer le modèle sont automatiquement suivies et affichées. Consultez Afficher la traçabilité du magasin de caractéristiques.

Quand vous utilisez le modèle à des fins d’inférence, vous pouvez faire en sorte qu’il récupère les valeurs des caractéristiques du magasin de caractéristiques. Vous pouvez également servir le modèle avec Service de modèle et cela recherche automatiquement les fonctionnalités publiées dans les magasins en ligne. Les modèles de magasin de fonctionnalités sont également compatibles avec l’interface pyfunc MLflow, ce qui vous permet d’utiliser MLflow pour traiter les inférences par lot avec des tables de fonctionnalités.

Si votre modèle utilise des variables d’environnement, découvrez comment les utiliser lors du service du modèle en ligne dans l’article Configuration de l’accès aux ressources à partir de points de terminaison de service de modèle.

Création d’un jeu de données d’entraînement

Pour sélectionner des fonctionnalités spécifiques à partir d’une table de fonctionnalités pour l’entraînement de modèles, créez un jeu de données d’entraînement à l’aide de l’API FeatureEngineeringClient.create_training_set (pour la caractérisation dans le catalogue Unity) ou FeatureStoreClient.create_training_set (pour le magasin de fonctionnalités de l’espace de travail) et d’un objet appelé FeatureLookup. Un FeatureLookup spécifie chaque caractéristique à utiliser dans le jeu d’entraînement, y compris le nom de la table de caractéristiques, le ou les noms des caractéristiques ainsi que la ou les clés à utiliser lors de la jointure de la table de caractéristiques avec le DataFrame passé à create_training_set. Pour plus d’informations, consultez Recherche de fonctionnalités.

Utilisez le paramètre feature_names lorsque vous créez FeatureLookup. feature_names accepte un nom de caractéristique unique, une liste de noms de caractéristiques ou None pour rechercher toutes les caractéristiques (à l’exception des clés primaires) dans la table de caractéristiques au moment de la création du jeu d’entraînement.

Remarque

Le type et l’ordre de colonnes lookup_key de ce DataFrame doit correspondre au type et ordre des clés primaires (hors clés de timestamp) de la table de caractéristiques de référence.

Cet article contient des exemples de code pour les deux versions de la syntaxe.

Dans cet exemple, le DataFrame retourné par trainingSet.load_df contient une colonne pour chaque caractéristique dans feature_lookups. Il conserve toutes les colonnes du DataFrame fournie à create_training_set, à l’exception de celles exclues à l’aide de exclude_columns.

Caractérisation dans le catalogue Unity

from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup

# The model training uses two features from the 'customer_features' feature table and
# a single feature from 'product_features'
feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['total_purchases_30d', 'total_purchases_7d'],
      lookup_key='customer_id'
    ),
    FeatureLookup(
      table_name='ml.recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

fe = FeatureEngineeringClient()

# Create a training set using training DataFrame and features from Feature Store
# The training DataFrame must contain all lookup keys from the set of feature lookups,
# in this case 'customer_id' and 'product_id'. It must also contain all labels used
# for training, in this case 'rating'.
training_set = fe.create_training_set(
  df=training_df,
  feature_lookups=feature_lookups,
  label='rating',
  exclude_columns=['customer_id', 'product_id']
)

training_df = training_set.load_df()

Magasin de fonctionnalités de l'espace de travail

from databricks.feature_store import FeatureLookup, FeatureStoreClient

# The model training uses two features from the 'customer_features' feature table and
# a single feature from 'product_features'
feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['total_purchases_30d', 'total_purchases_7d'],
      lookup_key='customer_id'
    ),
    FeatureLookup(
      table_name='recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

fs = FeatureStoreClient()

# Create a training set using training DataFrame and features from Feature Store
# The training DataFrame must contain all lookup keys from the set of feature lookups,
# in this case 'customer_id' and 'product_id'. It must also contain all labels used
# for training, in this case 'rating'.
training_set = fs.create_training_set(
  df=training_df,
  feature_lookups=feature_lookups,
  label='rating',
  exclude_columns=['customer_id', 'product_id']
)

training_df = training_set.load_df()

Créer un jeu d’entraînement (TrainingSet) quand les clés de recherche ne correspondent pas aux clés primaires

Utilisez l’argument lookup_key dans FeatureLookup pour le nom de colonne dans le jeu d’entraînement. create_training_set effectue une jointure ordonnée entre les colonnes du jeu d’entraînement spécifié dans l’argument lookup_key en utilisant l’ordre dans lequel les clés primaires ont été spécifiées lors de la création de la table de caractéristiques.

Dans cet exemple, recommender_system.customer_features possède les clés primaires suivantes : customer_id, dt.

La table de caractéristiques recommender_system.product_features a la clé primaire product_id.

Si training_df a les colonnes suivantes :

  • cid
  • transaction_dt
  • product_id
  • rating

Le code suivant permet de créer les recherches de caractéristiques correctes pour le TrainingSet :

Caractérisation dans le catalogue Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['total_purchases_30d', 'total_purchases_7d'],
      lookup_key=['cid', 'transaction_dt']
    ),
    FeatureLookup(
      table_name='ml.recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

Magasin de fonctionnalités de l'espace de travail

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['total_purchases_30d', 'total_purchases_7d'],
      lookup_key=['cid', 'transaction_dt']
    ),
    FeatureLookup(
      table_name='recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

Quand create_training_set est appelée, elle crée un jeu de données d’entraînement en effectuant une jointure gauche, en joignant les tables recommender_system.customer_features et training_df au moyen des clés (customer_id, dt) correspondant à (cid, transaction_dt), comme illustré dans le code suivant :

Caractérisation dans le catalogue Unity

customer_features_df = spark.sql("SELECT * FROM ml.recommender_system.customer_features")
product_features_df = spark.sql("SELECT * FROM ml.recommender_system.product_features")

training_df.join(
  customer_features_df,
  on=[training_df.cid == customer_features_df.customer_id,
      training_df.transaction_dt == customer_features_df.dt],
  how="left"
).join(
  product_features_df,
  on="product_id",
  how="left"
)

Magasin de fonctionnalités de l'espace de travail

customer_features_df = spark.sql("SELECT * FROM recommender_system.customer_features")
product_features_df = spark.sql("SELECT * FROM recommender_system.product_features")

training_df.join(
  customer_features_df,
  on=[training_df.cid == customer_features_df.customer_id,
      training_df.transaction_dt == customer_features_df.dt],
  how="left"
).join(
  product_features_df,
  on="product_id",
  how="left"
)

Créer un TrainingSet qui contient deux caractéristiques portant le même nom que des tables de caractéristiques différentes

Utilisez l’argument facultatif output_name dans la FeatureLookup. Le nom fourni est utilisé à la place du nom de caractéristique dans le DataFrame retourné par TrainingSet.load_df. Par exemple, avec le code suivant, le DataFrame retourné par training_set.load_df comprend les colonnes customer_height et product_height.

Caractérisation dans le catalogue Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['height'],
      lookup_key='customer_id',
      output_name='customer_height',
    ),
    FeatureLookup(
      table_name='ml.recommender_system.product_features',
      feature_names=['height'],
      lookup_key='product_id',
      output_name='product_height'
    ),
  ]

fe = FeatureEngineeringClient()

with mlflow.start_run():
  training_set = fe.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id']
  )
  training_df = training_set.load_df()

Magasin de fonctionnalités de l'espace de travail

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['height'],
      lookup_key='customer_id',
      output_name='customer_height',
    ),
    FeatureLookup(
      table_name='recommender_system.product_features',
      feature_names=['height'],
      lookup_key='product_id',
      output_name='product_height'
    ),
  ]

fs = FeatureStoreClient()

with mlflow.start_run():
  training_set = fs.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id']
  )
  training_df = training_set.load_df()

Créer un TrainingSet à l’aide de la même fonctionnalité plusieurs fois

Pour créer un TrainingSet à l’aide de la même fonctionnalité jointe par des clés de recherche différentes, utilisez plusieurs FeatureLookups. Utilisez un unique output_name pour chaque sortie FeatureLookup.

Caractérisation dans le catalogue Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.taxi_data.zip_features',
      feature_names=['temperature'],
      lookup_key=['pickup_zip'],
      output_name='pickup_temp'
    ),
    FeatureLookup(
      table_name='ml.taxi_data.zip_features',
      feature_names=['temperature'],
      lookup_key=['dropoff_zip'],
      output_name='dropoff_temp'
    )
  ]

Magasin de fonctionnalités de l'espace de travail

feature_lookups = [
    FeatureLookup(
      table_name='taxi_data.zip_features',
      feature_names=['temperature'],
      lookup_key=['pickup_zip'],
      output_name='pickup_temp'
    ),
    FeatureLookup(
      table_name='taxi_data.zip_features',
      feature_names=['temperature'],
      lookup_key=['dropoff_zip'],
      output_name='dropoff_temp'
    )
  ]

Créer un TrainingSet pour les modèles Machine Learning non supervisés

Définissez label=None lors de la création d’un TrainingSet pour les modèles d’entraînement non supervisés. Par exemple, le TrainingSet suivant peut être utilisé pour mettre en cluster différents clients dans des groupes en fonction de leurs intérêts :

Caractérisation dans le catalogue Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['interests'],
      lookup_key='customer_id',
    ),
  ]

fe = FeatureEngineeringClient()
with mlflow.start_run():
  training_set = fe.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label=None,
    exclude_columns=['customer_id']
  )

  training_df = training_set.load_df()

Magasin de fonctionnalités de l'espace de travail

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['interests'],
      lookup_key='customer_id',
    ),
  ]

fs = FeatureStoreClient()
with mlflow.start_run():
  training_set = fs.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label=None,
    exclude_columns=['customer_id']
  )

  training_df = training_set.load_df()

Entraîner des modèles et effectuer une inférence par lots avec des tables de caractéristiques

Quand vous entraînez un modèle en utilisant des caractéristiques du magasin de caractéristiques, le modèle conserve les références aux caractéristiques. Quand vous utilisez le modèle à des fins d’inférence, vous pouvez faire en sorte qu’il récupère les valeurs des caractéristiques du magasin de caractéristiques. Vous devez fournir la ou les clés primaires des caractéristiques utilisées dans le modèle. Le modèle récupère les caractéristiques dont il a besoin du magasin de caractéristiques dans votre espace de travail. Il joint ensuite les valeurs de caractéristique en fonction des besoins lors du scoring.

Pour prendre en charge la recherche de caractéristiques au moment de l’inférence :

  • Vous devez enregistrer le modèle à l’aide de la méthode log_model de FeatureEngineeringClient(pour la caractérisation dans le catalogue Unity) ou de FeatureStoreClient (pour le magasin de fonctionnalités de l’espace de travail).
  • Vous devez utiliser le DataFrame retourné par TrainingSet.load_df pour entraîner le modèle. Si vous modifiez ce DataFrame de quelque façon que ce soit avant de l’utiliser pour l’entraînement du modèle, les modifications ne sont pas appliquées quand vous utilisez le modèle à des fins d’inférence. Les performances du modèle s’en trouvent altérées.
  • Le type de modèle doit avoir un python_flavor correspondant dans MLflow. MLflow prend en charge la plupart des frameworks d’entraînement de modèle Python, notamment :
    • scikit-learn
    • keras
    • PyTorch
    • SparkML
    • LightGBM
    • XGBoost
    • TensorFlow Keras (en utilisant le python_flavormlflow.keras)
  • Modèles MLflow pyfunc personnalisés

Caractérisation dans le catalogue Unity

# Train model
import mlflow
from sklearn import linear_model

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['total_purchases_30d'],
      lookup_key='customer_id',
    ),
    FeatureLookup(
      table_name='ml.recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

fe = FeatureEngineeringClient()

with mlflow.start_run():

  # df has columns ['customer_id', 'product_id', 'rating']
  training_set = fe.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id', 'product_id']
  )

  training_df = training_set.load_df().toPandas()

  # "training_df" columns ['total_purchases_30d', 'category', 'rating']
  X_train = training_df.drop(['rating'], axis=1)
  y_train = training_df.rating

  model = linear_model.LinearRegression().fit(X_train, y_train)

  fe.log_model(
    model=model,
    artifact_path="recommendation_model",
    flavor=mlflow.sklearn,
    training_set=training_set,
    registered_model_name="recommendation_model"
  )

# Batch inference

# If the model at model_uri is packaged with the features, the FeatureStoreClient.score_batch()
# call automatically retrieves the required features from Feature Store before scoring the model.
# The DataFrame returned by score_batch() augments batch_df with
# columns containing the feature values and a column containing model predictions.

fe = FeatureEngineeringClient()

# batch_df has columns ‘customer_id’ and ‘product_id’
predictions = fe.score_batch(
    model_uri=model_uri,
    df=batch_df
)

# The ‘predictions’ DataFrame has these columns:
# ‘customer_id’, ‘product_id’, ‘total_purchases_30d’, ‘category’, ‘prediction’

Magasin de fonctionnalités de l'espace de travail

# Train model
import mlflow
from sklearn import linear_model

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['total_purchases_30d'],
      lookup_key='customer_id',
    ),
    FeatureLookup(
      table_name='recommender_system.product_features',
      feature_names=['category'],
      lookup_key='product_id'
    )
  ]

fs = FeatureStoreClient()

with mlflow.start_run():

  # df has columns ['customer_id', 'product_id', 'rating']
  training_set = fs.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id', 'product_id']
  )

  training_df = training_set.load_df().toPandas()

  # "training_df" columns ['total_purchases_30d', 'category', 'rating']
  X_train = training_df.drop(['rating'], axis=1)
  y_train = training_df.rating

  model = linear_model.LinearRegression().fit(X_train, y_train)

  fs.log_model(
    model=model,
    artifact_path="recommendation_model",
    flavor=mlflow.sklearn,
    training_set=training_set,
    registered_model_name="recommendation_model"
  )

# Batch inference

# If the model at model_uri is packaged with the features, the FeatureStoreClient.score_batch()
# call automatically retrieves the required features from Feature Store before scoring the model.
# The DataFrame returned by score_batch() augments batch_df with
# columns containing the feature values and a column containing model predictions.

fs = FeatureStoreClient()

# batch_df has columns ‘customer_id’ and ‘product_id’
predictions = fs.score_batch(
    model_uri=model_uri,
    df=batch_df
)

# The ‘predictions’ DataFrame has these columns:
# ‘customer_id’, ‘product_id’, ‘total_purchases_30d’, ‘category’, ‘prediction’

Utiliser des valeurs de caractéristiques personnalisées lors du scoring d’un modèle empaqueté avec les métadonnées de caractéristique

Par défaut, un modèle empaqueté avec les métadonnées de fonctionnalités recherche des fonctionnalités dans les tables de fonctionnalités lors de l’inférence. Pour utiliser des valeurs de fonctionnalités personnalisées pour le scoring, incluez-les dans le DataFrame transmis à FeatureEngineeringClient.score_batch (pour la caractérisation dans le catalogue Unity) ou FeatureStoreClient.score_batch (pour le magasin de fonctionnalités de l’espace de travail).

Par exemple, supposons que vous empaquetiez un modèle avec ces deux caractéristiques :

Caractérisation dans le catalogue Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['account_creation_date', 'num_lifetime_purchases'],
      lookup_key='customer_id',
    ),
  ]

Magasin de fonctionnalités de l'espace de travail

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['account_creation_date', 'num_lifetime_purchases'],
      lookup_key='customer_id',
    ),
  ]

Lors de l’inférence, vous pouvez fournir des valeurs personnalisées pour la caractéristique account_creation_date en appelant score_batch sur un DataFrame qui comprend une colonne nommée account_creation_date. Dans ce cas, l’API recherche uniquement la caractéristique num_lifetime_purchases dans le magasin de caractéristiques et utilise les valeurs de colonne account_creation_date personnalisées fournies pour le scoring du modèle.

Caractérisation dans le catalogue Unity

# batch_df has columns ['customer_id', 'account_creation_date']
predictions = fe.score_batch(
  model_uri='models:/ban_prediction_model/1',
  df=batch_df
)

Magasin de fonctionnalités de l'espace de travail

# batch_df has columns ['customer_id', 'account_creation_date']
predictions = fs.score_batch(
  model_uri='models:/ban_prediction_model/1',
  df=batch_df
)

Entraîner et évaluer un modèle en utilisant une combinaison de caractéristiques du magasin de caractéristiques et de données résidant en dehors du magasin de caractéristiques

Vous pouvez entraîner un modèle en utilisant une combinaison de caractéristiques du magasin de caractéristiques et de données hors du magasin de caractéristiques. Quand vous empaquetez le modèle avec des métadonnées de caractéristique, le modèle récupère les valeurs de caractéristiques du magasin de caractéristique à des fins d’inférence.

Pour entraîner un modèle, incluez les données supplémentaires sous forme de colonnes dans le DataFrame transmis à FeatureEngineeringClient.create_training_set (pour l’ingénierie des fonctionnalités dans le catalogue Unity) ou FeatureStoreClient.create_training_set (pour le Magasin de fonctionnalités d’espace de travail). Cet exemple utilise la caractéristique total_purchases_30d du magasin de caractéristiques et la colonne externe browser.

Caractérisation dans le catalogue Unity

feature_lookups = [
    FeatureLookup(
      table_name='ml.recommender_system.customer_features',
      feature_names=['total_purchases_30d'],
      lookup_key='customer_id',
    ),
  ]

fe = FeatureEngineeringClient()

# df has columns ['customer_id', 'browser', 'rating']
training_set = fe.create_training_set(
  df=df,
  feature_lookups=feature_lookups,
  label='rating',
  exclude_columns=['customer_id']  # 'browser' is not excluded
)

Magasin de fonctionnalités de l'espace de travail

feature_lookups = [
    FeatureLookup(
      table_name='recommender_system.customer_features',
      feature_names=['total_purchases_30d'],
      lookup_key='customer_id',
    ),
  ]

fs = FeatureStoreClient()

# df has columns ['customer_id', 'browser', 'rating']
training_set = fs.create_training_set(
  df=df,
  feature_lookups=feature_lookups,
  label='rating',
  exclude_columns=['customer_id']  # 'browser' is not excluded
)

Lors de l’inférence, le DataFrame utilisé dans FeatureStoreClient.score_batch doit inclure la colonne browser.

Caractérisation dans le catalogue Unity

# At inference, 'browser' must be provided
# batch_df has columns ['customer_id', 'browser']
predictions = fe.score_batch(
  model_uri=model_uri,
  df=batch_df
)

Magasin de fonctionnalités de l'espace de travail

# At inference, 'browser' must be provided
# batch_df has columns ['customer_id', 'browser']
predictions = fs.score_batch(
  model_uri=model_uri,
  df=batch_df
)

Charger des modèles et traiter les inférences par lot à l’aide de MLflow

Après l’enregistrement d’un modèle à l’aide de la méthode log_model de FeatureEngineeringClient (pour l’ingénierie de caractéristiques dans Unity Catalog) ou de FeatureStoreClient (pour Workspace Feature Store), MLflow peut être utilisé lors de l’inférence. MLflow.pyfunc.predict récupère les valeurs des fonctionnalités du Feature Store et joint également toutes les valeurs fournies au moment de l’inférence. Vous devez fournir la ou les clés primaires des caractéristiques utilisées dans le modèle.

Remarque

L’inférence par lot avec MLflow nécessite MLflow version 2.11 et ultérieure.

# Train model
import mlflow
from sklearn import linear_model

feature_lookups = [
  FeatureLookup(
    table_name='ml.recommender_system.customer_features',
    feature_names=['total_purchases_30d'],
    lookup_key='customer_id',
  ),
  FeatureLookup(
    table_name='ml.recommender_system.product_features',
    feature_names=['category'],
    lookup_key='product_id'
  )
]

fe = FeatureEngineeringClient()

with mlflow.start_run():

  # df has columns ['customer_id', 'product_id', 'rating']
  training_set = fe.create_training_set(
    df=df,
    feature_lookups=feature_lookups,
    label='rating',
    exclude_columns=['customer_id', 'product_id']
  )

  training_df = training_set.load_df().toPandas()

  # "training_df" columns ['total_purchases_30d', 'category', 'rating']
  X_train = training_df.drop(['rating'], axis=1)
  y_train = training_df.rating

  model = linear_model.LinearRegression().fit(X_train, y_train)

  fe.log_model(
    model=model,
    artifact_path="recommendation_model",
    flavor=mlflow.sklearn,
    training_set=training_set,
    registered_model_name="recommendation_model",
    #refers to the default value of "result_type" if not provided at inference
    params={"result_type":"double"},
  )

# Batch inference with MLflow

# NOTE: the result_type parameter can only be used if a default value
# is provided in log_model. This is automatically done for all models
# logged using Databricks Runtime for ML 15.0 or above.
# For earlier Databricks Runtime versions, use set_result as shown below.

# batch_df has columns ‘customer_id’ and ‘product_id’
model = mlflow.pyfunc.load_model(model_version_uri)

# If result_type parameter is provided in log_model
predictions = model.predict(df, {"result_type":"double"})

# If result_type parameter is NOT provided in log_model
model._model_impl.set_result_type("double")
predictions = model.predict(df)