Entrenamiento de modelos con vistas de características

Las vistas de características permiten entrenar modelos con cálculos de características correctos a un momento dado y búsqueda automática de características en la inferencia. Para obtener información sobre cómo definir vistas de características, vea Vistas de características.

Requirements

• Las funcionalidades deben crearse como Feature Views. Consulte Vistas de funciones.
• Para los requisitos de CustomUDF y FeatureViewSource, consulta la referencia de la API de Feature Views.

Métodos de API

create_training_set()

Después de crear vistas de características, el siguiente paso es crear datos de entrenamiento para el modelo. Para ello, pase un conjunto de datos etiquetado a create_training_set, que garantiza automáticamente un cálculo preciso a un momento dado de cada valor de característica.

Por ejemplo:

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

Llame a TrainingSet.load_df para combinar los datos de entrenamiento originales con características calculadas dinámicamente en un momento dado.

El df argumento debe cumplir los siguientes requisitos:

• Debe contener todas las columnas de entidad a las que hacen referencia las definiciones de características.
• Debe contener la columna timeseries a la que hacen referencia las definiciones de características.
• Debe contener todas las columnas declaradas en cualquier RequestSource esquema. Los tipos se validan con el esquema declarado. Los errores de coincidencia generan un error (sin conversión implícita).
• Debe contener columnas de etiqueta.
• El conjunto de nombres de columna de entidad, los nombres de columna de series temporales y los nombres de columna de funcionalidades de solicitud deben ser únicos globalmente en todos los orígenes.

Corrección en un momento dado: Para la agregación y las características respaldadas por un origen de tabla (ColumnSelection), las características se calculan utilizando únicamente los datos de origen disponibles antes del timestamp de cada fila, para evitar que se filtren datos futuros en el entrenamiento del modelo. Para las RequestSource características, el valor se toma directamente de la fila etiquetada del DataFrame.

log_model()

Use MLflow para registrar un modelo con metadatos de características para el seguimiento de linaje y la búsqueda automática de características durante la inferencia:

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
)

El flavor parámetro especifica el módulo de variante del modelo de MLflow que se va a usar, como mlflow.sklearn o mlflow.xgboost.

Los modelos registrados con un TrainingSet hacen un seguimiento automático del linaje de las características usadas en el entrenamiento. Cuando el conjunto de entrenamiento incluye RequestSource características, las columnas RequestSource se agregan a la firma del modelo de MLflow como entradas requeridas. Esto garantiza que el esquema de API del endpoint de servicio refleje los campos que los solicitantes deben proporcionar durante la inferencia. Para detalles, consulte Entrenar modelos con tablas de características.

Para FeatureViewSource, registra el atributo derivado y sus atributos de origen antes de registrar un modelo. Los datos de entrada de la solicitud que necesitan las dependencias transitivas también son necesarios en tiempo de inferencia. Consulte Dependencias UDF personalizadas para los requisitos del paquete modelo.

score_batch()

Realice la inferencia por lotes con la búsqueda 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 los metadatos de características almacenados con el modelo para calcular automáticamente las características correctas a un momento dado para la inferencia, lo que garantiza la coherencia con el entrenamiento. Para detalles, consulte Entrenar modelos con tablas de características.

Flujo de trabajo de ejemplo

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

Entrenamiento con características de RequestSource

Cuando su modelo requiere datos que se proporcionan en el momento de la inferencia (como los detalles de transacción de una llamada API), use RequestSource funcionalidades junto con funcionalidades respaldadas por tablas. Durante el entrenamiento, las columnas RequestSource se extraen del DataFrame etiquetado.

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 los valores de la solicitud con CustomUDF

Para transformar los datos de la solicitud, use CustomUDF en lugar de ColumnSelection. Este ejemplo utiliza la función de transformación logarítmica de transacciones:

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

El resultado incluye las columnas originales transaction_amount y label, además de log_transaction_amount. La UDF lee transaction_amount de cada fila. Las columnas de solicitud requeridas para el cálculo no pueden listarse en exclude_columns.

Entrenamiento con características de FeatureViewSource

FeatureViewSource permite que un CustomUDF consuma las salidas de otras características, incluidas las características derivadas. Envía los resultados que quieras a create_training_set. No necesitas listar sus dependencias intermedias.

Para cada fila de entrada, Azure Databricks resuelve el grafo de dependencias completo:

1. Calcula las características ascendentes respaldadas por tablas utilizando sus claves de entidad, marcas de tiempo y definiciones de ventanas. Utiliza materializaciones offline compatibles cuando está disponible.
2. Lee las columnas de petición requeridas del DataFrame de entrada y evalúa las características respaldadas por peticiones.
3. Evalúa las características derivadas siguiendo el orden de dependencias, de modo que cada UDF recibe los resultados previos de los que depende.

La característica derivada no introduce otra ventana temporal ni búsqueda en un punto en el tiempo. Sus funcionalidades de origen conservan su propia semántica temporal. El DataFrame debe contener la entidad, la marca de tiempo y las columnas de solicitud necesarias para quienes están aguas arriba, incluso cuando solo se solicita la característica derivada final.

Por ejemplo, utiliza la función registrada de margen, que combina revenue_sum_7d y 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()

El resultado contiene label y margin. Las características de ingresos y costes se calculan pero no se devuelven como columnas adicionales. Para incluir los ingresos en los datos de entrenamiento, recupéralos con revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") y pasa features=[margin, revenue]. Esto también se aplica a cadenas multinivel: solicitar la característica final no devuelve cada salida intermedia.

Puedes combinar características respaldadas por solicitudes, con respaldo en tabla y derivadas en la misma features lista. Para combinar sus valores dentro de una UDF, represente los valores de la solicitud como variables y haga referencia a ellos junto con variables basadas en tablas en un FeatureViewSource.

Para experimentar, construye objetos locales Feature , incluyendo su grafo ascendente, sin registrarlos. Usar create_training_set, opcionalmente con label=None, para inspeccionar los resultados. compute_features no soporta RequestSource ni FeatureViewSource.

Dependencias personalizadas de UDF

Para el cálculo offline, declara paquetes Python en la cláusula UDF ENVIRONMENTdel Unity Catalog. Instalar un paquete solo en el portátil no lo instala en el entorno UDF.

Para el servicio de modelos, también pasa explícitamente los paquetes requeridos a log_model. Ni la UDF ENVIRONMENT ni una especificación de función con nombre que contenga vistas de funciones proporcionan automáticamente estos requisitos del modelo. Incluye las dependencias que necesitan las UDF ascendentes, así como las salidas de las características solicitadas.

Después de entrenar un modelo scikit-learn con transaction_training_set.load_df(), regístralo con ese mismo conjunto de entrenamiento. Incluye NumPy y un paquete de búsqueda compatible:

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

Los endpoints de servicio deben usar automáticamente databricks-feature-lookup la versión 1.15.0 o posterior, que admite el cálculo bajo demanda de UDF de Unity Catalog para funciones de RequestSource. Mantén las versiones de tus paquetes UDF consistentes entre entornos offline y de servicio para evitar diferencias en los valores computados. Para endpoints de Feature Serving sin modelo, declara paquetes en create_feature_spec en su lugar. Consulta Añadir dependencias de Python.

Entrenamiento con características de streaming

Al definir una transmisión, Databricks administra una pipeline que escribe datos de transmisión en una tabla Delta. create_training_set lee de esta tabla de ingesta y realiza uniones puntuales en el tiempo con su DataFrame etiquetado, igual que los atributos por lotes de un DeltaTableSource. Para detalles sobre la configuración de la ingestión, el relleno y la deduplicación, véase Ingestión y relleno.

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

Combinación de características de procesamiento por lotes y streaming

Las características de procesamiento por lotes y streaming se pueden usar juntas en el mismo conjunto de entrenamiento y modelo. En el tiempo de servicio, las características por lotes se buscan desde tiendas sin conexión o en línea, y las características de streaming se buscan desde tiendas en línea.

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

El modelo registrado con log_model() realiza búsquedas de características desde la tienda en línea y configura la firma del modelo para ambos tipos de origen.

Lo que llega al modelo sin formato en el momento de servicio

El envoltorio del modelo del Feature Store filtra las columnas antes de pasarlas al modelo sin procesar: