Share via

Usare le funzionalità per eseguire il training dei modelli

  • Articolo

Questo articolo descrive come eseguire il training dei modelli usando Progettazione funzionalità in Unity Catalog o l'archivio delle funzionalità dell'area di lavoro locale. È prima necessario creare un set di dati di training, che definisce le funzionalità da usare e come aggiungerle. Quindi, quando si esegue il training di un modello, il modello mantiene i riferimenti alle funzionalità.

Quando si esegue il training di un modello usando Progettazione funzionalità in Unity Catalog, è possibile visualizzare la derivazione del modello in Esplora cataloghi. Le tabelle e le funzioni usate per creare il modello vengono registrate e visualizzate automaticamente. Vedere Visualizzare la derivazione dell'archivio funzionalità.

Quando si usa il modello per l'inferenza, è possibile scegliere di recuperarli dall'archivio funzionalità. È anche possibile gestire il modello con Model Serving e cercherà automaticamente le funzionalità pubblicate nei negozi online. I modelli di archivio funzionalità sono compatibili anche con l'interfaccia pyfunc MLflow, quindi è possibile usare MLflow per eseguire l'inferenza batch con le tabelle delle funzionalità.

Se il modello usa variabili di ambiente, vedere Configurare l'accesso alle risorse dai modelli che servono gli endpoint.

Un modello può usare al massimo 50 tabelle e 100 funzioni per il training.

Creare un set di dati di training

Per selezionare funzionalità specifiche da una tabella delle funzionalità per il training del modello, creare un set di dati di training usando l'API FeatureEngineeringClient.create_training_set (per Progettazione funzionalità in Unity Catalog) o FeatureStoreClient.create_training_set (per l'archivio funzionalità dell'area di lavoro) e un oggetto denominato FeatureLookup. Specifica FeatureLookup ogni funzionalità da usare nel set di training, incluso il nome della tabella delle funzionalità, i nomi delle funzionalità e le chiavi da usare per unire la tabella delle funzionalità con il dataframe passato a create_training_set. Per altre informazioni, vedere Ricerca funzionalità.

Usare il feature_names parametro quando si crea un oggetto FeatureLookup. feature_names accetta un singolo nome di funzionalità, un elenco di nomi di funzionalità o Nessuno per cercare tutte le funzionalità (escluse le chiavi primarie) nella tabella delle funzionalità al momento della creazione del set di training.

Nota

Il tipo e l'ordine delle lookup_key colonne nel dataframe devono corrispondere al tipo e all'ordine delle chiavi primarie (escluse le chiavi timestamp) della tabella delle funzionalità di riferimento.

Questo articolo include esempi di codice per entrambe le versioni della sintassi.

In questo esempio, il dataframe restituito da trainingSet.load_df contiene una colonna per ogni funzionalità in feature_lookups. Mantiene tutte le colonne del dataframe fornito a create_training_set , ad eccezione di quelle escluse usando exclude_columns.

Progettazione delle funzionalità nel catalogo 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()

Archivio delle funzionalità dell'area di lavoro

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

Creare un oggetto TrainingSet quando le chiavi di ricerca non corrispondono alle chiavi primarie

Usare l'argomento lookup_key in FeatureLookup per il nome della colonna nel set di training. create_training_set esegue un join ordinato tra le colonne del set di training specificato nell'argomento utilizzando l'ordine lookup_key in cui sono state specificate le chiavi primarie al momento della creazione della tabella delle funzionalità.

In questo esempio, ha le chiavi primarie recommender_system.customer_features seguenti: customer_id, dt.

La recommender_system.product_features tabella delle funzionalità ha la chiave product_idprimaria .

Se contiene training_df le colonne seguenti:

  • cid
  • transaction_dt
  • product_id
  • rating

Il codice seguente creerà le ricerche di funzionalità corrette per :TrainingSet

Progettazione delle funzionalità nel catalogo 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'
    )
  ]

Archivio delle funzionalità dell'area di lavoro

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'
    )
  ]

Quando create_training_set viene chiamato, crea un set di dati di training eseguendo un left join, unendo le tabelle recommender_system.customer_features e training_df usando le chiavi (customer_id,dt) corrispondenti a (cid,transaction_dt), come illustrato nel codice seguente:

Progettazione delle funzionalità nel catalogo 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"
)

Archivio delle funzionalità dell'area di lavoro

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

Creare un set di training contenente due funzionalità con lo stesso nome da tabelle di funzionalità diverse

Usare l'argomento output_name facoltativo in FeatureLookup. Il nome specificato viene usato al posto del nome della funzionalità nel dataframe restituito da TrainingSet.load_df. Ad esempio, con il codice seguente, il dataframe restituito da training_set.load_df include colonne customer_height e product_height.

Progettazione delle funzionalità nel catalogo 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()

Archivio delle funzionalità dell'area di lavoro

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

Creare un set di training usando la stessa funzionalità più volte

Per creare un oggetto TrainingSet usando la stessa funzionalità unita da chiavi di ricerca diverse, usare più featureLookups. Usare un valore univoco output_name per ogni output featureLookup.

Progettazione delle funzionalità nel catalogo 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'
    )
  ]

Archivio delle funzionalità dell'area di lavoro

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'
    )
  ]

