Treinar modelos com Feature Views

Feature Views permitem treinar modelos com cálculo de atributos correto em um determinado ponto no tempo e busca automática de atributos durante a inferência. Para obter informações sobre como definir exibições de recursos, consulte Exibições de Recursos.

Requisitos

• Os atributos devem ser criados como visões de atributos. Consulte Visualizações de recursos.
• Para requisitos de CustomUDF e FeatureViewSource, consulte a referência da API Feature Views.

Métodos de API

create_training_set()

Depois de criar Exibições de Recursos, a próxima etapa é criar dados de treinamento para seu modelo. Para fazer isso, passe um conjunto de dados rotulado para create_training_set, o que garante automaticamente a computação precisa pontual de cada valor de característica.

Por exemplo:

FeatureEngineeringClient.create_training_set(
    df: DataFrame,                                # DataFrame with training data
    features: Optional[List[Feature]],            # List of Feature objects
    label: Union[str, List[str], None],           # Label column name(s)
    exclude_columns: Optional[List[str]] = None,  # Optional: columns to exclude
) -> TrainingSet

Chame TrainingSet.load_df para integrar os dados originais de treinamento com recursos calculados dinamicamente em um momento no tempo.

O df argumento deve atender aos seguintes requisitos:

• Deve conter todas as colunas de entidade referenciadas por definições de funcionalidades.
• Deve conter a coluna série temporal referenciada por definições de características.
• Deve conter todas as colunas declaradas em qualquer RequestSource esquema. Os tipos são validados em relação ao esquema declarado. Incompatibilidades geram um erro (sem conversão implícita).
• Deve conter coluna(s) de etiqueta.
• O conjunto de nomes de coluna de entidade, nomes de coluna de série temporal e nomes de coluna de recursos de solicitação deve ser globalmente exclusivo em todas as origens.

Correção de ponto no tempo: para agregação e recursos ColumnSelection apoiados por uma origem de tabela, os recursos são calculados usando apenas os dados de origem disponíveis antes do carimbo de data/hora de cada linha, para evitar vazamento de dados futuros no treinamento do modelo. Para RequestSource atributos, o valor é obtido diretamente da linha do DataFrame rotulada.

log_model()

Use o MLflow para registrar um modelo com metadados de recurso para acompanhamento de linhagem e pesquisa automática de recursos durante a inferência:

FeatureEngineeringClient.log_model(
    model,                                    # Trained model object
    artifact_path: str,                       # Path to store model artifact
    flavor: ModuleType,                       # MLflow flavor module (e.g., mlflow.sklearn)
    training_set: TrainingSet,                # TrainingSet used for training
    registered_model_name: Optional[str],     # Optional: register model in Unity Catalog
    extra_pip_requirements: Optional[List[str]] = None, # Optional: Additional serving dependencies
)

O parâmetro flavor especifica o módulo variante do modelo MLflow a ser usado, como mlflow.sklearn ou mlflow.xgboost.

Modelos registrados com um TrainingSet rastreiam automaticamente a linhagem dos recursos usados no treinamento. Quando o conjunto de treinamento inclui recursos RequestSource, as colunas RequestSource são adicionadas à assinatura do modelo do MLflow como entradas necessárias. Isso garante que o esquema de API do endpoint de serviço reflita os campos que os chamadores devem fornecer no momento da inferência. Para obter detalhes, consulte Treinar modelos com tabelas de recursos.

Para FeatureViewSource, registre o atributo derivado e seus atributos upstream antes de registrar um modelo. As entradas de solicitação necessárias para dependências transitivas também são exigidas no momento da inferência. Veja Dependências UDF personalizadas para requisitos de pacotes de modelo.

score_batch()

Execute a inferência em lote com a busca automática de características.

FeatureEngineeringClient.score_batch(
    model_uri: str,                           # URI of logged model
    df: DataFrame,                            # DataFrame with entity keys and timestamps
) -> DataFrame

score_batch usa os metadados de recursos armazenados com o modelo para calcular automaticamente os recursos temporais corretos para uso na inferência, garantindo consistência com o treinamento. Para obter detalhes, consulte Treinar modelos com tabelas de recursos.

Fluxo de trabalho de exemplo

import mlflow
from databricks.feature_engineering import FeatureEngineeringClient
from sklearn.ensemble import RandomForestClassifier

fe = FeatureEngineeringClient()

# Assume features are registered in UC
# labeled_df should have columns "user_id", "transaction_time", and "is_fraud"

# 1. Create training set using Feature Views
training_set = fe.create_training_set(
    df=labeled_df,
    features=features,
    label="is_fraud",
)

# 2. Load training data with computed features
training_df = training_set.load_df()
X = training_df.drop("is_fraud").toPandas()
y = training_df.select("is_fraud").toPandas().values.ravel()

# 3. Train model
model = RandomForestClassifier().fit(X, y)

# 4. Log model with feature metadata
with mlflow.start_run():
    fe.log_model(
        model=model,
        artifact_path="fraud_model",
        flavor=mlflow.sklearn,
        training_set=training_set,
        registered_model_name="main.ecommerce.fraud_model",
    )

# 5. Batch scoring with automatic feature lookup
# inference_df must contain the same entity and timeseries columns
# used during training. Features are automatically computed.
predictions = fe.score_batch(
    model_uri="models:/main.ecommerce.fraud_model/1",
    df=inference_df,
)
predictions.display()

Treinamento com recursos do RequestSource

Quando o modelo exigir dados fornecidos em tempo de inferência (como detalhes da transação de uma chamada à API), use recursos RequestSource junto com recursos com suporte de tabela. Durante o treinamento, colunas RequestSource são extraídas do DataFrame rotulado.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    DeltaTableSource, Feature, FieldDefinition, RequestSource,
    ScalarDataType, ColumnSelection,
)

fe = FeatureEngineeringClient()

# RequestSource provides transaction data at inference time
request_source = RequestSource(
    schema=[
        FieldDefinition(name="transaction_amount", data_type=ScalarDataType.DOUBLE),
        FieldDefinition(name="vendor_id", data_type=ScalarDataType.STRING),
        FieldDefinition(name="transaction_id", data_type=ScalarDataType.STRING),
        FieldDefinition(name="transaction_time", data_type=ScalarDataType.DATE),
    ]
)

delta_source = DeltaTableSource(
    catalog_name="catalog",
    schema_name="schema",
    table_name="vendor_data",
)

# A column selection feature from the request source (pass-through)
latest_transaction_amount = Feature(
    source=request_source,
    function=ColumnSelection("transaction_amount"),
    name="latest_transaction_amount",
)

# A lookup feature from a delta table
vendor_category = Feature(
    source=delta_source,
    function=ColumnSelection("vendor_category"),
    entity=["vendor_id"],
    timeseries_column="transaction_time",
    name="vendor_category",
)

# labels_df must contain: transaction_id, transaction_time, vendor_id,
# transaction_amount, and the label column.
ts = fe.create_training_set(
    df=labels_df,
    features=[latest_transaction_amount, vendor_category],
    label="is_fraud",
    exclude_columns=["card_id"],
)

import mlflow
from sklearn.ensemble import RandomForestClassifier

with mlflow.start_run():
    training_df = ts.load_df().toPandas()
    X = training_df.drop(columns=["is_fraud"])
    y = training_df["is_fraud"]
    model = RandomForestClassifier().fit(X, y)

    # log_model() adds RequestSource columns to the MLflow model signature
    fe.log_model(
        model=model,
        artifact_path="fraud_model",
        flavor=mlflow.sklearn,
        training_set=ts,
        registered_model_name="catalog.schema.fraud_model",
    )

Transforme valores da solicitação com CustomUDF

Para transformar dados de requisição, use CustomUDF em vez de ColumnSelection. Este exemplo utiliza a função de transformação logarítmica de transação:

log_transaction_amount = fe.get_feature(
    full_name="main.ecommerce.log_transaction_amount"
)
request_df = spark.createDataFrame(
    [(0.0, 0), (99.0, 1)],
    "transaction_amount DOUBLE, label INT",
)

transaction_training_set = fe.create_training_set(
    df=request_df,
    features=[log_transaction_amount],
    label="label",
)
transaction_training_set.load_df().show()

O resultado inclui as colunas originais transaction_amount e label, além de log_transaction_amount. A UDF lê transaction_amount de cada linha. Colunas de solicitação necessárias para computação não podem ser listadas em exclude_columns.

Treine com recursos do FeatureViewSource

FeatureViewSource permite que CustomUDF consuma saídas de outras funcionalidades, incluindo funcionalidades derivadas. Passe as saídas que você deseja para create_training_set. Você não precisa listar as dependências intermediárias deles.

Para cada linha de entrada, o Azure Databricks resolve o grafo de dependência completo:

1. Calcula atributos upstream baseados em tabela usando suas chaves de entidade, marcas de tempo e definições de janela. Ele utiliza materializações offline compatíveis quando disponíveis.
2. Lê as colunas de requisição necessárias do DataFrame de entrada e avalia atributos baseados em requisições.
3. Avalia características derivadas em ordem de dependência, para que cada UDF receba seus resultados a montante.

O atributo derivado não introduz outra janela de tempo nem uma consulta em um ponto específico no tempo. Suas características de origem mantêm suas próprias semânticas temporais. O DataFrame deve conter as colunas de entidade, carimbo de data/hora e solicitação necessárias para esses upstreams, mesmo quando apenas o recurso derivado final é solicitado.

