功能视图

Important

此功能目前以公共预览版提供。工作区管理员可以从预览页控制对此功能的访问。请参阅 Manage Azure Databricks 预览版。

通过功能视图，可以从数据源中定义和计算特征。可以使用各种源（Delta 表、Kafka Stream 和请求时间数据）和计算（时间窗口聚合、简单列选择等）定义特征。本指南介绍以下工作流：

功能开发 工作流
- 用于 create_feature 定义可在模型训练和服务工作流中使用的 Unity Catalog 功能对象。
- 或者在本地创建 Feature 对象，并稍后使用 register_feature 将它们持久化到 Unity Catalog。在本地构建的特性可以在注册之前与它一起使用 create_training_set 。
模型训练 工作流
- 使用 create_training_set 来计算机器学习中的时间点聚合特征。有关使用功能视图训练的详细文档，请参阅使用功能视图训练模型。
特征具体化和服务 工作流
- 在定义了create_feature功能或使用get_feature检索到该功能之后，可以使用materialize_features将该功能或功能集物化到离线存储，以便高效重复使用，或者将功能物化到在线商店以进行在线服务。
- 使用 create_training_set 和物化视图准备一个脱机批量训练数据集。

有关 API 的详细信息，请参阅功能视图 API 参考。

要求

无服务器计算，或运行 Databricks Runtime 17.0 ML 或更高版本的经典计算群集。
必须安装自定义 Python 包。每次运行笔记本时，运行以下代码行：
```
%pip install databricks-feature-engineering>=0.16.0
dbutils.library.restartPython()
```

快速入门示例

有关可运行的快速入门笔记本，请参阅示例笔记本。

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    CronSchedule, DeltaTableSource, Feature, AggregationFunction,
    Sum, Avg, ColumnSelection, TableTrigger,
    TumblingWindow, SlidingWindow,
    OfflineStoreConfig, OnlineStoreConfig,
)
from datetime import timedelta

CATALOG_NAME = "main"
SCHEMA_NAME = "feature_store"
TABLE_NAME = "transactions"

# 1. Create data source
source = DeltaTableSource(
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    table_name=TABLE_NAME,
)

# 2. Define features locally (no catalog/schema needed yet)
avg_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), TumblingWindow(window_duration=timedelta(days=30))),
    name="avg_transaction_30d",
)

sum_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), SlidingWindow(window_duration=timedelta(days=7), slide_duration=timedelta(days=1))),
    # name auto-generated: "amount_sum_sliding_7d_1d"
)

fe = FeatureEngineeringClient()

# 3. Explore features with compute_features
feature_df = fe.compute_features(features=[avg_feature, sum_feature])
feature_df.display()

# 4. Create training set using local features
# `labeled_df` should have columns "user_id", "transaction_time", and "target".
training_set = fe.create_training_set(
    df=labeled_df,
    features=[avg_feature, sum_feature],
    label="target",
)
training_set.load_df().display()

# 5. Register features in Unity Catalog
avg_feature = fe.register_feature(
    feature=avg_feature,
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
)
sum_feature = fe.register_feature(
    feature=sum_feature,
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
)

# 6. Or use create_feature for a one-step define-and-register workflow
latest_amount = fe.create_feature(
    source=source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    name="latest_amount",
)

# 7. Train model
with mlflow.start_run():
    training_df = training_set.load_df()

    # training code

    fe.log_model(
        model=model,
        artifact_path="recommendation_model",
        flavor=mlflow.sklearn,
        training_set=training_set,
        registered_model_name=f"{CATALOG_NAME}.{SCHEMA_NAME}.recommendation_model",
    )

# 8. (Optional) Materialize features for serving
# Features must be registered in UC before calling materialize_features
online_config = OnlineStoreConfig(
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    table_name_prefix="customer_features_serving",
    online_store_name="customer_features_store",
)

# Aggregation features use CronSchedule and support both offline and online configs
fe.materialize_features(
    features=[avg_feature, sum_feature],
    offline_config=OfflineStoreConfig(
        catalog_name=CATALOG_NAME,
        schema_name=SCHEMA_NAME,
        table_name_prefix="customer_features",
    ),
    online_config=online_config,
    trigger=CronSchedule(
        quartz_cron_expression="0 0 * * * ?",  # Hourly
        timezone_id="UTC",
    ),
)

# ColumnSelection features use TableTrigger and only support online config
fe.materialize_features(
    features=[latest_amount],
    online_config=online_config,
    trigger=TableTrigger(),
)

示例笔记本

功能视图快速入门笔记本

获取笔记本

流式功能

除了 Delta 表中的批处理功能外，还可以为实时用例定义来自流式处理源的功能。流式处理功能使用与批处理功能相同的功能类（相同的 Feature 构造函数、相同的聚合函数、相同的训练和服务工作流），因此从批处理升级到实时需要最少的代码更改。具体化后，流式处理功能将次秒端到端新鲜度（p99 延迟 200 毫秒）直接传送到提供终结点的模型。

若要使用流式功能，请先设置 Stream，然后使用 StreamSource 引用它。流源支持以 Kafka 作为输入，并自动维护摄取（Delta）表，将其作为用于训练的数据的历史副本。

定义流式功能

StreamSource 通过其三部分名称（catalog.schema.stream_name）引用 Stream。 Stream 不是 Unity 目录安全对象，但它的范围限定为 Unity 目录架构，访问受 Stream 引入表的约束。实体、时间顺序和函数定义中的列引用必须带有 value. 前缀或 key. 指示要读取的 Kafka 消息的哪一部分。使用点表示法（例如， value.user.address.city）支持嵌套字段。

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    StreamSource,
    Feature,
    AggregationFunction,
    Sum,
    RollingWindow,
)
from datetime import timedelta

client = FeatureEngineeringClient()

stream_source = StreamSource(
    full_name="my_catalog.my_schema.my_stream",
)

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

StreamSource 上的筛选条件

使用 filter_condition 在聚合之前从流中筛选行，就像在 DeltaTableSource 上一样。

stream_source = StreamSource(
    full_name="my_catalog.my_schema.my_stream",
    filter_condition="value.event_type = 'purchase'",
)

流中的列选择

ColumnSelection 功能适用于流式处理源。所选列表示每个实体在该流中的最新的值，同时保证时间点准确性。

from databricks.feature_engineering.entities import ColumnSelection

passenger_count = Feature(
    name="passenger_count",
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=ColumnSelection(column="value.passenger_count"),
)

访问嵌套字段

可以使用点表示法（例如， value.nested_field.amount）访问嵌套 JSON 字段。在提供服务时，请求负载和响应使用叶节点名称（例如，使用 amount 而不是 value.amount）。叶节点名称在模型或特征规范中的所有实体、时间序列和功能输出列中必须是唯一的，因为服务终结点使用叶名称来路由值。

流式处理功能的时间窗口

流式处理功能仅支持使用 RollingWindow 进行聚合。滚动窗口会基于最新数据持续重新计算，这与流式数据源的实时特性相契合。 TumblingWindow 和 SlidingWindow 专为针对固定历史时间间隔的批量计算而设计。

流式处理功能示例笔记本

流式处理功能视图快速入门笔记本

获取笔记本

模型训练和推理

若要使用特征视图训练模型并运行批量推理，包括log_model()score_batch()，以及create_training_set()，请参阅使用特征视图训练模型。

特征具体化

定义功能后，可以将这些功能具体化为脱机或在线商店，以便在培训和服务工作流中高效重复使用。具体化功能后，可以使用 CPU 模型服务为模型提供服务。有关详细信息，请参阅具体化功能视图。

最佳做法

功能命名

