特色視圖

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 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 類別——相同的 Feature 建構器、相同的聚合函式、相同的訓練與服務工作流程——因此從批次升級到即時時所需的程式碼變更非常少。一旦實體化，串流功能便可直接將低於 1 秒的端對端資料新鮮度（p99 延遲為 200 毫秒）傳遞至你的模型服務端點。

要使用串流功能，請先設定串流，然後使用 StreamSource 參照它。串流來源支援以 Kafka 作為輸入來源，並自動維護擷取用的 Delta 資料表，作為資料的歷史副本以供訓練使用。

定義串流功能

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

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小時窗口對行為變化反應迅速，但可能引入變異，降低模型效能。如果模型在分布移動時準確度下降，使用較長的視窗來穩定輸入。
翻滾式和滑動式窗戶比捲動（連續）窗更具可擴展性。大多數使用情境建議先用滑動視窗。

性能

在一次 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-07-14