Träna modeller med Feature Views

Med funktionsvyer kan du träna modeller med rätt funktionsberäkning och automatisk funktionssökning vid slutsatsdragning. Information om hur du definierar funktionsvyer finns i Funktionsvyer.

Kravspecifikation

• Funktioner måste skapas som Feature Views. Se Funktionsvyerna.

API-metoder

create_training_set()

När du har skapat funktionsvyer är nästa steg att skapa träningsdata för din modell. För att göra detta skickar du en märkt datauppsättning till create_training_set, som automatiskt säkerställer korrekt beräkning vid tidpunkt för varje funktionsvärde.

Som exempel:

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

Anropa TrainingSet.load_df för att ansluta ursprungliga träningsdata med dynamiskt beräknade tidsspecifika funktioner.

Argumentet df måste uppfylla följande krav:

• Måste innehålla alla entitetskolumner som refereras av funktionsdefinitioner.
• Måste innehålla kolumnen timeseries som refereras till av funktionsdefinitioner.
• Måste innehålla alla kolumner som deklarerats i ett RequestSource schema. Typer verifieras mot det deklarerade schemat. Felmatchningar orsakar ett fel (ingen implicit typkonvertering).
• Ska innehålla etikettkolumn(er).
• Uppsättningen med entitetskolumnnamn, tidseriekolumnnamn och namn på funktionskolumner för begäran måste vara globalt unika för alla källor.

Rätt tidpunkt: För aggregering och ColumnSelection funktioner som backas upp av en tabellkälla beräknas funktioner med endast källdata som är tillgängliga före varje rads tidsstämpel, för att förhindra framtida dataläckage i modellträningen. För RequestSource funktioner hämtas värdet direkt från den märkta DataFrame-raden.

log_model()

Använd MLflow för att logga en modell med funktionsmetadata för ursprungsspårning och automatisk funktionssökning under slutsatsdragning:

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
)

Parametern flavor anger den MLflow-modellsmakmodul som ska användas, till exempel mlflow.sklearn eller mlflow.xgboost.

Modeller som loggas med en TrainingSet spårar automatiskt ursprung till funktionerna som används i träning. När träningsuppsättningen innehåller RequestSource funktioner läggs kolumnerna RequestSource till i MLflow-modellsignaturen som nödvändiga indata. Detta säkerställer att serverslutpunktens API-schema återspeglar de fält som anroparna måste ange vid slutsatsdragningstidpunkten. Mer information finns i Träningsmodeller med funktioner.

score_batch()

Utför batchinferens med automatisk funktionssökning:

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

score_batch använder funktionsmetadata som lagras med modellen för att automatiskt beräkna rätt funktioner vid tidpunkt för slutsatsdragning, vilket säkerställer konsekvens med träning. Mer information finns i Träningsmodeller med funktioner.

Exempelarbetsflöde

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

Utbildning med RequestSource-funktioner

När din modell kräver data som tillhandahålls vid inferens (till exempel transaktionsinformation från ett API-anrop) använder du RequestSource funktioner tillsammans med tabellbaserade funktioner. Under träningen extraheras kolumner från den märkta DataFrame.

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

Träning med strömningsfunktioner

När du definierar en stream hanterar Databricks en inmatningspipeline som skriver dataström till en Delta-tabell. create_training_set läser från den här inmatningstabellen och utför punkt-i-tid-kopplingar mot din märkta DataFrame, precis som batchfunktioner från en DeltaTableSource. Mer information om konfiguration, återfyllnad och deduplicering finns i Inmatning och återfyllnad.

Exempel

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

Kombinera batch- och strömningsfunktioner

Batch- och strömningsfunktioner kan användas tillsammans i samma träningsuppsättning och modell. Vid serveringstillfället letas batchfunktioner upp från offline- eller onlinebutiker, och streamingfunktionerna letas upp från onlinebutiker.

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

Modellen som loggas med log_model() utför funktionssökningar från onlinebutiken och konfigurerar modellsignaturen för båda källtyperna.

Vad når råmodellen vid driftstid

Modellomslutningen för Funktionsarkivet filtrerar kolumner innan de skickas till råmodellen.