Modellen trainen met kenmerkenweergaven

Met Feature Views kunt u modellen trainen met tijdstipcorrecte featureberekening en automatische featureopzoeking tijdens inferentie. Voor informatie over het definiëren van functieweergaven, zie Functieweergaven.

Requirements

• Features moeten als Feature Views worden gemaakt. Zie functieweergaven.
• Voor CustomUDF en FeatureViewSource vereisten, zie de Feature Views API-referentie.

API-methoden

create_training_set()

Nadat u functieweergaven hebt gemaakt, bestaat de volgende stap uit het maken van trainingsgegevens voor uw model. Om dit te doen, geeft u een gelabelde gegevensset door aan create_training_set, waarmee automatisch een nauwkeurige berekening van elke functiewaarde naar een bepaald tijdstip wordt gegarandeerd.

Voorbeeld:

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

Roep TrainingSet.load_df aan om oorspronkelijke trainingsgegevens te koppelen met tijdstipgebonden dynamisch berekende kenmerken.

Het df argument moet voldoen aan de volgende vereisten:

• Moet alle entiteitskolommen bevatten waarnaar wordt verwezen door functiedefinities.
• Moet de kolom timeseries bevatten waarnaar wordt verwezen door functiedefinities.
• Moet alle kolommen bevatten die zijn gedeclareerd in een RequestSource schema. Typen worden gevalideerd op basis van het gedeclareerde schema. Niet-overeenkomsten resulteren in een fout (geen impliciete typeconversie).
• Moet labelkolom(s) bevatten.
• De set kolomnamen van entiteiten, tijdreekskolomnamen en kolomnamen van aanvraagfuncties moet globaal uniek zijn in alle bronnen.

Correctheid van een bepaald tijdstip: Voor aggregatie en ColumnSelection functies die worden ondersteund door een tabelbron, worden functies berekend met behulp van alleen brongegevens die beschikbaar zijn vóór de tijdstempel van elke rij, om toekomstige gegevenslekken in modeltraining te voorkomen. Voor RequestSource functies wordt de waarde rechtstreeks uit de gelabelde DataFrame-rij gehaald.

log_model()

Gebruik MLflow om een model te registreren met functiemetagegevens voor het bijhouden van herkomst en het automatisch opzoeken van functies tijdens deductie:

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
)

De flavor parameter geeft de module MLflow-modelsmaak op die moet worden gebruikt, zoals mlflow.sklearn of mlflow.xgboost.

Modellen die zijn geregistreerd met een TrainingSet houden automatisch de gegevensherkomst bij naar de kenmerken die in de training worden gebruikt. Wanneer de trainingsset RequestSource kenmerken bevat, worden de RequestSource kolommen als vereiste invoer toegevoegd aan de modelhandtekening van MLflow. Dit zorgt ervoor dat het API-schema van het servereindpunt overeenkomt met de velden die aanroepers moeten opgeven tijdens inferentietijd. Zie Modellen trainen met functietabellen voor meer informatie.

Voor FeatureViewSource, registreer de afgeleide feature en de upstream-features voordat een model wordt gelogd. Invoergegevens van het verzoek die nodig zijn voor transitieve afhankelijkheden, zijn ook nodig tijdens inferentie. Zie Custom UDF-afhankelijkheden voor modelpakketvereisten.

score_batch()

Batchinference uitvoeren met automatische kenmerkopzoeking

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

score_batch maakt gebruik van de functiemetagegevens die zijn opgeslagen met het model om automatisch tijdspecifieke correcte features voor inference te berekenen, waardoor consistentie met de training wordt gewaarborgd. Zie Modellen trainen met functietabellen voor meer informatie.

Voorbeeldwerkstroom

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

Training met RequestSource-functies

Wanneer uw model gegevens vereist die tijdens inferentie worden verstrekt (zoals transactiegegevens van een API-aanroep), gebruikt u naast RequestSource-functies ook tabelgebaseerde functies. Tijdens de training worden RequestSource kolommen geëxtraheerd uit het gelabelde 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",
    )

Transformeer verzoekwaarden met CustomUDF

Om verzoekgegevens te transformeren, gebruik CustomUDF in plaats van ColumnSelection. Dit voorbeeld gebruikt de transaction log-transform functie:

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

Het resultaat omvat het origineel transaction_amount en label kolommen, plus log_transaction_amount. De UDF leest transaction_amount uit elke rij. Verzoekkolommen die nodig zijn voor berekening kunnen niet worden vermeld in exclude_columns.

Train met FeatureViewSource-functies

