使用功能视图训练模型

特征视图使您能够训练模型，并按时间点正确计算特征，同时在推理时自动查找特征。 有关定义功能视图的信息，请参阅 功能视图。

要求

• 必须作为功能视图来创建功能。 请参阅 功能视图。

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 功能，该值直接从标记的 DataFrame 行获取。

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

使用流式功能进行训练

定义 流时，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() 记录的模型会从在线存储中查找特征，并为这两种源类型配置模型签名。

在服务时输入到原始模型的内容

功能存储的模型包装器会在将列传递给原始模型之前，先过滤这些列。