Uso de características para entrenar modelos

  • Artículo

En este artículo se describe cómo puede entrenar modelos mediante la ingeniería de características en Unity Catalog o el almacén de características del área de trabajo local. Primero debe crear un conjunto de datos de entrenamiento, que define las características que se van a usar y cómo unirse a ellas. Luego, cuando entrene un modelo, este conservará las referencias a las características.

Al entrenar un modelo mediante la ingeniería de características en Unity Catalog, puede ver el linaje del modelo en el Explorador de catálogos. Se realiza un seguimiento de las tablas y funciones que se usaron para crear el modelo y se las muestra automáticamente. Consulte Visualización del linaje del almacén de características.

Cuando el modelo se usa para la inferencia, puede elegir que recupere los valores de las características del almacén de características. También puede servir el modelo para que busque automáticamente las características publicadas en los almacenes en línea. Los modelos de almacén de características también son compatibles con la interfaz pyfunc de MLflow, por lo que puede usar MLflow para realizar la inferencia por lotes con tablas de características.

Si el modelo usa variables de entorno, obtenga más información sobre cómo usarlos al atender el modelo en línea en Configurar el acceso a los recursos desde los puntos de conexión de servicio del modelo.

Creación de un conjunto de datos de entrenamiento

Para seleccionar características específicas de una tabla de características para el entrenamiento de modelos, cree un conjunto de datos de entrenamiento con la API FeatureEngineeringClient.create_training_set (para la ingeniería de características en Unity Catalog) o FeatureStoreClient.create_training_set (para Workspace Feature Store) y un objeto denominado FeatureLookup. Un objeto FeatureLookup especifica cada una de las características que se va a usar en el conjunto de entrenamiento, incluidos el nombre de la tabla de características, los nombres de las características y las claves que se van a usar al unir la tabla de características al objeto DataFrame pasado a create_training_set. Consulte Búsqueda de características para obtener más información.

Use el parámetro feature_names cuando cree un objeto FeatureLookup. feature_names toma un solo nombre de característica, una lista de nombres de características o el valor Ninguno para buscar todas las características (excepto las claves principales) en la tabla de características en el momento en el que se crea el conjunto de entrenamiento.

Nota:

El tipo y el orden de las columnas lookup_key de ese DataFrame debe coincidir con el tipo y el orden de las claves principales (excepto las claves de marca de tiempo) de la tabla de características de referencia.

En este artículo se incluyen ejemplos de código para ambas versiones de la sintaxis.

En este ejemplo, el objeto DataFrame que trainingSet.load_df devuelve contiene una columna para cada característica de feature_lookups. Conserva todas las columnas de DataFrame proporcionadas a create_training_set, excepto las excluidas con exclude_columns.

Ingeniería de características en Unity Catalog

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

Almacén de características del área de trabajo

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

Creación de TrainingSet cuando las claves de búsqueda no coinciden con las claves principales

Use el argumento lookup_key en FeatureLookup para el nombre de columna del conjunto de entrenamiento. create_training_set realiza una combinación ordenada entre las columnas del conjunto de entrenamiento especificado en el argumento lookup_key, para lo que utiliza el orden en el que se especificaron las claves principales cuando se creó la tabla de características.

En este ejemplo, recommender_system.customer_features tiene las claves principales siguientes: customer_id, dt.

La tabla de características recommender_system.product_features tiene la clave principal product_id.

Si training_df tiene las columnas siguientes:

  • cid
  • transaction_dt
  • product_id
  • rating

el código siguiente creará las búsquedas de características correctas para TrainingSet:

Ingeniería de características en Unity Catalog

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

Workspace Feature Store

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

Cuando se llama a create_training_set, este crea un conjunto de datos de entrenamiento mediante la combinación de la izquierda, al unir las tablas recommender_system.customer_features y training_df con las claves (customer_id,dt) correspondientes a (cid,transaction_dt), como se muestra en el código siguiente:

Ingeniería de características en Unity Catalog

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

Workspace Feature Store

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

Creación de un objeto TrainingSet que contiene dos características con el mismo nombre de tablas de características distintas

Use el argumento opcional output_name en FeatureLookup. El nombre proporcionado se usa en lugar del nombre de la característica en el objeto DataFrame que TrainingSet.load_df devuelve. Por ejemplo, con el código siguiente, el objeto DataFrame que training_set.load_df devuelve incluye las columnas customer_height y product_height.

Ingeniería de características en Unity Catalog

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

Almacén de características del área de trabajo

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

Creación de un TrainingSet usando la misma característica varias veces

Para crear un TrainingSet con la misma característica unida por claves de búsqueda diferentes, use varios FeatureLookups. Use un valor único output_name para cada salida de FeatureLookup.

Ingeniería de características en Unity Catalog

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

Workspace Feature Store

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

Creación de TrainingSet para modelos de Machine Learning no supervisados

Establezca label=None al crear un objeto TrainingSet para modelos de aprendizaje no supervisados. Por ejemplo, el objeto TrainingSet siguiente se puede usar para agrupar distintos clientes en grupos en función de sus intereses:

Ingeniería de características en Unity Catalog

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

Almacén de características del área de trabajo

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

Entrenamiento de modelos y realización de inferencias por lotes con tablas de características

Cuando se entrena un modelo con características del almacén correspondiente, el modelo conserva las referencias a las características. Cuando el modelo se usa para la inferencia, puede elegir que recupere los valores de las características del almacén de características. Debe proporcionar las claves principales de las características que se usan en el modelo. El modelo recupera las características que necesita del almacén de características de su área de trabajo. A continuación, combina los valores de las características según sea necesario durante la puntuación.

Para admitir la búsqueda de características en tiempo de inferencia:

  • Debe registrar el modelo mediante el método log_model de FeatureEngineeringClient (para la ingeniería de características en Unity Catalog) o FeatureStoreClient (para Workspace Feature Store).
  • Debe usar el objeto DataFrame que TrainingSet.load_df devuelve para entrenar el modelo. Si modifica este objeto DataFrame de cualquier forma antes de usarlo para entrenar el modelo, las modificaciones no se aplican cuando el modelo se use para inferencia. Esto disminuye el rendimiento del modelo.
  • El tipo de modelo debe tener un valor python_flavor correspondiente en MLflow. MLflow admite la mayoría de los marcos de entrenamiento de modelos de Python, entre los que se incluyen:
    • scikit-learn
    • keras
    • PyTorch
    • SparkML
    • LightGBM
    • XGBoost
    • TensorFlow Keras (con python_flavormlflow.keras)
  • Modelos pyfunc de MLflow personalizados

Ingeniería de características en Unity Catalog

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

Almacén de características del área de trabajo

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

Uso de valores de características personalizadas al puntuar un modelo empaquetado con metadatos de características

De forma predeterminada, un modelo empaquetado con metadatos de características busca características de las tablas de características en la inferencia. Para usar valores de características personalizadas para la puntuación, inclúyalos en el objeto DataFrame que se pasa a FeatureEngineeringClient.score_batch (para la ingeniería de características en Unity Catalog) o FeatureStoreClient.score_batch (para Workspace Feature Store).

Por ejemplo, supongamos que empaqueta un modelo con estas dos características:

Ingeniería de características en Unity Catalog

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

Workspace Feature Store

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

En la inferencia, puede proporcionar valores personalizados para la característica account_creation_date si llama a score_batch en un objeto DataFrame que incluya una columna denominada account_creation_date. En este caso, la API solo busca la característica num_lifetime_purchases del almacén de características y usa los valores de columna account_creation_date personalizados que se han proporcionado para la puntuación del modelo.

Ingeniería de características en Unity Catalog

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

Almacén de características del área de trabajo

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

Entrenamiento y puntuación de un modelo con una combinación de características y datos del almacén de características que residen fuera de este

Puede entrenar un modelo con una combinación de características y datos del almacén de características externos a este. Cuando se empaqueta el modelo con metadatos de características, este recupera los valores de las características del almacén de características para la inferencia.

Para entrenar un modelo, incluya los datos adicionales como columnas en el objeto DataFrame que se pasa a FeatureEngineeringClient.create_training_set (para la ingeniería de características en Unity Catalog) o FeatureStoreClient.create_training_set (para Workspace Feature Store). En este ejemplo se usa la característica total_purchases_30d del almacén de características y la columna externa browser.

Ingeniería de características en Unity Catalog

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
)

Almacén de características del área de trabajo

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
)

En la inferencia, el objeto DataFrame que se usa en FeatureStoreClient.score_batch debe incluir la columna browser.

Ingeniería de características en Unity Catalog

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

Workspace Feature Store

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

Carga de modelos y realización de la inferencia por lotes mediante MLflow

Una vez que se haya registrado un modelo mediante el log_model método de (para la ingeniería de FeatureEngineeringClient características en el catálogo de Unity) o FeatureStoreClient (para el almacén de características del área de trabajo), MLflow se puede usar en la inferencia. MLflow.pyfunc.predict recupera los valores de características de Feature Store y también combina los valores proporcionados en el momento de la inferencia. Debe proporcionar las claves principales de las características que se usan en el modelo.

Nota:

La inferencia por lotes con MLflow requiere MLflow versión 2.11 y posteriores.

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