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.

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
)

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.

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

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 mais detalhes sobre a configuração de ingestão, o reenchimento e a deduplicação, consulte Ingestão e reenchimento.

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: