Use recurso para modelos de ensino

Este artigo descreve como você pode ensinar modelos usando engenharia de recursos no Catálogo do Unity ou na Loja de recursos workspace local. Você deve primeiro criar um dataset de treinamento, que define os recursos a serem usados e como join los. Então, quando você ensina um modelo, o modelo retém as referências aos recursos.

Ao ensinar um modelo usando recurso engenharia no Unity Catalog, você pode view a linhagem do modelo no Catalog Explorer. As tabelas e funções usadas para criar o modelo são rastreadas e exibidas automaticamente. Veja view repositório de recursos de linhagem.

Ao usar o modelo para inferência, o senhor pode optar por fazer com que ele recupere valores de recurso do repositório de recursos. Os modelos do repositório de recursos também são compatíveis com a interface pyfunc do MLflow, de modo que o senhor pode usar o MLflow para realizar inferência de lotes com tabelas de recursos.

Criar um conjunto de dados de treinamento

Para selecionar um recurso específico de uma tabela de recursos para treinamento de modelo, crie um dataset de treinamento usando a API FeatureEngineeringClient.create_training_set (para engenharia de recursos no catálogo do Unity) ou FeatureStoreClient.create_training_set (para armazenamento de recursos workspace ) e um objeto chamado FeatureLookup. Um FeatureLookup especifica cada recurso a ser usado no conjunto de treinamento, incluindo o nome da tabela de recursos, os nomes dos recursos e as key a serem usadas ao ingressar na tabela de recursos com o DataFrame passado para create_training_set. Consulte o recurso Lookup para mais informações.

Use o parâmetro feature_names ao criar um FeatureLookup. feature_names usa um único nome de recurso, uma lista de nomes de recursos ou None para procurar todos os recursos (excluindo key primária ) 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 da key primária (excluindo key 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 em feature_lookups. Ele preserva todas as colunas do DataFrame fornecidas para create_training_set , exceto aquelas excluídas usando exclude_columns.

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

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

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

Use o argumento lookup_key em FeatureLookup para o nome da coluna no conjunto de treinamento. create_training_set executa uma join ordenada entre as colunas do conjunto de treinamento especificado no argumento lookup_key usando a ordem na qual a key primária foi especificada quando a tabela de recursos foi criada.

Neste exemplo, recommender_system.customer_features tem a seguinte key primária: customer_id, dt.

A tabela de recursos recommender_system.product_features tem key primária product_id.

Se o training_df tiver as seguintes colunas:

• cid

• transaction_dt

• product_id

• rating

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

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

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 dataset de treinamento realizando uma join à esquerda , joinas tabelas recommender_system.customer_features e training_df usando a key (customer_id,dt) correspondente a (cid,transaction_dt), conforme mostrado no código a seguir:

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

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

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

Use o argumento opcional output_name no 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.

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

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

Crie um TrainingSet usando o mesmo recurso várias vezes

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

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

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

Crie um TrainingSet para uma versão não supervisionada do machine learning

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

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

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

modelos de ensino e realizar inferência de lotes com tabelas de recursos

Quando você ensina um modelo usando recursos da Feature Store, o modelo mantém as referências aos recursos. Ao usar o modelo para inferência, você pode optar por fazer com que ele recupere valores de recursos do Feature Store. Você deve fornecer a(s key(s) primária(s) dos recursos usados no modelo. O modelo recupera os recursos necessários do Feature Store em sua workspace. Em seguida, ele join os valores de recurso conforme necessário durante a pontuação.

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

• Você deve logs o modelo usando o método log_model de FeatureEngineeringClient (para recurso engenharia no Unity Catalog) ou FeatureStoreClient (para workspace recurso Store).

• Você deve usar o DataFrame retornado por TrainingSet.load_df para ensinar o modelo. Se você modificar este DataFrame de alguma forma antes de usá-lo para ensinar 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 python_flavor correspondente no MLflow. O MLflow suporta a maioria das estruturas de treinamento de modelo Python, incluindo:

  • Scikit-Learn

  • Keras

  • PyTorchName

  • SparkMLName

  • Light GBM

  • XGBoostGenericName

  • TensorFlow Keras (usando o python_flavor mlflow.keras)

• Modelos pyfunc MLflow personalizados

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

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

Use valores de recursos personalizados ao pontuar um pacote de modelo com metadados de recursos

Por default, um modelo de pacote com metadados de recursos consulta recursos nas tabelas de recursos na inferência. Para usar valores de recurso customizados para pontuação, inclua-os no DataFrame passado para FeatureEngineeringClient.score_batch (para engenharia de recurso no Unity Catalog) ou FeatureStoreClient.score_batch (para recurso Store workspace ).

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

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

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 denominada account_creation_date. Nesse caso, a API procura apenas o recurso num_lifetime_purchases do Feature Store e usa os valores de coluna personalizados account_creation_date fornecidos para a pontuação do modelo.

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

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

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

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

Para ensinar um modelo, inclua os dados extras como colunas no DataFrame passado para FeatureEngineeringClient.create_training_set (para recurso de engenharia no Unity Catalog) ou FeatureStoreClient.create_training_set (para workspace recurso Store). Este exemplo usa o recurso total_purchases_30d do Feature Store e a coluna externa browser.

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
)

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 coluna browser .

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

# 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 realizar inferência de lotes usando o MLflow

Após um modelo ter sido registrado usando o método log_model de FeatureEngineeringClient (para recurso engenharia no Unity Catalog) ou FeatureStoreClient (para workspace repositório de recursos), o MLflow pode ser usado na inferência. MLflow.pyfunc.predict recupera valores de recurso do repositório de recursos e também junta quaisquer valores fornecidos no momento da inferência. O senhor deve fornecer o key(s) principal (is) do recurso usado no modelo.

Observação

A inferência de lotes com o MLflow requer o MLflow versão 2.11 e o acima.

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