FeatureViewSource laat een CustomUDF andere feature-output consumeren, inclusief afgeleide features. Geef de uitgangen door die je wilt create_training_set. Je hoeft hun intermediaire afhankelijkheden niet op te sommen.

Voor elke invoerrij lost Azure Databricks de volledige afhankelijkheidsgrafiek op:

1. Berekent door tabellen ondersteunde upstream-kenmerken op basis van hun entiteitssleutels, tijdstempels en vensterdefinities. Het gebruikt compatibele offline materialisaties wanneer beschikbaar.
2. Leest de benodigde verzoekkolommen uit de invoer DataFrame en evalueert de op verzoeken gebaseerde functies.
3. Evalueert afgeleide features in afhankelijkheidsvolgorde, zodat elke UDF zijn upstream resultaten ontvangt.

De afgeleide functie introduceert geen nieuw tijdvenster of tijdpunt-in-tijd-opzoeking. De upstream-functies behouden hun eigen tijdsemantiek. Het DataFrame moet de entiteit-, tijdstempel- en verzoekkolommen bevatten die door die upstreams nodig zijn, zelfs wanneer alleen de laatste afgeleide functie wordt aangevraagd.

Gebruik bijvoorbeeld de geregistreerde margefunctie, die cost_sum_7d en revenue_sum_7d combineert:

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

Het resultaat bevat label en margin. De omzet- en kostenkenmerken worden berekend, maar worden niet als extra kolommen teruggegeven. Om omzet in de trainingsdata op te nemen, haal deze op met revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") en geef deze door met features=[margin, revenue]. Dit geldt ook voor meerlagige ketens: het aanvragen van de laatste feature geeft niet elke tussenliggende output terug.

Je kunt op aanvragen gebaseerde, op tabellen gebaseerde en afgeleide features combineren in dezelfde features-lijst. Om hun waarden binnen één UDF te combineren, vertegenwoordig je de verzoekwaarden als features en verwijs je naar deze naast tabel-ondersteunde features in een FeatureViewSource.

Voor experimenteerdoeleinden construeer lokale Feature-objecten, inclusief hun bovenliggende graaf, zonder deze te registreren. Gebruik create_training_set, optioneel met label=None, om de resultaten te inspecteren. compute_features ondersteunt RequestSource of FeatureViewSource niet.

Aangepaste UDF-afhankelijkheden

Voor offline berekening declareer je Python-pakketten in de ENVIRONMENT Unity Catalog UDF-clausule. Het installeren van een pakket alleen in het notebook installeert het niet in de UDF-omgeving.

Voor model serving geef je ook expliciet de benodigde pakketten door aan log_model. Noch de ENVIRONMENT van de UDF, noch een benoemde functiespecificatie met Feature Views leveren deze modelvereisten automatisch. Voeg zowel de afhankelijkheden toe die nodig zijn voor upstream UDFs als de uitvoer van de gevraagde features.

Nadat u een scikit-learn-model hebt getraind op transaction_training_set.load_df(), registreert u het met dezelfde trainingsset. Voeg NumPy en een compatibel zoekpakket toe:

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

Serving endpoints zouden automatisch versie 1.15.0 of later moeten oppikken databricks-feature-lookup , die on-demand Unity Catalog UDF-berekening voor RequestSource-functies ondersteunt. Houd de versies van je UDF-pakketten consistent tussen offline en serverend omgevingen om verschillen in berekende waarden te voorkomen. Voor Feature Serving eindpunten zonder model, declareer in plaats daarvan pakketten op op create_feature_spec . Zie Add Python dependencies.

Training met streaming-functies

Wanneer u een stream definieert, beheert Databricks een opnamepijplijn waarmee streamgegevens naar een Delta-tabel worden geschreven. create_training_set leest uit deze opnametabel en voert punt-in-time joins uit op uw gelabelde DataFrame, net zoals batchfuncties van een DeltaTableSource. Zie Ingestion and backfill voor meer informatie over de configuratie van gegevensinname, backfill en deduplicatie.

Voorbeeld

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

Batch- en streamingfuncties combineren

Batch- en streamingfuncties kunnen samen worden gebruikt in dezelfde trainingsset en hetzelfde model. Bij het serveren worden batchkenmerken opgehaald uit offline- of online stores, en streamingkenmerken worden opgehaald uit online stores.

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

Het model dat is geregistreerd met log_model() voert functiezoekacties uit de online winkel uit en configureert de modelhandtekening voor beide brontypen.

Wat het ongefilterde model bereikt tijdens de bediening

De wrapper van het Feature Store-model filtert kolommen voordat deze worden doorgegeven aan het onbewerkte model: