Treinar modelos com Feature Views

As Feature Views permitem-lhe treinar modelos com cálculo de características historicamente correto e consulta automática de características durante a inferência. Para informações sobre a definição de Feature Views, consulte Feature Views.

Requisitos

• As funcionalidades devem ser criadas como Vistas de Funcionalidades. Ver Vistas de Funcionalidades.
• Para os requisitos de CustomUDF e FeatureViewSource, consulte a referência da API de Feature Views.

Métodos API

create_training_set()

Depois de criar Feature Views, o passo seguinte é criar dados de treino para o seu modelo. Para fazer isto, passe um conjunto de dados rotulado para create_training_set, que garante automaticamente o cálculo temporal preciso de cada valor do atributo.

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

Chamar TrainingSet.load_df para associar dados de treino originais com características computadas dinamicamente num momento específico no tempo.

O df argumento deve cumprir os seguintes requisitos:

• Deve conter todas as colunas de entidade referenciadas por definições de características.
• Deve conter a coluna da série temporal referenciada pelas definições de características.
• Deve conter todas as colunas declaradas em qualquer RequestSource esquema. Os tipos são validados contra o esquema declarado. As incompatibilidades geram um erro (sem conjuração implícita).
• Deve conter coluna(s) de rótulos.
• O conjunto de nomes de colunas de entidades, nomes de colunas de séries temporais e nomes de colunas de características de pedido deve ser globalmente único em todas as fontes.

Correção em ponto no tempo: Para agregação e ColumnSelection características suportadas por uma tabela como fonte, as características são calculadas usando apenas dados disponíveis antes do carimbo temporal de cada linha, para evitar vazamento futuro de dados no treino do modelo. Para os RequestSource atributos, o valor é obtido diretamente da linha rotulada do DataFrame.

log_model()

Use o MLflow para registar um modelo com metadados de características para rastreio de linhagem e pesquisa automática de características 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 flavor parâmetro especifica o módulo de sabor do modelo MLflow a utilizar, como mlflow.sklearn ou mlflow.xgboost.

Modelos registados com um TrainingSet rastreiam automaticamente a linhagem com base nas características usadas no treino. Quando o conjunto de treino inclui RequestSource funcionalidades, as RequestSource colunas são adicionadas à assinatura do modelo MLflow como entradas necessárias. Isto garante que o esquema da API do endpoint de serviço reflete os campos que os chamadores devem fornecer no momento da inferência. Para detalhes, veja Modelos de comboios com tabelas de características.

Para FeatureViewSource, regista a funcionalidade derivada e as respetivas funcionalidades ascendentes antes de registar um modelo. As entradas de pedido necessárias para dependências transitivas também são necessárias no momento da inferência. Consulte Dependências UDF personalizadas para requisitos de pacotes modelo.

score_batch()

Realize inferência em lote com pesquisa automática de funcionalidades:

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

score_batch utiliza os metadados de características armazenados com o modelo para calcular automaticamente as características corretas no ponto no tempo para inferência, garantindo consistência com o treinamento. Para detalhes, veja Modelos de comboios com tabelas de características.

Exemplo de fluxo de trabalho

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

Treino com funcionalidades do RequestSource

Quando o seu modelo necessita de dados fornecidos no momento da inferência (como detalhes de transações de uma chamada de API), use RequestSource características juntamente com recursos suportados por tabela. Durante o treino, as 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",
    )

Transformar valores do pedido com CustomUDF

Para transformar dados de pedido, use CustomUDF em vez de ColumnSelection. Este exemplo utiliza a funcionalidade 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 em cada linha. Colunas de pedido necessárias para computação não podem ser listadas em exclude_columns.

Treinar com funcionalidades do FeatureViewSource

FeatureViewSource permite que CustomUDF consuma saídas de outras funcionalidades, incluindo funcionalidades derivadas. Passe os resultados que pretende para create_training_set. Não precisas de listar as dependências intermédias deles.

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

1. Calcula atributos upstream suportados por tabela com base nas suas chaves de entidade, marcas temporais e definições de janelas. Utiliza materializações offline compatíveis quando disponíveis.
2. Lê as colunas necessárias do pedido a partir do DataFrame de entrada e avalia as características baseadas no pedido.
3. Avalia as características derivadas por ordem de dependência, para que cada UDF receba os seus resultados a montante.

A funcionalidade derivada não introduz outra janela temporal ou consulta pontual no tempo. As suas funcionalidades ascendentes mantêm a sua própria semântica temporal. A DataFrame deve incluir a entidade, a marca temporal e as colunas de solicitação necessárias pelas dependências a montante, mesmo quando apenas a característica derivada final é solicitada.

Por exemplo, use a funcionalidade de margem registada, 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 funcionalidades de receita e custo são calculadas, mas não são devolvidas como colunas extra. Para incluir a receita nos dados de treino, recupere-a com revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") e passe features=[margin, revenue]. Isto aplica-se também a cadeias multinível: pedir a funcionalidade final não retorna cada saída intermédia.

Pode combinar atributos suportados por pedidos, suportados por tabela e derivados na mesma lista features. Para combinar os respetivos valores num único UDF, represente os valores da solicitação como características e faça referência aos mesmos juntamente com características suportadas por tabela num FeatureViewSource.

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

Dependências UDF personalizadas

Para processamento offline, declare pacotes Python na cláusula ENVIRONMENT da UDF do Unity Catalog. Instalar um pacote só no portátil não o instala no ambiente UDF.

Para disponibilizar modelos, passe também, de forma explícita, os pacotes necessários para log_model. Nem a UDF ENVIRONMENT nem uma especificação nomeada de funcionalidade que contenha Vistas de Funcionalidade fornecem automaticamente estes requisitos do modelo. Incluir as dependências de que os UDFs a montante necessitam, bem como as saídas da funcionalidade solicitada.

Depois de treinar um modelo scikit-learn em transaction_training_set.load_df(), registe-o com esse mesmo conjunto de treino. Inclua o NumPy e um pacote de pesquisa 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 endpoints de serviço devem adotar automaticamente a versão 1.15.0 ou posterior de databricks-feature-lookup, que suporta o cálculo UDF a pedido no Unity Catalog para funcionalidades de RequestSource. Mantenha as versões dos seus pacotes 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.

Treino com funcionalidades de streaming

Quando defines um Stream, o Databricks gere um pipeline de ingestão que escreve os dados do stream numa tabela Delta. create_training_set lê desta tabela de ingestão e realiza junções pontuais no tempo com o seu DataFrame rotulado, tal como as características em lote de um DeltaTableSource. Para detalhes sobre a configuração da ingestão, o preenchimento retroativo e a deduplicação, consulte Ingestão e preenchimento retroativo.

Exemplo

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

Mistura de funcionalidades em lote e streaming

Funcionalidades de batch e streaming podem ser usadas em conjunto no mesmo conjunto de treino e modelo. Na altura do serviço, as funcionalidades de lote são consultadas em lojas offline ou online, e as funcionalidades de streaming são consultadas nas lojas online.

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

O modelo registado com log_model() efetua consultas de características no armazenamento online e configura a assinatura do modelo para ambos os tipos de origem.

O que chega ao modelo bruto no momento da inferência

O wrapper do modelo Feature Store filtra as colunas antes de as passar ao modelo bruto: