Trénování modelů pomocí zobrazení funkcí

Důležité

Tato funkce je ve verzi Public Preview. Správci pracovního prostoru můžou řídit přístup k této funkci ze stránky Previews . Viz Manage Azure Databricks preview.

Zobrazení funkcí umožňují trénovat modely se správnými výpočty funkcí k určitému bodu v čase a automatickým vyhledáváním funkcí při odvozování. Informace o definování zobrazení funkcí naleznete v tématu Zobrazení funkcí.

Requirements

Metody rozhraní API

create_training_set()

Po vytvoření zobrazení funkcí je dalším krokem vytvoření trénovacích dat pro váš model. Uděláte to tak, že předáte datovou sadu označenou jako datovou sadu create_training_set, která automaticky zajistí přesný výpočet jednotlivých hodnot funkce k určitému bodu v čase.

Například:

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

Zavolejte TrainingSet.load_df, abyste spojili původní trénovací data s dynamicky vypočítanými atributy k určitému bodu v čase.

Argument df musí splňovat následující požadavky:

  • Musí obsahovat všechny sloupce entit odkazované podle definic vlastností.
  • Musí obsahovat sloupec časové řady odkazovaný definicemi vlastností.
  • Musí obsahovat všechny sloupce deklarované v jakémkoli RequestSource schématu. Typy se ověřují proti deklarovanému schématu. Neshody vyvolávají chybu (bez implicitního přetypování).
  • Měly by obsahovat sloupce popisků.
  • Sada názvů sloupců entit, názvů sloupců časové řady a názvů sloupců funkcí dotazů musí být globálně jedinečná ve všech zdrojích.

Správnost k určitému bodu v čase: Pro agregaci a ColumnSelection funkce zálohované zdrojem tabulky se funkce počítají pouze pomocí zdrojových dat dostupných před časovým razítkem každého řádku, aby se zabránilo budoucímu úniku dat do trénování modelu. U RequestSource funkcí se hodnota přebírá přímo z označeného řádku datového rámce.

log_model()

Použití MLflow k protokolování modelu s metadaty funkcí pro sledování rodokmenu a automatické vyhledávání funkcí během odvozování:

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
)

Parametr flavor určuje modul příchuť modelu MLflow , který se má použít, například mlflow.sklearn nebo mlflow.xgboost.

Modely zaznamenané pomocí TrainingSet automaticky sledují vztah k funkcím používaným při trénování. Když trénovací sada obsahuje RequestSource atributy, sloupce RequestSource se přidají do podpisu modelu MLflow jako nutné vstupy. Tím se zajistí, že schéma rozhraní API obslužného koncového bodu odpovídá polím, která musí volající poskytnout v době inferencí. Podrobnosti najdete v tématu Trénování modelů s tabulkami vlastností.

score_batch()

Provedení dávkového odvozování pomocí automatického vyhledávání funkcí:

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

score_batch používá metadata funkcí uložená s modelem k automatickému výpočtu správných funkcí k určitému bodu v čase pro odvozování, což zajišťuje konzistenci s trénováním. Podrobnosti najdete v tématu Trénování modelů s tabulkami vlastností.

Ukázkový pracovní postup

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

Školení s funkcemi RequestSource

Pokud váš model vyžaduje data, která jsou k dispozici v době odvozování (například podrobnosti o transakcích z volání rozhraní API), používejte RequestSource funkce společně s funkcemi založenými na tabulce. Během trénování se sloupce extrahují z označeného datového rámce RequestSource.

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

Školení s funkcemi streamování

Když definujete Stream, Databricks spravuje kanál příjmu dat, který zapisuje streamová data do tabulky Delta. create_training_set čte z této tabulky příjmu dat a provádí spojení k určitému bodu v čase s označeným datovým rámcem stejně jako dávkové funkce z objektu DeltaTableSource. Podrobnosti o konfiguraci příjmu dat, obnovení a odstranění duplicitních dat najdete v tématu Příjem dat a obnovení.

Příklad

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

Kombinování dávkových a streamovacích funkcí

Funkce batch a streamování je možné používat společně ve stejné trénovací sadě a modelu. V době obsluhy se dávkové funkce vyhledávají z offline nebo online obchodů a funkce streamování se vyhledávají z online obchodů.

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

Model zaznamenaný pomocí log_model() vyhledává příznaky v online úložišti a nakonfiguruje signaturu modelu pro oba typy zdrojů.

Co se dostane k surovému modelu při nasazení

Obálka modelu úložiště funkcí filtruje sloupce před jejich předáním do nezpracovaného modelu:

Typ sloupce Dosáhne k vnitřnímu modelu?
Explicitní výstupy vlastností (ColumnSelection, agregace) Ano
RequestSource sloupce deklarované jako vlastnosti Ano
Sloupce entit (vyhledávací klíče) Ne (pokud není explicitně deklarováno jako funkce)
Sloupce časových řad Ne (pokud není explicitně deklarováno jako funkce)
  • Last updated on