对业务关键功能使用描述性名称。
跨团队遵循一致的命名约定。
开始开发功能时，请使用自动生成的名称。

时间范围

将窗口边界与业务周期（每日、每周）对齐。
较短的窗口可以捕捉到最近的趋势，但可能会有较多噪声。较长的窗口会产生更稳定的功能分布，但可能会错过最近的行为转变。根据用例中的信号变化速度来选择。例如，7 天窗口可平滑每日波动并生成一致的模型输入，而 1 小时窗口对行为更改做出快速反应，但可能会引入降低模型性能的方差。如果模型的准确性在分布移动时下降，请使用更长的窗口来稳定输入。
翻转和滑动窗口比滚动（连续）窗口更具可伸缩性。从滑动窗口开始，适用于大多数用例。

Performance

在单个 materialize_features 调用中具体化来自同一数据源的功能，以最大程度地减少数据扫描。
对同一数据源上的功能使用相同的粒度（例如，所有 1 小时或全部 1 天幻灯片持续时间），以便在具体化期间更好地分组。

实体列与筛选条件

使用同一源表中的功能时，请使用此决策指南：

在需要不同聚合级别时，请使用 entity （在 create_feature）：

客户级功能 （每个客户一行）： entity=["customer_id"]
客户-商家功能 （每个客户多行）： entity=["customer_id", "merchant_id"]
不同的聚合级别可以共享相同的 DeltaTableSource：为每个功能定义指定不同的 entity 值

需要在同一聚合级别筛选行时，请使用 filter_condition （on DeltaTableSource）：

仅高价值交易： filter_condition="amount > 100" （仍按客户聚合）
仅包含已完成订单：filter_condition="status = 'completed'"（仍按每位客户进行聚合）

经验法则： 如果您的更改会导致每个实体值生成不同数量的行，请在功能定义中使用不同的entity值。如果只是筛选哪些行促成了相同的聚合，请对源使用 filter_condition 。

常见模式

客户分析

from databricks.feature_engineering.entities import AggregationFunction, Sum, Count, RollingWindow

fe = FeatureEngineeringClient()
features = [
    # Recency: Number of transactions in the last day
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Count(input="transaction_id"), RollingWindow(window_duration=timedelta(days=1)))),

    # Frequency: transaction count over the last 90 days
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Count(input="transaction_id"), RollingWindow(window_duration=timedelta(days=90)))),

    # Monetary: total spend in the last month
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=30)))),
]

走向分析

# Compare recent vs. historical behavior
fe = FeatureEngineeringClient()
recent_avg = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

historical_avg = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=7), delay=timedelta(days=7))),
)

季节性模式

# Same day of week, 4 weeks ago
fe = FeatureEngineeringClient()
weekly_pattern = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=1), delay=timedelta(weeks=4))),
)

局限性

实体和时间序列列的名称必须与在 API 中使用的 create_training_set 训练（已标记）数据集和特征定义之间匹配。
训练数据集中用作 label 列的列名不应存在于用于定义 Features 的源表中。
API 支持的函数列表（UDAF）是有限的。请参阅支持的函数。
实体列不能为类型 DATE 或类型 TIMESTAMP。
RequestSource仅支持在ScalarDataType（INTEGER、FLOAT、BOOLEAN、STRING、DOUBLE、LONG、TIMESTAMP、DATE、SHORT）中定义的标量数据类型。不支持复杂类型，如数组、映射和结构。
RequestSource 不支持聚合函数或时间窗口。只能使用 ColumnSelection 函数。
实体列名称集、时间序列列名称集和请求特征列名称必须在训练集或服务终端中的所有来源中具有全局唯一性。

score_batch 在无服务器计算上可能无法成功。可通过使用运行 Databricks Runtime 17.0 ML 或更高版本的经典计算群集来规避此问题。

有关具体化限制的详细信息，请参阅限制。

反馈

此页面是否有帮助？

Last updated on 2026-07-14