特色視圖

Important

這項功能目前處於公開預覽版。工作區管理員可以從 「預覽 」頁面控制對此功能的存取。請參閱管理 Azure Databricks 預覽。

特徵檢視讓你能從資料來源定義並計算特徵。特徵可透過多種來源（Delta 表格、Kafka 串流及請求時資料）及計算方式（時間窗彙整、簡單欄位選擇等）來定義。本指南涵蓋以下工作流程：

功能開發工作流程
- 用於 create_feature 定義可用於模型訓練與服務工作流程的 Unity Catalog 功能物件。
- 或者，你可以在本地建構 Feature 物件，之後再用 register_feature 來持久化到 Unity 目錄。本地開發的功能可在註冊前使用 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 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(),
)

範例筆記本

功能檢視快速入門筆記本

拿筆記本

串流功能

當功能值必須持續更新而非批次排程時，請使用串流功能。串流與批次功能使用相同的 Feature 建構子、聚合函式，以及訓練與服務工作流程。

串流功能不會直接出現在離線商店。在訓練與批次推論中，Databricks 會從來源計算特徵值。

串流功能有以下要求：

你必須提供 online_config。串流功能不支援 offline_config。
你無法在一次 materialize_features 通話中同時結合串流和批次功能。針對每種觸發類型分別呼叫一次。
你只能將 transformation_sql 與 StreamSource 搭配使用，不能與 DeltaTableSource 搭配使用。
串流實體化只處理管線開始後抵達的紀錄，不會回填歷史紀錄。滾動視窗彙總只有在第一個完整資料視窗抵達後才會回傳完整結果。

選擇串流功能來源

根據你的新鮮度需求和現有的攝取環境選擇來源：

當亞秒級的新鮮度是首要考量時，請使用 StreamSource。 StreamSource 功能提供 P99 的端到端延遲 200 毫秒。先建立串流，然後使用 StreamSource 參照它。串流來源支援 Kafka 做為輸入來源，並自動維護擷取 Delta 資料表，作為供訓練使用的資料歷史副本。
當您已經有可低延遲寫入 Delta 資料表的資料擷取管道時，請使用 DeltaTableSource。新鮮度大約只有數十秒。
當你還沒有低延遲擷取管道時，請使用 Zerobus 來填入 DeltaTableSource。 Zerobus 的擷取時間約為數十秒，因此預期功能新鮮度低於分鐘。

使用 StreamSource 定義串流功能

A StreamSource 以其三部分名稱catalog.schema.stream_name（）來指稱串流。串流不是 Unity Catalog 的可保護物件，但其範圍受限於 Unity Catalog 結構描述，且其存取權限由串流的擷取資料表控管。實體、時間序列及函式定義中的欄位參考必須以或 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 來啟用變更資料摘要（CDF）。

以下範例定義並實現一個結合 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(),
)

使用由 Zerobus 填充的 Delta 表格

由 Zerobus 填充的 Delta 表格可作為串流功能來源。 Zerobus 不會自動設定 delta.enableChangeDataFeed=true 。你必須在目標 Delta 表上手動設定此屬性，才能將其作為串流功能來源使用。

串流來源的濾波條件

在彙整前，請用 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。你不必手寫這個結構，也可以透過對串流的攝取表執行相同的投影來推導，這樣就能揭露 key and 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）。葉節點名稱必須在模型或特徵規格內所有實體、時間序列及特徵名稱中唯一，因為服務端點會使用葉節點名稱來路由值。

串流功能的時段

串流功能具備支援 RollingWindow 與 SawtoothWindow 聚合功能。兩者都能即時讓特徵值維持最新；差別在於各自能高效計算的回溯長度不同。 TumblingWindow 和 SlidingWindow 設計用於固定的歷史時間區間批次運算，且無法與串流資料來源搭配使用。

使用 RollingWindow 來處理較短的回溯視窗。 RollingWindow 功能根據串流資料計算特徵資料，確保資料進出視窗時的精確度達到亞秒級。特徵聚合值是從串流本身即時建立並儲存在本地。具體化必須先執行完整一個 window_duration，視窗才會完全填滿。 RollingWindow 無法像 sawtooth 那樣良好地擴展到長達數月或數年的視窗長度。
使用 SawtoothWindow 用於較長的回溯視窗。 SawtoothWindow 功能會讀取串流中最近兩天的資料，並以串流擷取資料表中的批次資料補強這部分的最新資料。這讓 Databricks 能快速初始化長視窗串流功能，前提是擷取表已包含至少涵蓋整個視窗長度的資料。 A SawtoothWindow 在物質化開始約兩天後即可服務。其window_duration必須大於兩天（強制執行的最小值），並且支援 Sum、Avg、Max、First、Min、Count、VarPop、VarSamp、Last、StddevPop 和 StddevSamp 聚合函數。

Databricks 建議對於超過 7 天的特徵持續時間使用 SawtoothWindow。對於超過兩天且最長達七天的時間範圍，請決定您更重視 RollingWindow 的固定長度精確性，還是 SawtoothWindow 更快投入生產的就緒性。

Important

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 （開啟 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))),
)

Limitations

在 API 中使用 create_training_set 時，實體欄位與時間序列欄位的名稱必須在訓練（標記）資料集與特徵定義間相符。
訓練資料集中用作 label 欄位的欄位名稱不應存在於用於定義 Features 的來源資料表中。
API 支援 create_feature 有限的函式列表（UDAFs）。請參閱支援的功能。
實體欄位不能為類型 DATE 或 TIMESTAMP。
RequestSource僅支援定義於 ScalarDataType （INTEGER， FLOATBOOLEANSTRINGDOUBLELONGTIMESTAMPDATE） SHORT的純量資料型態。複雜型態如陣列、映射與結構體不被支援。
RequestSource 不支援聚合功能或時間窗。只能使用 ColumnSelection 函數。
實體欄位名稱、時間序列欄位名稱及請求功能欄位名稱必須在訓練集或服務端點的所有來源間全域唯一。

score_batch 在無伺服器運算上可能無法成功。你可以透過使用運行 Databricks Runtime 17.0 ML 或以上版本的經典運算叢集來繞過這個問題。

關於具體化的限制，請參見限制。

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-09-30