Trainieren von Modellen mit Feature Views

Mit Feature Views können Sie Modelle mit zeitpunktgenauer Feature-Berechnung und automatischem Feature-Abruf bei der Inferenz trainieren. Informationen zum Definieren von Featureansichten finden Sie unter Featureansichten.

Anforderungen

• Features müssen als Feature Views erstellt werden. Siehe Funktionsansichten.
• Für CustomUDF und FeatureViewSource Anforderungen siehe die Feature Views API-Referenz.

API-Methoden

create_training_set()

Nachdem Sie Featureansichten erstellt haben, besteht der nächste Schritt darin, Schulungsdaten für Ihr Modell zu erstellen. Übergeben Sie dazu ein beschriftetes Dataset an create_training_set, das automatisch die zeitpunktgenaue Berechnung jedes Merkmalswerts sicherstellt.

Beispiel:

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

Rufen Sie TrainingSet.load_df auf, um ursprüngliche Schulungsdaten mit zu einem bestimmten Zeitpunkt dynamisch computierten Merkmalen zu verbinden.

Das df Argument muss die folgenden Anforderungen erfüllen:

• Muss alle Entitätsspalten enthalten, auf die von Featuredefinitionen verwiesen wird.
• Muss die Zeitserienspalte enthalten, auf die von Featuredefinitionen verwiesen wird.
• Muss alle spalten enthalten, die in einem beliebigen RequestSource Schema deklariert sind. Typen werden anhand des deklarierten Schemas überprüft. Bei Nichtübereinstimmungen wird ein Fehler ausgelöst (keine implizite Umwandlung).
• Sollte Beschriftungsspalte(n) enthalten.
• Die Gruppe von Entitätsspaltennamen, Spaltennamen der Zeitserien und Anforderungsfunktionsspaltennamen muss global für alle Quellen eindeutig sein.

Punkt-in-Time-Korrektheit: Für Aggregation und ColumnSelection Features, die von einer Tabellenquelle unterstützt werden, werden Features nur mithilfe von Quelldaten berechnet, die vor dem Zeitstempel jeder Zeile verfügbar sind, um zukünftige Datenlecks in Modellschulungen zu verhindern. Bei RequestSource-Features wird der Wert direkt aus der gekennzeichneten Zeile des DataFrames entnommen.

log_model()

Verwenden Sie MLflow, um ein Modell mit Featuremetadaten für die Nachverfolgung von Linien und die automatische Featuresuche während der Ableitung zu protokollieren:

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
)

Der flavor Parameter gibt das zu verwendende MLflow-Modell-Variantenmodul an, z. B. mlflow.sklearn oder mlflow.xgboost.

Modelle, die mit einer TrainingSet protokolliert werden, verfolgen automatisch die Abstammung zu den Merkmalen, die in der Schulung verwendet werden. Wenn der Schulungssatz RequestSource Features enthält, werden die RequestSource Spalten als erforderliche Eingaben zur MLflow-Modellsignatur hinzugefügt. Dadurch wird sichergestellt, dass das API-Schema des Service-Endpunkts die Felder anzeigt, die Aufrufer zum Zeitpunkt der Inferenz bereitstellen müssen. Ausführliche Informationen finden Sie unter "Modelle mit Merkmalstabellen trainieren".

Für FeatureViewSource, registrieren Sie das abgeleitete Merkmal und seine vorgelagerten Merkmale, bevor ein Modell protokolliert wird. Eingaben, die von transitiven Abhängigkeiten benötigt werden, sind auch während der Inferenz erforderlich. Siehe benutzerdefinierte UDF-Abhängigkeiten für Anforderungen an Modellpakete.

score_batch()

Durchführen der Batch-Inferenz mit automatischer Merkmalsuche:

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

score_batch verwendet die Feature-Metadaten, die mit dem Modell gespeichert sind, um zeitpunktgenau korrekte Features automatisch für die Inferenzen zu berechnen, wodurch die Konsistenz mit dem Training sichergestellt wird. Ausführliche Informationen finden Sie unter "Modelle mit Merkmalstabellen trainieren".

Beispielworkflow

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

Schulung mit RequestSource-Features

Wenn Ihr Modell Daten erfordert, die zur Ableitungszeit bereitgestellt werden (z. B. Transaktionsdetails aus einem API-Aufruf), verwenden Sie RequestSource Features zusammen mit tabellengestützten Features. Während der Schulung werden Spalten aus dem beschrifteten DataFrame RequestSource extrahiert.

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

Transformieren Sie Anfragewerte mit CustomUDF

Um Anfragedaten zu transformieren, verwenden CustomUDF Sie statt ColumnSelection. Dieses Beispiel verwendet die Transaktionslog-Transformation-Funktion:

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

Das Ergebnis enthält die ursprünglichen Spalten transaction_amount und label sowie log_transaction_amount. Der UDF liest transaction_amount aus jeder Zeile. Anfragespalten, die für die Berechnung erforderlich sind, können nicht in exclude_columnsaufgeführt werden.

Trainieren mit FeatureViewSource-Funktionen

FeatureViewSource erlaubt es CustomUDF, die Ausgaben anderer Features zu nutzen, einschließlich abgeleiteter Features. Übergeben Sie die gewünschten Ausgaben an create_training_set. Sie müssen ihre Zwischenabhängigkeiten nicht auflisten.

Für jede Eingabezeile löst Azure Databricks den vollständigen Abhängigkeitsgraphen auf:

1. Berechnung von in Tabellen gespeicherten Upstream-Features anhand ihrer Entitätsschlüssel, Zeitstempel und Fensterdefinitionen. Es verwendet, sofern verfügbar, kompatible Offline-Materialisierungen.
2. Liest die erforderlichen Anfragespalten aus dem eingegebenen DataFrame und bewertet anforderungsunterstützte Funktionen.
3. Bewertet abgeleitete Features in Abhängigkeitsreihenfolge, sodass jeder UDF seine Upstream-Ergebnisse erhält.

Das abgeleitete Feature führt weder ein weiteres Zeitfenster noch ein Point-in-Time-Lookup ein. Die Upstream-Funktionen behalten ihre eigene Zeit-Semantik bei. Der DataFrame muss die Entität, den Zeitstempel und die Anforderungsspalten enthalten, die von diesen Upstreams benötigt werden, selbst wenn nur die endgültige abgeleitete Funktion angefordert wird.

Verwenden Sie beispielsweise die registrierte Randfunktion, die cost_sum_7d und revenue_sum_7d kombiniert:

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

Das Ergebnis enthält label und margin. Die Umsatz- und Kostenmerkmale werden berechnet, aber nicht als zusätzliche Spalten zurückgegeben. Um Einnahmen in die Trainingsdaten einzubeziehen, rufen Sie sie mit revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") ab und übergeben Sie features=[margin, revenue]. Dies gilt auch für mehrstufige Ketten: Das Anfordern des finalen Merkmals gibt nicht jedes Zwischenergebnis zurück.

Sie können Anfrage-gestützte, tabellenbasierte und abgeleitete Funktionen in derselben features Liste kombinieren. Um ihre Werte in einem UDF zu kombinieren, repräsentieren Sie die Anfragewerte als Features und referenzieren Sie sie zusammen mit tabellengestützten Features in einem FeatureViewSource.

Für Experimente konstruiere lokale Feature Objekte, einschließlich ihres vorgelagerten Graphen, ohne sie zu registrieren. Verwenden Sie create_training_set, optional mit label=None, um die Ergebnisse zu überprüfen. compute_features unterstützt nicht RequestSource oder FeatureViewSource.

Benutzerdefinierte UDF-Abhängigkeiten

Für Offline-Berechnungen deklarieren Sie Python-Pakete in der ENVIRONMENT Klausel des Unity Catalog UDF. Das Installieren eines Pakets allein im Notebook installiert es nicht in der UDF-Umgebung.

Für das Modellserving übergebe auch explizit die erforderlichen Pakete an log_model. Weder die UDF ENVIRONMENT noch eine benannte Funktionsspezifikation mit Funktionsansichten stellt diese Modellanforderungen automatisch bereit. Fügen Sie Abhängigkeiten ein, die von vorgelagerten UDFs benötigt werden, sowie die angeforderten Feature-Ausgaben.

Nachdem ein Scikit-learn-Modell mit transaction_training_set.load_df() trainiert wurde, protokollieren Sie es mit demselben Trainingssatz. Fügen Sie NumPy und ein kompatibles Suchpaket hinzu:

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

Bereitstellungsendpunkte sollten automatisch databricks-feature-lookup Version 1.15.0 oder höher verwenden, die die bedarfsgesteuerte Berechnung von Unity-Catalog-UDFs für RequestSource-Features unterstützt. Halten Sie Ihre UDF-Paketversionen konsistent zwischen Offline- und Serving-Umgebungen, um Unterschiede in den berechneten Werten zu vermeiden. Für Feature Serving Endpunkte ohne Modell deklarieren Sie stattdessen Pakete auf .create_feature_spec Siehe Add Python dependencies.

Schulung mit Streamingfeatures

Wenn Sie einen Stream definieren, verwaltet Databricks eine Aufnahmepipeline, die Datenstromdaten in eine Delta-Tabelle schreibt. create_training_set liest aus dieser Erfassungstabelle und führt Point-in-Time-Joins mit Ihrem gelabelten DataFrame aus, genau wie Batch-Features aus einer DeltaTableSource. Details zur Konfiguration der Datenerfassung, zu Backfill und zur Deduplizierung finden Sie unter Datenerfassung und Backfill.

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

Mischen von Batch- und Streamingfeatures

Batch- und Streamingfeatures können zusammen im gleichen Schulungssatz und -modell verwendet werden. Zum Bereitstellungszeitpunkt werden Batch-Features aus Offline- oder Online-Stores abgerufen und Streaming-Features werden aus Online-Stores abgerufen.

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

Das mit log_model() protokollierte Modell führt Feature-Abfragen aus dem Online-Store durch und konfiguriert die Modellsignatur für beide Quelltypen.

Was erreicht das Rohmodell zur Ausführungszeit

Der Feature Store Modell-Wrapper filtert Spalten, bevor sie an das unverarbeitete Modell übergeben werden.