Обучение моделей с помощью представлений функций

Important

Эта функция доступна в общедоступной предварительной версии. Администраторы рабочей области могут управлять доступом к этой функции на странице "Предварительные версии ". См. статью "Управление предварительными версиями Azure Databricks".

Feature Views позволяют обучать модели с корректным на заданный момент времени вычислением признаков и автоматическим поиском признаков на этапе инференса. Сведения об определении представлений функций см. в разделе "Представления компонентов".

Требования

Компоненты должны быть созданы в виде представлений компонентов. См. Представления возможностей.

Методы API

`create_training_set()`

После создания представлений компонентов необходимо создать обучающие данные для модели. Для этого передайте помеченный набор create_training_setданных, который автоматически обеспечивает точное вычисление каждого значения функции на определенный момент времени.

Рассмотрим пример.

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

Вызовите TrainingSet.load_df, чтобы присоединить исходные обучающие данные с функциями, динамически вычисляемыми на момент времени.

Аргумент df должен соответствовать следующим требованиям:

Должен содержать все указанные столбцы сущностей, на которые ссылаются определения функций.
Должен содержать столбец временного ряда, на который ссылаются определения признаков.
Должен содержать все столбцы, объявленные в любой RequestSource схеме. Типы проверяются на основе объявленной схемы. Несовпадения типов приводят к ошибке (неявное приведение типов не выполняется).
Должен содержать столбец или столбцы меток.
Набор имен столбцов сущностей, имена столбцов временных рядов и имена столбцов характеристик запросов должны быть глобально уникальными во всех источниках.

Правильность на определенный момент времени: Для агрегирования и ColumnSelection функций, поддерживаемых источником таблицы, функции вычисляются с помощью только исходных данных, доступных до метки времени каждой строки, чтобы предотвратить будущие утечки данных в обучение модели. Для RequestSource функций значение берется непосредственно из строки помеченного кадра данных.

`log_model()`

Используйте MLflow для регистрации модели с метаданными признаков для отслеживания их родословной и автоматического поиска признаков при инференсе.

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
)

Параметр flavor задает модуль варианта модели MLflow, используемый, например, mlflow.sklearn или mlflow.xgboost.

Модели, зарегистрированные с TrainingSet, автоматически отслеживают происхождение признаков, используемых в обучении. Если набор обучения включает RequestSource функции, RequestSource столбцы добавляются в подпись модели MLflow в качестве необходимых входных данных. Это обеспечивает то, что схема API конечной точки обслуживания отражает поля, которые вызывающие стороны должны предоставить во время вывода предсказаний. Дополнительные сведения см. в разделе "Обучение моделей с таблицами компонентов".

`score_batch()`

Выполнение пакетного вывода с помощью автоматического поиска функций:

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

score_batch использует метаданные признаков, хранящиеся в модели, для автоматического вычисления временно-корректных признаков для вывода, обеспечивая согласованность с обучением. Дополнительные сведения см. в разделе "Обучение моделей с таблицами компонентов".

Пример рабочего процесса

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

Обучение с помощью функций RequestSource

Если для модели требуются данные, предоставляемые во время вывода (например, сведения о транзакциях из вызова API), используйте RequestSource функции вместе с функциями, поддерживаемыми таблицами. Во время обучения столбцы 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",
    )

Обучение с помощью функций потоковой передачи

При определении Stream Databricks управляет конвейером приема, который записывает потоковые данные в таблицу Delta. create_training_set считывает данные из этой таблицы загрузки и выполняет соединения по состоянию на определенный момент времени с размеченным DataFrame, так же, как и пакетные признаки из DeltaTableSource. Подробные сведения о конфигурации приема данных, дозаполнении и дедупликации см. в разделе Прием данных и дозаполнение.

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

Сочетание функций пакетной и потоковой передачи

Функции пакетной и потоковой передачи можно использовать вместе в одном наборе обучения и модели. Во время обработки запросов пакетные признаки извлекаются из офлайн- или онлайн-хранилищ, а потоковые признаки — из онлайн-хранилищ.

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

Модель, журналируемая с помощью log_model(), выполняет получение признаков из онлайн-хранилища и настраивает сигнатуру модели для обоих типов источников данных.

Что достигает необработанной модели во время обслуживания

Оболочка модели Feature Store фильтрует столбцы перед передачей в сырую модель:

Тип столбца	Достигает внутренней модели?
Явные выходные характеристики (`ColumnSelection`, агрегирование)	Да
`RequestSource` столбцы, объявленные как функции	Да
Столбцы сущностей (ключи поиска)	Нет (если явно объявлено как функция)
Столбцы Timeseries	Нет (если явно объявлено как функция)

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-07-13