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.
• För CustomUDF och FeatureViewSource krav, se referensen Feature Views API.

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
    extra_pip_requirements: Optional[List[str]] = None, # Optional: Additional serving dependencies
)

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.

För FeatureViewSource, registrera den härledda funktionen och dess uppströmsfunktioner innan en modell loggas. Indata som begärs av transitiva beroenden krävs också vid inferens. Se Anpassade UDF-beroenden för krav på modellpaket.

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

Transformera förfrågningsvärden med CustomUDF

För att transformera begärandata, använd CustomUDF istället för ColumnSelection. Detta exempel använder transaktionslogg-transformfunktionen:

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

Resultatet inkluderar originalet transaction_amount och label kolumnerna, plus log_transaction_amount. UDF läser transaction_amount från varje rad. Begäransoloner som krävs för beräkning kan inte listas i exclude_columns.

Träna med funktionerna i FeatureViewSource.

FeatureViewSource låter en CustomUDF konsumera andra funktionsutdata, inklusive härledda funktioner. Skicka den utdata du vill till create_training_set. Du behöver inte lista deras mellanliggande beroenden.

För varje inmatningsrad löser Azure Databricks hela beroendegrafen:

1. Beräknar tabellstödda uppströmsfunktioner med hjälp av deras entitetsnycklar, tidsstämplar och fönsterdefinitioner. Den använder kompatibla offline-materialiseringar när det finns tillgängligt.
2. Läser kolumner som krävs för begäran från indata-DataFrame och utvärderar begärandebaserade funktioner.
3. Utvärderar härledda funktioner i beroendeordning, så att varje UDF får sina uppströmsresultat.

Den härledda funktionen introducerar inte ett nytt tidsfönster eller en uppslagning vid en viss tidpunkt. Dess uppströmsfunktioner behåller sin egen tidssemantik. DataFrame måste innehålla entitet, tidsstämpel och önskekolumner som behövs av dessa upstreams, även när endast den sista härledda funktionen efterfrågas.

Använd till exempel den registrerade funktionen marginal, som kombinerar revenue_sum_7d och 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()

Resultatet innehåller label och margin. Intäkts- och kostnadsfunktionerna beräknas men returneras inte som extra kolumner. För att inkludera intäkter i träningsdata, hämta dem med revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") och skicka vidare features=[margin, revenue]. Detta gäller även för flernivåkedjor: att begära den sista funktionen returnerar inte varje mellanliggande utgång.

Du kan kombinera begärandebaserade, tabellbaserade och härledda attribut i samma features lista. För att kombinera deras värden inom en UDF, representera begäranvärdena som funktioner och referera dem tillsammans med tabellbaserade funktioner i en FeatureViewSource.

För experiment, konstruera lokala Feature objekt, inklusive deras uppströms graf, utan att registrera dem. Använd create_training_set, valfritt med label=None, för att inspektera resultaten. compute_features stöder inte RequestSource eller FeatureViewSource.

Anpassade UDF-beroenden

För offline-beräkning, deklarera Python-paket i Unity Catalog UDF:s ENVIRONMENT klausul. Att installera ett paket enbart i anteckningsboken installerar det inte i UDF-miljön.

För modellservering, skicka också de nödvändiga paketen explicit till log_model. Varken UDF:erna ENVIRONMENT eller en namngiven funktionsspecifikation som innehåller funktionsvyer tillhandahåller automatiskt dessa modellkrav. Inkludera beroenden som krävs av överordnade UDF:er samt utdata för de begärda funktionerna.

Efter att ha tränat en scikit-learn-modell på transaction_training_set.load_df(), logga den tillsammans med samma träningsdata. Inkludera NumPy och ett kompatibelt sökpaket:

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

Serveringsslutpunkter bör automatiskt hämta databricks-feature-lookup version 1.15.0 eller senare, som stöder Unity Catalog UDF-beräkning vid behov för RequestSource-funktioner. Håll dina UDF-paketversioner konsekventa mellan offline- och servingsmiljöer för att undvika skillnader i beräknade värden. För Feature Serving-endpoints utan modell, deklarera paket på create_feature_spec istället. Se Lägg till Python-beroenden.

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 av datainmatning, bakåtfyllning och deduplicering finns i Datainmatning och bakåtfyllning.

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.