Trenowanie modeli za pomocą widoków funkcji

Widoki cech umożliwiają trenowanie modeli z wykorzystaniem prawidłowego w danym momencie obliczania cech oraz automatycznego wyszukiwania cech na etapie wnioskowania. Aby uzyskać informacje na temat definiowania widoków funkcji, zobacz Widoki funkcji.

Requirements

• Funkcje muszą być tworzone jako widoki funkcji. Zobacz Widoki funkcji.
• Wymagania dotyczące CustomUDF i FeatureViewSource opisano w dokumentacji referencyjnej interfejsu API Feature Views.

Metody interfejsu API

create_training_set()

Po utworzeniu widoków funkcji następnym krokiem jest utworzenie danych szkoleniowych dla modelu. W tym celu przekaż oznaczony zestaw danych do create_training_set, który automatycznie zapewnia dokładne obliczanie wartości każdej cechy w punkcie w czasie.

Przykład:

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

Wywołaj metodę TrainingSet.load_df, aby połączyć oryginalne dane treningowe z dynamicznie obliczonymi cechami w punkcie czasowym.

Argument df musi spełniać następujące wymagania:

• Musi zawierać wszystkie kolumny jednostek, do których odwołują się definicje funkcji.
• Musi zawierać kolumnę szeregów czasowych, do której odwołują się definicje cech.
• Musi zawierać wszystkie kolumny zadeklarowane w dowolnym RequestSource schemacie. Typy są weryfikowane względem zadeklarowanego schematu. Niezgodności powodują błąd (bez niejawnego rzutowania).
• Powinna zawierać kolumny etykiet.
• Zestaw nazw kolumn jednostki, nazw kolumn czasowników i nazw kolumn funkcji żądania musi być globalnie unikatowy we wszystkich źródłach.

Poprawność punktu w czasie: W przypadku agregacji i ColumnSelection funkcji wspieranych przez źródło tabeli funkcje są obliczane przy użyciu tylko danych źródłowych dostępnych przed sygnaturą czasową każdego wiersza, aby zapobiec wyciekowi danych w przyszłości do trenowania modelu. W przypadku RequestSource funkcji wartość jest pobierana bezpośrednio z oznaczonego wiersza ramki danych.

log_model()

Użyj biblioteki MLflow, aby zarejestrować model z metadanymi funkcji na potrzeby śledzenia pochodzenia i automatycznego wyszukiwania funkcji podczas wnioskowania:

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
)

Parametr flavor określa moduł wersji modelu MLflow do użycia, taki jak mlflow.sklearn lub mlflow.xgboost.

Modele rejestrowane przy użyciu TrainingSet automatycznie śledzą pochodzenie cech używanych w trenowaniu. Gdy zestaw trenowania zawiera RequestSource funkcje, RequestSource kolumny są dodawane do podpisu modelu MLflow jako wymagane dane wejściowe. Gwarantuje to, że schemat API obsługującego punkt końcowy odzwierciedla pola, które muszą być dostarczone w momencie wnioskowania. Aby uzyskać szczegółowe informacje, zobacz Trenowanie modeli za pomocą tabel funkcji.

Dla FeatureViewSource zarejestruj cechę pochodną oraz jej cechy źródłowe przed zarejestrowaniem modelu. Wymagane są również żądania wejść wymagane przez zależności przechodnie w czasie wnioskowania. Zobacz Niestandardowe zależności UDF dla wymagań pakietów modeli.

score_batch()

Wykonywanie wnioskowania wsadowego za pomocą automatycznego wyszukiwania funkcji:

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

score_batch używa metadanych funkcji przechowywanych w modelu, aby automatycznie obliczać prawidłowe funkcje punktu w czasie na potrzeby wnioskowania, zapewniając spójność trenowania. Aby uzyskać szczegółowe informacje, zobacz Trenowanie modeli za pomocą tabel funkcji.

Przykładowy przepływ pracy

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

Trenowanie za pomocą funkcji RequestSource

Jeśli Twój model wymaga danych, które są udostępniane podczas wnioskowania (na przykład szczegóły transakcji z wywołania interfejsu API), użyj cech RequestSource razem z cechami opartymi na tabelach. Podczas treningu kolumny RequestSource są wyodrębniane z oznaczonej ramki danych.

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

Transformacja wartości żądań za pomocą CustomUDF

Aby przekształcić dane żądań, użyj CustomUDF zamiast .ColumnSelection Ten przykład wykorzystuje funkcję transformacji logu transakcji:

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

Wynik obejmuje oryginalne kolumny transaction_amount i label, oraz log_transaction_amount. UDF odczytuje transaction_amount z każdego wiersza. Kolumny żądania wymagane do obliczeń nie mogą być wymienione w exclude_columns.

Trenowanie przy użyciu funkcji FeatureViewSource

FeatureViewSource pozwala elementowi CustomUDF wykorzystywać wyniki innych cech, w tym cech pochodnych. Przekaż wyjścia, które chcesz przekazać do create_training_set. Nie musisz wymieniać ich zależności pośrednich.

Dla każdego wiersza wejściowego Azure Databricks rozwiązuje pełny graf zależności:

1. Oblicza funkcje poparte tabelą w górze, wykorzystując ich klucze encji, znaczniki czasu i definicje okien. Wykorzystuje kompatybilne materializacje offline, gdy są dostępne.
2. Odczytuje z wejściowej ramki danych kolumny wymagane przez żądanie i oblicza cechy oparte na żądaniach.
3. Ocena cech wyprowadzonych odbywa się zgodnie z kolejnością zależności, dzięki czemu każdy UDF otrzymuje wyniki z poprzedzających zależności.

Pochodna cecha nie wprowadza kolejnego okna czasowego ani wyszukiwania dla określonego punktu w czasie. Jego funkcje wejściowe zachowują własną semantykę czasową. DataFrame musi zawierać entycję, znaczniki czasu i kolumny żądań potrzebne tym upstreamom, nawet gdy żądana jest tylko ostateczna funkcja pochodna.

Na przykład użyj zarejestrowanej funkcji marginesów, która łączy revenue_sum_7d i 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()

Wynik zawiera label i margin. Parametry przychodów i kosztów są obliczane, ale nie zwracane jako dodatkowe kolumny. Aby uwzględnić przychody w danych treningowych, pobierz je za pomocą revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") i przekaż features=[margin, revenue]. Dotyczy to również łańcuchów wielopoziomowych: żądanie wyniku końcowego nie powoduje zwrócenia wszystkich wyników pośrednich.

Możesz łączyć cechy oparte na żądaniach, oparte na tabelach i cechy pochodne na tej samej liście features. Aby połączyć ich wartości w jednym UDF, reprezentuj wartości żądań jako cechy i odwołuj się do nich obok cech opartych na tabelach w .FeatureViewSource

Na potrzeby eksperymentów twórz lokalne obiekty Feature, wraz z ich grafem źródłowym, bez rejestrowania ich. Użyj create_training_set, opcjonalnie z label=None, aby sprawdzić wyniki. compute_features nie wspiera RequestSource ani FeatureViewSource.

Niestandardowe zależności UDF

Na potrzeby obliczeń offline należy zadeklarować pakiety Python w klauzuli ENVIRONMENT funkcji UDF Unity Catalog. Samo zainstalowanie pakietu w notebooku nie oznacza jego instalacji w środowisku UDF.

Dla obsługiwania modelu należy również przesłać wymagane pakiety eksplicytowo do log_model. Ani element ENVIRONMENT UDF, ani nazwana specyfikacja cech zawierająca widoki cech nie zapewniają automatycznie tych wymagań modelu. Uwzględnij zależności wymagane przez nadrzędne funkcje UDF, a także żądane dane wyjściowe cech.

Po wytrenowaniu modelu scikit-learn na transaction_training_set.load_df() zarejestruj go, używając tego samego zbioru treningowego. Dołącz NumPy oraz kompatybilny pakiet wyszukiwania:

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

Punkty końcowe obsługujące powinny automatycznie używać wersji 1.15.0 lub nowszej databricks-feature-lookup, która obsługuje obliczenia UDF w Unity Catalog na żądanie dla cech RequestSource. Utrzymuj wersje pakietów UDF spójne między środowiskiem offline a obsługującym, aby uniknąć różnic w obliczanych wartościach. W przypadku punktów końcowych Feature Serving bez modelu zadeklaruj pakiety w elemencie create_feature_spec. Zobacz Dodaj zależności w Python.

Trenowanie za pomocą funkcji przesyłania strumieniowego

Gdy definiujesz Stream, usługa Databricks zarządza potokiem pozyskiwania danych, który zapisuje dane strumienia do tabeli Delta. create_training_set odczytuje dane z tej tabeli ingestii i wykonuje złączenia point-in-time z oznaczoną ramką danych, podobnie jak cechy wsadowe z DeltaTableSource. Szczegóły dotyczące konfiguracji pobierania, uzupełniania i deduplikacji można znaleźć w artykule Pobieranie i uzupełnianie.

Przykład

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

Mieszanie funkcji wsadowych i przesyłanych strumieniowo

Funkcje usługi Batch i przesyłania strumieniowego mogą być używane razem w tym samym zestawie trenowania i modelu. W czasie obsługi funkcje wsadowe są sprawdzane z sklepów w trybie offline lub online, a funkcje przesyłania strumieniowego są przeglądane ze sklepów online.

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

Model zarejestrowany przy użyciu log_model() wykonuje wyszukiwanie funkcji ze sklepu online i konfiguruje podpis modelu dla obu typów źródłowych.

Co dociera do surowego modelu w momencie obsługi

Osłona modelu magazynu cech filtruje kolumny przed przekazaniem ich do surowego modelu.