Ensinar modelos com Recurso Views

info

Visualização

Esse recurso está em Prévia Pública. Os administradores do espaço de trabalho podem controlar o acesso a esse recurso na página Pré-visualizações . Consulte Gerenciar prévias do Databricks.

Views de Recurso permitem que você ensine modelos com cálculo de recurso correto em um determinado momento e pesquisa automática de recurso na inferência. Para obter informação sobre como definir recursos View, consulte recursos View.

Requisitos

• Recursos devem ser criados como Views de Recurso. Consulte View de recurso.
• Para os requisitos de CustomUDF e FeatureViewSource, consulte a referência da API de recurso views.

Métodos da API

create_training_set()

Depois de criar View de recursos, o próximo passo é criar dados de treinamento para seu modelo. Para fazer isso, passe um dataset rotulado para create_training_set, que garante automaticamente o cálculo preciso de cada valor de recurso em um determinado momento.

Por exemplo:

Python

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 join os dados de treinamento originais com o recurso de cálculo dinâmico pontual.

O argumento df deve atender aos seguintes requisitos:

• Deve conter todas as colunas de entidade referenciadas pelas definições de recurso.
• Deve conter a coluna de série temporal referenciada pelas definições de recurso.
• Deve conter todas as colunas declaradas em qualquer esquema RequestSource. Tipos são validados contra o esquema declarado. Incompatibilidades geram um erro (sem conversão implícita).
• Deve conter coluna(s) de rótulo.
• O conjunto de nomes de colunas de entidade, nomes de colunas de séries temporais e nomes de colunas de recursos de solicitação deve ser globalmente exclusivo em todas as fontes.

Correção pontual: Para agregações e recursos ColumnSelection baseados em uma fonte 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 futuro de dados no treinamento do modelo. Para o recurso RequestSource , o valor é obtido diretamente da linha DataFrame rótulo.

log_model()

Use MLflow para log um modelo com metadados de recursos para acompanhamento de linhagem e busca automática de recursos durante a inferência:

Python

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 de sabor do modelo MLflow a ser usado, como mlflow.sklearn ou mlflow.xgboost.

Os registros de modelos com um TrainingSet rastreiam automaticamente a linhagem até o recurso usado no treinamento. Quando o conjunto de treinamento inclui RequestSource recurso, as colunas RequestSource são adicionadas à assinatura do modelo MLflow como entradas obrigatórias. Isso garante que o esquema da API do endpoint de serviço reflita os campos que os chamadores devem fornecer no momento da inferência. Para mais detalhes, consulte ensinar modelos com tabelas de recursos.

Para FeatureViewSource, registre o recurso derivado e seus recursos de upstream antes de registrar em log um modelo. As entradas de solicitação necessárias pelas dependências transitivas também são necessárias no momento da inferência. Consulte Dependências de UDF personalizadas para conhecer os requisitos de pacote do modelo.

score_batch()

Realizar inferência de lotes com busca automática de recursos:

Python

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 recurso armazenados com o modelo para compute automaticamente o recurso correto em um determinado momento para inferência, garantindo a consistência com o treinamento. Para mais detalhes, consulte ensinar modelos com tabelas de recursos.

Exemplo de fluxo de trabalho

Python

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 recurso RequestSource

Quando seu modelo requer dados que são fornecidos no momento da inferência (como detalhes de transação de uma chamada API ), use o recurso RequestSource juntamente com o recurso baseado em tabela. Durante o treinamento, RequestSource colunas são extraídas do DataFrame de rótulo.

Python

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 de solicitação com CustomUDF

To transform request data, use CustomUDF instead of ColumnSelection. Este exemplo usa o recurso de log-transform de transação:

Python

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. O UDF lê transaction_amount de cada linha. As colunas de solicitação necessárias para o cálculo não podem ser listadas em exclude_columns.

Ensinar com recursos do FeatureViewSource

FeatureViewSource permite que um CustomUDF consuma outras saídas de recurso, incluindo recursos derivados. Passe as saídas que você deseja para create_training_set. Você não precisa listar as dependências intermediárias delas.

Para cada linha de entrada, o Databricks resolve o gráfico de dependência completo:

1. Calcula recursos de upstream baseados em tabela usando suas chaves de entidade, timestamps e definições de janela. Ele usa materializações offline compatíveis quando disponíveis.
2. Lê as colunas de solicitação necessárias do DataFrame de entrada e avalia os recursos com base em solicitações.
3. Avalia recursos derivados na ordem de dependência, para que cada UDF receba seus resultados de upstream.

O recurso derivado não introduz outra janela de tempo ou pesquisa pontual. Seus recursos upstream retêm suas próprias semânticas de tempo. O DataFrame deve conter as colunas de entidade, timestamp e solicitação necessárias para esses upstreams, mesmo quando apenas o recurso derivado final for solicitado.

Por exemplo, use o recurso de margem registrado, que combina revenue_sum_7d e cost_sum_7d:

Python

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. Os recursos de receita e custo são computados, mas não são retornado como colunas extras. Para incluir a 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 de vários níveis: solicitar o recurso final não retorna cada saída intermediária.

Você pode combinar recursos baseados em solicitação, baseados em tabela e derivados na mesma lista de features. Para combinar seus valores dentro de um UDF, represente os valores de solicitação como recursos e referencie-os junto com os recursos baseados em tabela em um FeatureViewSource.

Para experimentação, construa objetos Feature locais, incluindo seu gráfico upstream, sem registrá-los. Use create_training_set, opcionalmente com label=None, para inspecionar os resultados. compute_features não oferece suporte a RequestSource ou FeatureViewSource.

nota

O limite de cinco chamadas de Unity Catalog UDF por query também se aplica a queries de treinamento. Conte as chamadas de UDF necessárias para o gráfico de dependência completo, e não apenas para os recursos solicitados como saídas. Esse limite de query é separado do limite de profundidade do gráfico.

Dependências de UDF Personalizadas

For offline computation, declare Python pacotes in the Unity Catalog UDF's ENVIRONMENT clause. A instalação de um pacote apenas no notebook não o instala no ambiente UDF.

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

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

Python

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

Serving endpoints should automatically pick up databricks-feature-lookup version 1.15.0 or later, which supports on-demand Unity Catalog UDF computation for RequestSource recursos. Mantenha as versões dos pacotes UDF consistentes entre os ambientes offline e de serviço para evitar diferenças nos valores computados. Para endpoints do Feature Serving sem um modelo, declare pacotes em create_feature_spec. Consulte Adicionar dependências do Python.

Treinamento com recursos de transmissão

When you define a transmissão, Databricks manages an ingestion pipeline that writes transmissão data into a Delta table. create_training_set reads from this ingestion table and performs point-in-time joins against your rotulado DataFrame, just like batch recursos from a DeltaTableSource. For details on ingestion configuration, backfill, and deduplication, see Ingestion and backfill.

Exemplo

Python

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

Misturando recursos de lote e transmissão

Os recursos de lotes e transmissão podem ser usados juntos no mesmo conjunto de treinamento e modelo. No momento do serving, os recursos de lotes são pesquisados em lojas offline ou online, e os recursos de transmissão são pesquisados em lojas online.

Python

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

O modelo registrado com log_model() executa pesquisas de recursos do armazenamento online e configura a assinatura do modelo para ambos os tipos de fonte.

O que chega ao modelo bruto no momento do atendimento?

O wrapper do modelo Feature Store filtra as colunas antes de passá-las para o modelo bruto:

Tipo de coluna	Atinge o modelo interno?
Saídas de recurso explícitas (ColumnSelection, agregação)	Sim
RequestSource colunas declaradas como recurso	Sim
Colunas de entidade (chave de pesquisa)	Não (a menos que seja explicitamente declarado como um recurso)
Colunas de séries temporais	Não (a menos que seja explicitamente declarado como um recurso)

Tipo de coluna	Atinge o modelo interno?
Saídas de recurso explícitas (ColumnSelection, agregação)	Sim
RequestSource colunas declaradas como recurso	Sim
Colunas de entidade (chave de pesquisa)	Não (a menos que seja explicitamente declarado como um recurso)
Colunas de séries temporais	Não (a menos que seja explicitamente declarado como um recurso)