Compartilhar via

Usar recursos para treinar modelos

  • Artigo

Este artigo descreve como você pode treinar modelos usando a Engenharia de Recursos no Catálogo do Unity ou no Repositório de recursos do Espaço de trabalho local. Primeiro, você precisa criar um conjunto de dados de treinamento, que define os recursos a serem usados e como uni-los. Quando você treina um modelo, ele retém as referências aos recursos.

Ao treinar um modelo usando a Definição de Recursos no Catálogo do Unity, você pode exibir a linhagem do modelo no Gerenciador de Catálogos. Tabelas e funções que foram usadas para criar o modelo são automaticamente acompanhadas e exibidas. Confira Exibir linhagem do repositório de recursos.

Ao usar o modelo para inferência, você pode optar por fazer com que ele recupere valores de recursos do repositório de recursos. Você também pode servir o modelo com o Fornecimento de Modelo e ele pesquisará automaticamente os recursos publicados em lojas online. Os modelos de repositório de recursos também são compatíveis com a interface pyfunc MLflow, para que você possa usar o MLflow para executar a inferência em lote com tabelas de recursos.

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

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

Criar um conjunto de dados de treinamento

Para selecionar recursos específicos de uma tabela de recursos para treinamento de modelo, crie um conjunto de dados de treinamento usando o FeatureEngineeringClient.create_training_set (para Engenharia de Recursos no Catálogo do Unity) ou FeatureStoreClient.create_training_set (para o Workspace Feature Store) API e um objeto chamado FeatureLookup. Uma FeatureLookup especifica cada recurso a ser usado no conjunto de treinamento, incluindo o nome da tabela de recursos, os nomes dos recursos e as chaves a serem usadas ao unir a tabela de recursos com o DataFrame transmitido para create_training_set. Consulte Pesquisa de Recursosra obter mais informações.

Use o parâmetro feature_names ao criar um FeatureLookup. feature_names usa um só nome de recurso, uma lista de nomes de recursos ou a opção Nenhum para pesquisar todos os recursos (excluindo chaves primárias) na tabela de recursos no momento em que o conjunto de treinamento é criado.

Observação

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

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

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

Engenharia de Recursos no Catálogo do 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()

Repositório de Recursos do Workspace

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 chaves de pesquisa não corresponderem às chaves primárias

Use o argumento lookup_key na FeatureLookup para o nome da coluna no conjunto de treinamento. create_training_set executa uma junção ordenada entre as colunas do conjunto de treinamento especificado no argumento lookup_key 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 tabela de recursos recommender_system.product_features tem a chave primária product_id.

Se o training_df tiver as seguintes colunas:

  • cid
  • transaction_dt
  • product_id
  • rating

o seguinte código criará as pesquisas de recurso corretas para o TrainingSet:

Engenharia de Recursos no Catálogo do 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'
    )
  ]

Repositório de Recursos do Workspace

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 por meio das chaves (customer_id, dt) correspondentes a (cid, transaction_dt), conforme mostrado no seguinte código:

Engenharia de Recursos no Catálogo do 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"
)

Repositório de Recursos do Workspace

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 que contém dois recursos com o mesmo nome de diferentes tabelas de recursos

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

Engenharia de Recursos no Catálogo do 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()

Repositório de Recursos do Workspace

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 ingressado por chaves de pesquisa diferentes, use vários FeatureLookups. Use um output_name exclusivo para cada saída FeatureLookup.

Engenharia de Recursos no Catálogo do 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'
    )
  ]

Repositório de Recursos do Workspace

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 machine learning não supervisionados

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

Engenharia de Recursos no Catálogo do 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()

Repositório de Recursos do Workspace

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

Treinar modelos e realizar inferência em lote com tabelas de recursos

Quando você treina um modelo usando recursos do Repositório de Recursos, o modelo mantém as referências aos recursos. Ao usar o modelo para inferência, você pode optar por fazê-lo recuperar valores de recursos do Repositório de Recursos. Você precisa fornecer as chaves primárias dos recursos usados no modelo. O modelo recupera os recursos necessários do Repositório de Recursos no seu workspace. Em seguida, ele une os valores de recursos 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 FeatureEngineeringClient (para Engenharia de Recursos no Catálogo do Unity) ou FeatureStoreClient (para o Workspace Feature Store).
  • Você precisa 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 precisa ter um python_flavor correspondente no MLflow. O MLflow dá suporte à maioria das estruturas de treinamento de modelo do Python, incluindo:
    • scikit-learn
    • keras
    • PyTorch
    • SparkML
    • LightGBM
    • XGBoost
    • TensorFlow Keras (usando o python_flavormlflow.keras)
  • Modelos personalizados do pyfunc do MLflow

Engenharia de Recursos no Catálogo do 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’

Repositório de Recursos do Workspace

# 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 pesquisa recursos de tabelas de recursos em inferência. Para usar valores de recursos personalizados para pontuação, inclua-os no DataFrame passado para FeatureEngineeringClient.score_batch (para Engenharia de Recursos no Catálogo do Unity) ou FeatureStoreClient.score_batch (para o Repositório de Recursos do Workspace).

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

Engenharia de Recursos no Catálogo do Unity

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 Workspace

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 em um DataFrame que inclui uma coluna chamada account_creation_date. Nesse caso, a API pesquisa apenas o recurso num_lifetime_purchases do Repositório de Recursos e usa os valores da coluna account_creation_date personalizada fornecidos para a pontuação do modelo.

Engenharia de Recursos no Catálogo do Unity

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

# 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 Repositório de Recursos e dados localizados fora do Repositório de Recursos

Você pode treinar um modelo usando uma combinação de recursos do Repositório de Recursos e dados externos ao Repositório de Recursos. Quando você empacota o modelo com metadados de recurso, o modelo recupera os valores de recursos do Repositório de Recursos 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 do Unity) ou FeatureStoreClient.create_training_set (para o Repositório de Recursos do Workspace). Este exemplo usa o recurso total_purchases_30d do Repositório de Recursos e a coluna browser externa.

Engenharia de Recursos no Catálogo do 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
)

Repositório de Recursos do Workspace

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 precisa incluir a coluna browser.

Engenharia de Recursos no Catálogo do 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
)

Repositório de Recursos do Workspace

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

Carregar modelos e executar inferência em lote usando o MLflow

Depois que um modelo tiver sido registrado usando o método log_model de FeatureEngineeringClient (para Engenharia de Recursos no Catálogo do Unity) ou FeatureStoreClient (para o Workspace Feature Store), o MLflow poderá ser usado em inferência. MLflow.pyfunc.predict recupera valores de recursos do Repositório de Recursos e também une todos os valores fornecidos em tempo de inferência. Você precisa fornecer as chaves primárias dos recursos usados no modelo.

Observação

A inferência em lote com o MLflow requer o MLflow versão 2.11 e superior. Modelos que usam tabelas de recursos de série temporal não têm suporte. Para fazer inferência em lote com tabelas de recursos de série temporal, use score_batch. Confira Treinar modelos e realizar inferência em lote com tabelas de recursos.

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