功能视图

Feature Views 是 Databricks Feature Store 中定义和计算特征的推荐方式。可以使用各种源（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 参考。

Requirements

无服务器计算，或运行 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 support CronSchedule or TableTrigger, 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(),
)

示例笔记本

功能视图快速入门笔记本

获取笔记本

批处理功能

特征视图原生处理大多数单阶段批量特征。在聚合之前，你可以用 filter_condition 过滤源行，并用 transformation_sql 重塑这些行，这支持重命名、类型转换、算术以及其他按行的 SQL 表达式。然后你在滚动、翻滚或滑动窗口上应用支持的聚合函数之一。对于Delta表源，Databricks在具体化过程中以增量方式处理新的源数据。

对于功能视图不支持的步骤，可以使用 Spark 声明式管道：跨表的连接、多步聚合以及不在支持列表中的聚合函数。将流水线输出写入 Delta 表，然后将其作为 ColumnSelection 特征连接到 Feature Store，或者将其作为聚合特征的源。

流式功能

当功能值需要持续更新而非批量更新时，使用流式功能。流式和批处理功能使用相同的 Feature 构造函数、聚合函数以及训练和服务工作流程。

流媒体功能不会出现在离线商店。对于训练和批次推断，Databricks从源计算特征值。

流媒体功能有以下要求：

你必须提供一个 online_config。流媒体功能不支持 offline_config。
你不能在一次 materialize_features 通话中同时进行流媒体和批量功能。为每种触发器类型分别调用一次。
你只能将 transformation_sql 与 StreamSource 一起使用，不能与 DeltaTableSource 一起使用。
流式实体化仅处理管道启动后到达的记录，不会回填历史记录。滚动窗口聚合只有在第一个完整数据窗口到达后才返回完整结果。

选择流媒体功能源

根据你的新鲜度需求和现有的摄入方式选择来源：

当亚秒级新鲜度是首要考虑时，使用 StreamSource。 StreamSource 这些功能为 P99 实现了 200 毫秒的端到端延迟。先设置 Stream，然后使用StreamSource引用它。流式数据源支持使用 Kafka 作为输入源，并自动维护一个摄取 Delta 表，作为该数据的历史副本用于训练。
如果你已经有导入 Delta 表的低延迟摄取路径，请使用 DeltaTableSource。预计数据新鲜度在几十秒左右。

使用StreamSource定义流媒体功能

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

使用 DeltaTableSource 定义流媒体功能

要将定义在 DeltaTableSource 上的特征具体化为流式特征，请将 StreamingMode 作为触发器传递给 materialize_features。该特征定义使用与由 DeltaTableSource 支持的批处理特征相同的 API。 Delta 表源支持聚合和列选择功能。

源 Delta 表必须通过设置delta.enableChangeDataFeed=true启用更改数据馈送。

以下示例定义并实现了带有Delta表源的聚合特征。

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    DeltaTableSource,
    AggregationFunction,
    OnlineStoreConfig,
    Sum,
    RollingWindow,
    StreamingMode,
)
from datetime import timedelta

client = FeatureEngineeringClient()

source = DeltaTableSource(
    catalog_name="my_catalog",
    schema_name="my_schema",
    table_name="transactions",
)

feature = client.create_feature(
    catalog_name="my_catalog",
    schema_name="my_schema",
    name="user_purchase_sum",
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(
        operator=Sum(input="amount"),
        time_window=RollingWindow(window_duration=timedelta(hours=1)),
    ),
)

online_config = OnlineStoreConfig(
    catalog_name="my_catalog",
    schema_name="my_schema",
    table_name_prefix="streaming_features",
    online_store_name="my_online_store",
)

client.materialize_features(
    features=[feature],
    online_config=online_config,
    trigger=StreamingMode(),
)

流媒体源的滤波条件

在聚合前用于 filter_condition 过滤行，以筛选 a StreamSource 或 DeltaTableSource。

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

变换流源

transformation_sql在聚合或选择列之前应用 Spark SQL 投影，使用带点前缀的方式引用 key 和 value 结构体。它只支持按行表达式。有关支持和不支持操作的完整列表，请参见支持 transformation_sql 表达式。

设置 transformation_sql 时，还必须提供 dataframe_schema：投影输出的 Spark StructType JSON。你也可以不必手动编写此架构，而是通过对 Stream 的摄入表运行相同的投影来推导出它，其中包含 key 和 value 结构体。通过指定 LIMIT 0，该 SELECT 语句返回该模式而无需读取任何数据：

from databricks.feature_engineering.entities import StreamSource

transformation_sql = (
    "value.amount * value.conversion_rate AS converted_amount, "
    "struct(value.user_id AS user_id, value.event_time AS time) AS event"
)

ingestion_table = "my_catalog.my_schema.events_ingestion"
dataframe_schema = spark.sql(
    f"SELECT {transformation_sql} FROM {ingestion_table} LIMIT 0"
).schema.json()

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

来自流媒体来源的专栏选择

ColumnSelection 特征会针对每个实体键直接传递源中的最新值，不经过聚合。在训练时，特征值尊重时间点的准确性。

列选择功能没有TTL。要从在线商店移除选中的值，源必须对所选列输出空值。

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

从StreamSource访问嵌套字段

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

流式处理功能的时间窗口

流式功能支持用于聚合的 SawtoothWindow 和 RollingWindow。两者都能实时更新特征值；二者的区别在于各自能够高效计算的回溯窗口长度不同。 TumblingWindow 和 SlidingWindow 用于在固定的历史时间间隔内进行批量计算，不能与流式数据源一起使用。

对于较短的回溯窗口，使用 RollingWindow。 RollingWindow 功能根据数据流计算特征数据，确保数据进出窗口时的精度达到亚秒级。特征聚合值是实时从流本身构建并存储在本地的。实体化必须先运行整整 window_duration，窗口才会被完全填满。 RollingWindow无法像 sawtooth 那样良好地扩展到长达数月或数年的窗口长度。
对于较长的回溯窗口，请使用 SawtoothWindow。 SawtoothWindow 特征会读取 Stream 中最近两天的数据，并使用 Stream 的摄取表中的批量数据补充这一最新部分。这使得 Databricks 能够快速初始化长窗口流功能，前提是摄取表中已经包含至少覆盖整个窗口时长的数据。一个SawtoothWindow在开始实例化后大约两天即可提供服务。它的window_duration必须大于 2 天（强制执行的最小值），并且支持Sum、Avg、Min、StddevPop、Count、First、StddevSamp、Max、Last、VarPop和VarSamp聚合函数。

Databricks 建议对于超过 7 天的特征持续时间使用 SawtoothWindow。对于超过两天且不超过七天的窗口，请确定您更看重 RollingWindow 的固定长度精度，还是 SawtoothWindow 更快达到生产就绪状态的速度。

以下示例基于 StreamSource 定义一个采用锯齿窗口、时长 30 天的流式处理特征。

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

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

user_spend_30d = Feature(
    name="user_spend_30d",
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(
        operator=Sum(input="value.amount"),
        time_window=SawtoothWindow(window_duration=timedelta(days=30)),
    ),
)

流式处理功能示例笔记本

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

获取笔记本

模型训练和推理

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

特征具体化

定义功能后，可以将这些功能具体化为脱机或在线商店，以便在培训和服务工作流中高效重复使用。对于以 Delta 表源为后端的功能，Databricks 使用变更数据馈送 (CDF) 对源数据进行增量处理，从而保持持续物化的高效性。具体化功能后，可以使用 CPU 模型服务为模型提供服务。有关详细信息，请参阅具体化功能视图。

最佳做法

功能命名

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

时间范围

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

性能

在单个 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 或更高版本的经典计算群集来规避此问题。

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

上次更新时间为 2026-10-28