Creare un set di training per i modelli di Machine Learning non supervisionati

Impostare label=None quando si crea un oggetto TrainingSet per i modelli di apprendimento non supervisionati. Ad esempio, il trainingSet seguente può essere usato per raggruppare clienti diversi in gruppi in base ai propri interessi:

Progettazione delle funzionalità nel catalogo 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()

Archivio delle funzionalità dell'area di lavoro

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

Eseguire il training di modelli ed eseguire l'inferenza batch con le tabelle delle funzionalità

Quando si esegue il training di un modello usando le funzionalità di Feature Store, il modello mantiene i riferimenti alle funzionalità. Quando si usa il modello per l'inferenza, è possibile scegliere di recuperarli da Feature Store. È necessario specificare le chiavi primarie delle funzionalità usate nel modello. Il modello recupera le funzionalità necessarie da Feature Store nell'area di lavoro. Unisce quindi i valori delle funzionalità in base alle esigenze durante l'assegnazione dei punteggi.

Per supportare la ricerca delle funzionalità in fase di inferenza:

  • È necessario registrare il modello usando il log_model metodo ( FeatureEngineeringClient per Progettazione funzionalità nel catalogo Unity) o FeatureStoreClient (per l'archivio funzionalità dell'area di lavoro).
  • È necessario usare il dataframe restituito da TrainingSet.load_df per eseguire il training del modello. Se si modifica questo dataframe in qualsiasi modo prima di usarlo per eseguire il training del modello, le modifiche non vengono applicate quando si usa il modello per l'inferenza. Ciò riduce le prestazioni del modello.
  • Il tipo di modello deve avere un oggetto corrispondente python_flavor in MLflow. MLflow supporta la maggior parte dei framework di training dei modelli Python, tra cui:
    • scikit-learn
    • keras
    • PyTorch
    • SparkML
    • LightGBM
    • XGBoost
    • TensorFlow Keras (usando )python_flavormlflow.keras
  • Modelli pyfunc MLflow personalizzati

Progettazione delle funzionalità nel catalogo 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’

Archivio delle funzionalità dell'area di lavoro

# 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’

Usare valori di funzionalità personalizzati per l'assegnazione di punteggi a un modello in pacchetto con metadati delle funzionalità

Per impostazione predefinita, un modello in pacchetto con metadati di funzionalità cerca le funzionalità dalle tabelle delle funzionalità in corrispondenza dell'inferenza. Per usare valori di funzionalità personalizzati per l'assegnazione dei punteggi, includerli nel dataframe passato a FeatureEngineeringClient.score_batch (per progettazione delle funzionalità nel catalogo Unity) o FeatureStoreClient.score_batch (per l'archivio delle funzionalità dell'area di lavoro).

Si supponga, ad esempio, di creare un pacchetto di un modello con queste due funzionalità:

Progettazione delle funzionalità nel catalogo unity

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

Archivio delle funzionalità dell'area di lavoro

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

All'inferenza, è possibile fornire valori personalizzati per la funzionalità account_creation_date chiamando score_batch in un dataframe che include una colonna denominata account_creation_date. In questo caso l'API cerca solo la num_lifetime_purchases funzionalità di Feature Store e usa i valori di colonna personalizzati account_creation_date forniti per l'assegnazione dei punteggi del modello.

Progettazione delle funzionalità nel catalogo unity

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

Archivio delle funzionalità dell'area di lavoro

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

Eseguire il training e assegnare un punteggio a un modello usando una combinazione di funzionalità e dati che risiedono all'esterno di Feature Store

È possibile eseguire il training di un modello usando una combinazione di funzionalità e dati dell'archivio funzionalità dall'esterno di Feature Store. Quando si crea un pacchetto del modello con metadati di funzionalità, il modello recupera i valori delle funzionalità dall'Archivio funzionalità per l'inferenza.

Per eseguire il training di un modello, includere i dati aggiuntivi come colonne nel dataframe passato a FeatureEngineeringClient.create_training_set (per Progettazione delle funzionalità in Unity Catalog) o FeatureStoreClient.create_training_set (per l'archivio funzionalità dell'area di lavoro). Questo esempio usa la funzionalità total_purchases_30d di Feature Store e la colonna browseresterna .

Progettazione delle funzionalità nel catalogo 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
)

Archivio delle funzionalità dell'area di lavoro

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
)

In caso di inferenza, il dataframe usato in FeatureStoreClient.score_batch deve includere la browser colonna .

Progettazione delle funzionalità nel catalogo 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
)

Archivio delle funzionalità dell'area di lavoro

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

Caricare modelli ed eseguire l'inferenza batch con MLflow

Dopo che un modello è stato registrato usando il log_model metodo di (per Progettazione delle funzionalità nel catalogo Unity) o FeatureStoreClient (per l'archivio delle funzionalità dell'area FeatureEngineeringClient di lavoro), MLflow può essere usato all'inferenza. MLflow.pyfunc.predict recupera i valori delle funzionalità dall'Archivio funzionalità e aggiunge anche tutti i valori forniti in fase di inferenza. È necessario specificare le chiavi primarie delle funzionalità usate nel modello.

Nota

L'inferenza batch con MLflow richiede MLflow versione 2.11 e successive.

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