Por exemplo, use a característica de margem registrada, que combina revenue_sum_7d e cost_sum_7d:

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()
margin = fe.get_feature(full_name="main.ecommerce.margin")

# labeled_df contains customer_id, event_time, and label.
training_set = fe.create_training_set(
    df=labeled_df,
    features=[margin],
    label="label",
    exclude_columns=["customer_id", "event_time"],
)
training_df = training_set.load_df()

O resultado contém label e margin. As características de receita e custo são calculadas, mas não retornam como colunas extras. Para incluir receita nos dados de treinamento, recupere-a com revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") e passe features=[margin, revenue]. Isso também se aplica a cadeias multinível: solicitar a funcionalidade final não retorna cada saída intermediária.

Você pode combinar recursos baseados em solicitações, baseados em tabela e recursos derivados na mesma lista features. Para combinar seus valores dentro de um único UDF, represente os valores da solicitação como características e os referencia junto com recursos com suporte de tabela em um FeatureViewSource.

Para fins de experimentação, construa objetos Feature locais, incluindo seu grafo upstream, sem registrá-los. Use create_training_set, opcionalmente com label=None, para inspecionar os resultados. compute_features não suporta RequestSource nem FeatureViewSource.

Dependências de UDF personalizadas

Para computação offline, declare pacotes Python na cláusula ENVIRONMENT da UDF do Unity Catalog. Instalar um pacote sozinho no notebook não o instala no ambiente UDF.

Para servir modelos, também passe explicitamente os pacotes necessários para log_model. Nem o UDF ENVIRONMENT nem uma especificação de recurso nomeada contendo Feature Views fornecem automaticamente esses requisitos de modelo. Inclui dependências necessárias para UDFs upstream, bem como as saídas de recursos solicitadas.

Após treinar um modelo scikit-learn usando transaction_training_set.load_df(), registre-o com o mesmo conjunto de treinamento. Inclua o NumPy e um pacote de busca compatível:

import mlflow

fe.log_model(
    model=model,
    artifact_path="transaction_model",
    flavor=mlflow.sklearn,
    training_set=transaction_training_set,
    registered_model_name="main.ecommerce.transaction_model",
    extra_pip_requirements=[
        "numpy==1.26.4",
        "databricks-feature-lookup>=1.15.0",
    ],
)

Os pontos de extremidade de atendimento devem escolher automaticamente a databricks-feature-lookup versão 1.15.0 ou posterior, que oferece suporte à computação UDF do Catálogo do Unity sob demanda para recursos do RequestSource. Mantenha as versões do seu pacote UDF consistentes entre ambientes offline e de serviço para evitar diferenças nos valores computados. Para endpoints de Feature Serving sem modelo, declare pacotes em create_feature_spec em vez disso. Veja Adicionar dependências em Python.

Treinamento com recursos de streaming

Quando você define um Stream, o Databricks gerencia um pipeline de ingestão que grava dados de fluxo em uma tabela Delta. create_training_set lê desta tabela de ingestão e executa junções pontuais em relação ao DataFrame rotulado, assim como os recursos de lote de um DeltaTableSource. Para obter detalhes sobre configuração de ingestão, carga retroativa e deduplicação, veja Ingestão e carga retroativa.

Example

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    StreamSource,
    Feature,
    AggregationFunction,
    Sum,
    RollingWindow,
)
from datetime import timedelta

fe = FeatureEngineeringClient()

# Define a streaming feature
stream_source = StreamSource(full_name="my_catalog.my_schema.my_stream")

streaming_feature = Feature(
    name="user_purchase_sum",
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(
        operator=Sum(input="value.amount"),
        time_window=RollingWindow(window_duration=timedelta(hours=1)),
    ),
)

# Create training set — reads from the ingestion table
# labeled_df must contain "user_id", "event_time", and label columns.
# Entity and timeseries columns use leaf node names (not value. prefixes).
training_set = fe.create_training_set(
    df=labeled_df,
    features=[streaming_feature],
    label="is_fraud",
)

training_df = training_set.load_df()

Combinando funcionalidades em lote e de streaming

Os recursos de processamento em lote e em fluxo contínuo podem ser usados em conjunto no mesmo conjunto de treinamento e modelo. No momento da veiculação, os recursos em lote são consultados em repositórios offline ou online, e os recursos de streaming são consultados em repositórios online.

training_set = fe.create_training_set(
    df=labeled_df,
    features=[batch_feature, streaming_feature],
    label="is_fraud",
)

O modelo registrado com log_model() realiza pesquisas de características na loja virtual e configura a assinatura do modelo para ambos os tipos de fonte.

O que atinge o modelo bruto durante o tempo de serviço

O wrapper de modelo do Repositório de Recursos filtra as colunas antes de passá-las para o modelo bruto: