Partilhar via

Usar recursos para treinar modelos

  • Artigo

Este artigo descreve como você pode treinar modelos usando a Engenharia de Recursos no Unity Catalog ou no Workspace Feature Store local. Você deve primeiro criar um conjunto de dados de treinamento, que define os recursos a serem usados e como ingressar neles. Em seguida, quando você treina um modelo, o modelo retém referências aos recursos.

Ao treinar um modelo usando a Engenharia de Recursos no Catálogo Unity, você pode exibir a linhagem do modelo no Catalog Explorer. As tabelas e funções que foram usadas para criar o modelo são automaticamente rastreadas e exibidas. Consulte Ver linhagem da loja de funcionalidades.

Ao usar o modelo para inferência, você pode optar por recuperá-lo valores de feição do repositório de recursos. Você também pode servir o modelo com o Model Serving e ele pesquisará automaticamente os recursos publicados em lojas online. Os modelos de armazenamento de recursos também são compatíveis com a interface pyfunc do MLflow, portanto, você pode usar o MLflow para executar inferência em lote com tabelas de recursos.

Se o seu modelo usa variáveis de ambiente, saiba mais sobre como usá-las ao servir o modelo online em Configurar o acesso a recursos de pontos de extremidade de serviço de modelo.

Um modelo pode usar no máximo 50 mesas e 100 funções para treinamento.

Criar um conjunto de dados de preparação

Para selecionar recursos específicos de uma tabela de recursos para treinamento de modelo, crie um conjunto de dados de treinamento usando a FeatureEngineeringClient.create_training_set API (para Engenharia de Recursos no Catálogo Unity) ou FeatureStoreClient.create_training_set (para Repositório de Recursos de Espaço de Trabalho) e um objeto chamado FeatureLookup. A FeatureLookup especifica cada recurso a ser usado no conjunto de treinamento, incluindo o nome da tabela de recursos, o(s) nome(s) dos recursos e a(s) chave(s) a ser usada ao ingressar na tabela de recursos com o DataFrame passado para create_training_set. Consulte Pesquisa de recursos para obter mais informações.

Use o feature_names parâmetro ao criar um FeatureLookuparquivo . feature_names usa um único nome de recurso, uma lista de nomes de recursos ou Nenhum para procurar todos os recursos (excluindo chaves primárias) na tabela de recursos no momento em que o conjunto de treinamento é criado.

Nota

O tipo e a ordem das lookup_key colunas no DataFrame devem corresponder ao tipo e à ordem das chaves primárias (excluindo as chaves de carimbo de data/hora) da tabela de recursos de referência.

Este artigo inclui exemplos de código para ambas as versões da sintaxe.

Neste exemplo, o DataFrame retornado por trainingSet.load_df contém uma coluna para cada recurso no feature_lookups. Ele preserva todas as colunas do DataFrame fornecidas, create_training_set exceto aquelas excluídas usando exclude_columns.

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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

Criar um TrainingSet quando as teclas de pesquisa não corresponderem às chaves primárias

Use o argumento lookup_key no para o FeatureLookup nome da coluna no conjunto de treinamento. create_training_set Executa uma junção ordenada entre as colunas do conjunto de treinamento especificado no lookup_key argumento usando a ordem na qual as chaves primárias foram especificadas quando a tabela de recursos foi criada.

Neste exemplo, recommender_system.customer_features tem as seguintes chaves primárias: customer_id, dt.

A recommender_system.product_features tabela de recursos tem chave product_idprimária .

Se o training_df tem as seguintes colunas:

  • cid
  • transaction_dt
  • product_id
  • rating

O código a seguir criará as pesquisas de recursos corretas para o TrainingSet:

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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 é chamado, ele cria um conjunto de dados de treinamento executando uma junção à esquerda, unindo as tabelas recommender_system.customer_features e training_df usando as teclas (customer_id,dt) correspondentes a (cid,transaction_dt), conforme mostrado no código a seguir:

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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

Criar um TrainingSet contendo dois recursos com o mesmo nome de tabelas de recursos diferentes

Use o argumento opcional no FeatureLookup.output_name O nome fornecido é usado no lugar do nome do recurso no DataFrame retornado pelo TrainingSet.load_df. Por exemplo, com o código a seguir, o DataFrame retornado por training_set.load_df inclui colunas customer_height e product_height.

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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

Criar um TrainingSet usando o mesmo recurso várias vezes

Para criar um TrainingSet usando o mesmo recurso unido por diferentes chaves de pesquisa, use vários FeatureLookups. Use um exclusivo output_name para cada saída FeatureLookup.

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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

Criar um TrainingSet para modelos de aprendizado de máquina não supervisionados

Defina label=None ao criar um TrainingSet para modelos de aprendizagem não supervisionados. Por exemplo, o seguinte TrainingSet pode ser usado para agrupar diferentes clientes em grupos com base em seus interesses:

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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

Treine modelos e execute inferência em lote com tabelas de recursos

Quando você treina um modelo usando recursos da Feature Store, o modelo mantém referências aos recursos. Ao usar o modelo para inferência, você pode optar por recuperá-lo valores de feição do Repositório de Recursos. Você deve fornecer a(s) chave(s) primária(s) dos recursos usados no modelo. O modelo recupera os recursos necessários do Feature Store em seu espaço de trabalho. Em seguida, ele une os valores de recurso conforme necessário durante a pontuação.

Para dar suporte à pesquisa de recursos no momento da inferência:

  • Você deve registrar o modelo usando o log_model método de (para Engenharia de Recursos no Catálogo Unity) ou FeatureStoreClient (para o Repositório de Recursos do Espaço de FeatureEngineeringClient Trabalho).
  • Você deve usar o DataFrame retornado por TrainingSet.load_df para treinar o modelo. Se você modificar esse DataFrame de qualquer forma antes de usá-lo para treinar o modelo, as modificações não serão aplicadas quando você usar o modelo para inferência. Isso diminui o desempenho do modelo.
  • O tipo de modelo deve ter um correspondente python_flavor em MLflow. O MLflow suporta a maioria das estruturas de treinamento de modelo Python, incluindo:
    • scikit-learn
    • keras
    • PyTorch
    • Faísca
    • LightGBM
    • XGBoost
    • Keras TensorFlow (usando o python_flavormlflow.keras)
  • Modelos pyfunc MLflow personalizados

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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

Usar valores de recurso personalizados ao pontuar um modelo empacotado com metadados de recurso

Por padrão, um modelo empacotado com metadados de recursos procura recursos de tabelas de recursos na inferência. Para usar valores de recurso personalizados para pontuação, inclua-os no DataFrame passado para FeatureEngineeringClient.score_batch (para Engenharia de Recursos no Catálogo Unity) ou FeatureStoreClient.score_batch (para Repositório de Recursos de Espaço de Trabalho).

Por exemplo, suponha que você empacote um modelo com estes dois recursos:

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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

Na inferência, você pode fornecer valores personalizados para o recurso account_creation_date chamando score_batch um DataFrame que inclui uma coluna chamada account_creation_date. Nesse caso, a API procura apenas o num_lifetime_purchases recurso do Feature Store e usa os valores de coluna personalizados account_creation_date fornecidos para a pontuação do modelo.

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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

Treinar e pontuar um modelo usando uma combinação de recursos do Feature Store e dados que residem fora do Feature Store

Você pode treinar um modelo usando uma combinação de recursos do Feature Store e dados de fora do Feature Store. Quando você empacota o modelo com metadados de recurso, o modelo recupera valores de feição do Feature Store para inferência.

Para treinar um modelo, inclua os dados extras como colunas no DataFrame passadas para FeatureEngineeringClient.create_training_set (para Engenharia de Recursos no Catálogo Unity) ou FeatureStoreClient.create_training_set (para Repositório de Recursos de Espaço de Trabalho). Este exemplo usa o recurso total_purchases_30d do Feature Store e a coluna browserexterna .

Engenharia de recursos no catálogo da unidade

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
)

Repositório de recursos do espaço de trabalho

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
)

Na inferência, o DataFrame usado em FeatureStoreClient.score_batch deve incluir a browser coluna.

Engenharia de recursos no catálogo da unidade

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

Repositório de recursos do espaço de trabalho

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

Carregue modelos e execute inferência em lote usando MLflow

Depois que um modelo tiver sido registrado usando o log_model método de (para Engenharia de Recursos no Catálogo Unity) ou FeatureStoreClient (para o Repositório de Recursos de Espaço de FeatureEngineeringClient Trabalho), o MLflow poderá ser usado na inferência. MLflow.pyfunc.predict recupera valores de recursos do Feature Store e também se junta a quaisquer valores fornecidos no momento da inferência. Você deve fornecer a(s) chave(s) primária(s) dos recursos usados no modelo.

Nota

A inferência em lote com MLflow requer MLflow versão 2.11 e superior.

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