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.
Quando o senhor ensina um modelo usando o recurso engenharia em Unity Catalog, pode view 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. Ver recurso governança e 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.
Um modelo pode usar no máximo 50 tabelas e 100 funções para treinamento.
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()
Criar um TrainingSet ao usar o site view como uma tabela de recurso
Para usar o view como uma tabela de recurso, o senhor deve usar a versão 0.7.0 do databricks-feature-engineering
ou o acima, que está integrado ao Databricks Runtime 16.0 ML.
O view deve ser um simples SELECT view da tabela Delta de origem. Um SELECT view simples é definido como um view criado a partir de uma única tabela Delta em Unity Catalog que pode ser usada como uma tabela de recurso e cuja chave primária é selecionada sem cláusulas join, GROUP BY ou DISTINCT. As palavras-chave aceitáveis na instrução SQL são SELECT, FROM, WHERE, ORDER BY, LIMIT e OFFSET.
No exemplo a seguir, ml.recommender_system.customer_table
tem a chave primária cid
e dt
, em que dt
é uma coluna de série temporal. O exemplo pressupõe que o dataframe training_df
tenha as colunas cid
, dt
e label
:
from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup
customer_features_df = spark.sql("CREATE OR REPLACE VIEW ml.recommender_system.customer_features AS SELECT cid, dt, pid, rating FROM ml.recommender_system.customer_table WHERE rating > 3")
feature_lookups = [
FeatureLookup(
table_name='ml.recommender_system.customer_features',
feature_names=['pid', 'rating'],
lookup_key=['cid'],
timestamp_lookup_key='dt'
),
]
fe = FeatureEngineeringClient()
training_set = fe.create_training_set(
df=training_df,
feature_lookups=feature_lookups,
label='label'
)
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
deFeatureEngineeringClient
(para recurso engenharia no Unity Catalog) ouFeatureStoreClient
(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 MLflow requer MLflow versão 2.11 e acima. Não há suporte para modelos que usam tabelas de recurso de série temporal. Para fazer a inferência de lotes com tabelas de recursos de séries temporais, use score_batch
. Veja os modelos de ensino e realize a inferência de lotes com tabelas de recurso.
# 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)