特徵檢視 API 參考資料

Important

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

存取控制

特徵是可治理的 Unity 目錄物件。功能存取由、 CREATE FEATURE及 READ FEATURE Unity 目錄權限控制MANAGE。完整說明請參閱 Unity 目錄權限參考資料。

CREATE FEATURE — 在結構中建立特徵所必需。 create_feature 以及 register_feature 在父架構上的需求 CREATE FEATURE 。依照最小權限原則，在結構層級授予 CREATE FEATURE ;你也可以在目錄中授予，允許在該目錄中的任何架構中建立功能。
READ FEATURE — 必須讀取特徵及其資料。 get_feature、 create_training_set，以及讀取用於訓練或服務 READ FEATURE 所需的物質化特徵資料。 READ FEATURE 在結構或目錄中授予的權限適用於其包含的所有現有及未來功能。
MANAGE — 必須管理功能生命週期及補助金。刪除帶有 delete_feature的特徵，並實現具有 materialize_features 或 delete_materialized_feature的特徵，需要 MANAGE 在該特徵上。

所有功能操作也都需要 USE CATALOG 在父目錄和 USE SCHEMA 父架構上執行。關於如何MANAGEREAD FEATURE與應用於物質化，請參見權限。

功能檢視 API

`Feature` 構造者與 `register_feature()`

建議的做法是在本地建構一個 Feature 物件，然後用 register_feature 來持久化到 Unity 目錄。這個兩步驟工作流程讓你能在註冊前先嘗試各種功能（包括 create_training_set）。

Feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    entity: Optional[List[str]] = None,                    # Required for aggregation: entity columns
    timeseries_column: Optional[str] = None,               # Required for aggregation: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
)

FeatureEngineeringClient.register_feature()在 Unity 目錄中註冊本地建構的。Feature

FeatureEngineeringClient.register_feature(
    feature: Feature,       # Required: A Feature instance (not already registered)
    catalog_name: str,      # Required: UC catalog name
    schema_name: str,       # Required: UC schema name
) -> Feature

from databricks.feature_engineering.entities import Feature, DeltaTableSource, AggregationFunction, Sum, RollingWindow
from datetime import timedelta

# Step 1: Construct the feature locally
feature = Feature(
    source=DeltaTableSource(catalog_name="main", schema_name="store", table_name="transactions"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

# Step 2: Register in Unity Catalog
fe = FeatureEngineeringClient()
registered_feature = fe.register_feature(
    feature=feature,
    catalog_name="main",
    schema_name="store",
)

`create_feature()`

FeatureEngineeringClient.create_feature() 在 Unity 目錄中驗證、建構並立即註冊功能。當不需要先在本地嘗試這個功能時，就用這個方法。

FeatureEngineeringClient.create_feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    catalog_name: str,                                     # Required: The catalog name for the feature
    schema_name: str,                                      # Required: The schema name for the feature
    entity: Optional[List[str]] = None,                    # Required for aggregation: entity columns
    timeseries_column: Optional[str] = None,               # Required for aggregation: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
) -> Feature

參數：

source：特徵計算中使用的資料來源（DeltaTableSource，， StreamSource，或 RequestSource）。
function：將 AggregationFunction 運算子（例如）、 Sum(input="amount")輸入欄位與時間窗捆綁在一起。或者 ColumnSelection("column_name") 是直通功能。
catalog_name：Unity 目錄中該功能的名稱。
schema_name： Unity 目錄中該功能的結構名稱。
entity：定義聚合層級（主鍵）的欄位名稱列表。聚合功能所必需。例如， ["user_id"] 每個使用者的聚合。
timeseries_column：用於時間窗聚合的時間戳記欄位。聚合功能所必需。
name：可選的功能名稱。若省略，則由輸入欄位、函式與視窗自動產生（例如， amount_avg_rolling_7d）。
description：功能可選描述。

退貨： 一個已驗證過的功能實例

拋出： 若任何驗證失敗，則拋出 ValueError

`delete_feature()`

刪除 Unity 目錄中以完全限定名稱刪除的功能。

FeatureEngineeringClient.delete_feature(
    full_name: str,  # Required: '<catalog>.<schema>.<feature_name>'
) -> None

fe.delete_feature(full_name="main.store.amount_sum_rolling_7d")

在刪除某個功能之前，先移除或更新任何參考該功能的模型或功能規格。如果該特徵已經被物質化，請先刪除該物質化特徵。請參閱如何刪除物質化特徵。

自動產生名稱

省 name 略時，會自動產生一個名稱。產生的名稱遵循以下模式： {column}_{function}_{window}。例如：

price_avg_rolling_1h （1小時平均價格）
transaction_count_rolling_30d_1d （交易計數 30 天，事件時間戳後延遲 1 天）

支援的功能

彙總函數

Note

聚合函數被包裹在 AggregationFunction 一個帶有時間窗的框架中，如同時間窗所述。每個函式會接受 input 一個參數，指定要彙總的來源欄位。

功能	描述	使用案例範例
`Sum(input="column")`	數值總和	每位用戶每日應用程式使用量（幾分鐘）
`Avg(input="column")`	平均值	平均交易金額
`Count(input="column")`	記錄數	每位使用者的登入次數
`Min(input="column")`	最小值	穿戴裝置記錄的最低心率
`Max(input="column")`	最大值	每個會話的最高交易金額
`StddevPop(input="column")`	母體標準差	所有客戶每日交易金額的變動
`StddevSamp(input="column")`	樣本標準差	廣告活動點擊率的變異性
`VarPop(input="column")`	族群變異數	工廠中物聯網裝置感測器讀數的分布
`VarSamp(input="column")`	樣本變異數	電影評分在抽樣群組中的分布
`ApproxCountDistinct(input="column", relativeSD=0.05)`	近似唯一計數	購買物品數量不同
`ApproxPercentile(input="column", percentile=0.95, accuracy=100)`	近似百分位	P95 回應延遲
`First(input="column")`	第一個值	第一次登入時間戳記
`Last(input="column")`	最後一個值	最近一次購買金額

`ColumnSelection` （穿透聲）

ColumnSelection 從來源中選取單一欄位，且不應用任何聚合。它直接包裹在 function 參數中（而非內部 AggregationFunction）。回傳類型是從來源結構推斷出來的。

功能	描述	使用案例範例
`ColumnSelection("col")`	欄位的最新版本值（無聚合）	最新的供應商類別，請求欄位的直通

ColumnSelection 可搭配任何資料來源使用：

DeltaTableSource：透過時間點連接（無回溯視窗聚合）回傳每個實體鍵的最新值。
StreamSource：從串流中回傳每個實體鍵的最新值（無回溯視窗彙整）。
RequestSource：通過推論時提供的值（或訓練時從標記 DataFrame 中擷取）。

from databricks.feature_engineering.entities import (
    ColumnSelection, DeltaTableSource, Feature, FieldDefinition,
    RequestSource, ScalarDataType,
)

delta_source = DeltaTableSource(
    catalog_name="main", schema_name="feature_store", table_name="transactions",
)

request_source = RequestSource(
    schema=[
        FieldDefinition(name="session_duration", data_type=ScalarDataType.DOUBLE),
    ]
)

# ColumnSelection from a Delta table
latest_amount = Feature(
    source=delta_source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    name="latest_transaction_amount",
)

# ColumnSelection from a RequestSource
session_feature = Feature(
    source=request_source,
    function=ColumnSelection("session_duration"),
    name="session_duration",
)

範例：聚合與欄位選擇功能

以下範例展示了在同一資料來源上定義的特徵。

from databricks.feature_engineering.entities import (
    AggregationFunction, Feature, Sum, Avg, ApproxCountDistinct,
    ColumnSelection, RollingWindow,
)
from datetime import timedelta

window = RollingWindow(window_duration=timedelta(days=7))

sum_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(Sum(input="amount"), window),
)

avg_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(Avg(input="amount"), window),
)

distinct_count = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(ApproxCountDistinct(input="product_id", relativeSD=0.01), window),
)

# Column selection (no aggregation, no time window)
latest_amount = Feature(
    source=source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="event_time",
    name="latest_amount",
)

具有篩選條件的特徵

這個 filter_condition 參數允許你在計算聚合前，從來源資料表中篩選資料列。這就像一個 SQL WHERE 子句，會在分組和彙整資料之前套用。

Note

filter_condition 在彙總前過濾列，就像 SQL WHERE 子句在 GROUP BY。它不會改變特徵定義上的粒度，而粒度總是由定義 entity 。

當處理包含特徵計算所需資料超集的大型來源資料表時，過濾器非常有用，並減少在這些資料表上建立獨立視圖的需求。

from databricks.feature_engineering.entities import AggregationFunction, Sum, Count, RollingWindow, DeltaTableSource
from datetime import timedelta

# Source with filter applied at the source level
high_value_transactions = DeltaTableSource(
    catalog_name="main",
    schema_name="ecommerce",
    table_name="transactions",
    filter_condition="amount > 100",  # Only transactions over $100
)

high_value_sales = Feature(
    source=high_value_transactions,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=30))),
)

# Multiple conditions
completed_orders_source = DeltaTableSource(
    catalog_name="main",
    schema_name="ecommerce",
    table_name="orders",
    filter_condition="status = 'completed' AND payment_method = 'credit_card'",
)

completed_orders = Feature(
    source=completed_orders_source,
    entity=["user_id"],
    timeseries_column="order_time",
    function=AggregationFunction(Count(input="order_id"), RollingWindow(window_duration=timedelta(days=7))),
)

# Filter on a StreamSource
from databricks.feature_engineering.entities import StreamSource

purchase_stream = StreamSource(
    full_name="main.ecommerce.transactions_stream",
    filter_condition="value.event_type = 'purchase'",
)

purchase_total = Feature(
    source=purchase_stream,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(Sum(input="value.amount"), RollingWindow(window_duration=timedelta(hours=1))),
)

數據源

`DeltaTableSource`

DeltaTableSource 是一個短暫的Python物件，用來定義如何從來源資料表計算特徵。它不會建立新的表格。它規定了讀取資料與整合特徵的配置。

DeltaTableSource(
    catalog_name: str,                              # Required: Catalog name
    schema_name: str,                               # Required: Schema name
    table_name: str,                                # Required: Table name
    filter_condition: Optional[str] = None,         # Optional: SQL WHERE clause to filter source data
    transformation_sql: Optional[str] = None,       # Optional: SQL SELECT expression for column transformations
    dataframe_schema: Optional[str] = None,         # Required if transformation_sql is set: schema of the resulting DataFrame
)

參數：

catalog_name， schema_name，： table_name在 Unity 目錄中識別來源 Delta 資料表。
filter_condition：一個在聚合前套用的SQL WHERE 條款。範例："status = 'completed'"。
transformation_sql：一個套用到來源資料表的 SQL SELECT 表達式。利用此方法重新命名欄位、鑄造類型，或在聚合前計算衍生欄位。若省略，則所有欄位皆被選取（*）。範例："user_id, CAST(amount AS DOUBLE) AS amount, event_time"。
dataframe_schema：經過轉換後所得資料框架的結構，採用 Spark StructType JSON 格式（來源 df.schema.json()）。 若提供， transformation_sql 則為必備。 這會告訴系統你轉換後產生的欄位名稱和類型。

當兩者filter_condition同時transformation_sql設定時，所得查詢為： SELECT {transformation_sql} FROM {table} WHERE {filter_condition}。

Note

（ timeseries_column 在特徵定義中指定，而非在 DeltaTableSource上）必須為型態 TimestampType 或 DateType。整數型態可行，但會降低時間窗聚合的精度。

範例：用於 transformation_sql 欄位變換

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="raw_events",
    transformation_sql="user_id, CAST(price_cents AS DOUBLE) / 100 AS price, event_time",
    filter_condition="event_type = 'purchase'",
    dataframe_schema=spark.sql(
        "SELECT user_id, CAST(price_cents AS DOUBLE) / 100 AS price, event_time FROM main.analytics.raw_events LIMIT 0"
    ).schema.json(),
)

範例：從 PySpark 資料框架衍生 transformation_sql 與 dataframe_schema 取自

你可以將轉換寫成 PySpark 查詢，然後從產生的 DataFrame 中擷取結構：

df = spark.sql(f"""
  SELECT user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time
  FROM main.analytics.events
  WHERE event_date >= date_sub(current_date(), 7)
  LIMIT 0
""")

# Use df.schema.json() as the dataframe_schema
source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    transformation_sql="user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time",
    filter_condition="event_date >= date_sub(current_date(), 7)",
    dataframe_schema=df.schema.json(),
)

Note

transformation_sql 僅支援列式運算式（欄位重命名、鑄造、算術運算）。聚合功能是否COUNT(*)SUM()支援或不支援。而是用 AggregationFunction 特徵定義來處理。

`DeltaTableSource.from_sql()`

為了方便，你可以從 SQL 查詢建立一個 DeltaTableSource 。該方法解析查詢以自動擷取資料表名稱、 transformation_sql和 filter_condition。

DeltaTableSource.from_sql(
    sql: str,                           # Required: SQL SELECT query
    spark: SparkSession,                # Required: active SparkSession (for schema inference)
) -> DeltaTableSource

僅支援簡單的 SELECT ... FROM ... [WHERE ...] 查詢。複雜的 SQL（JOIN、子查詢、 CTE、UNION）則被拒絕。對於複雜查詢，直接構造 DeltaTableSource 與 transformation_sqlfilter_condition。

from databricks.feature_engineering.entities import (
    AggregationFunction,
    DeltaTableSource,
    Feature,
    Sum,
    TumblingWindow,
)

source = DeltaTableSource.from_sql(
    spark=spark,
    sql=f"SELECT customer_id, event_ts, amount * 2 AS doubled_amount, amount FROM {CATALOG}.{SCHEMA}.{TABLE}",
)

feature = Feature(
    source=source,
    function=AggregationFunction(Sum(input="doubled_amount"), time_window=TumblingWindow(window_duration=timedelta(days=7))),
    entity=["customer_id"], timeseries_column="event_ts",
)

迭代 `to_dataframe()`

用 source.to_dataframe() 來預覽將用於特徵計算的資料。這對於反覆filter_conditiontransformation_sql迭代直到產生預期結果非常有用。

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    filter_condition="event_type = 'purchase'",
)

# Preview the filtered source data
source.to_dataframe().display()

理解實體

實體欄位定義了你特徵的聚合層級。它們是在定義上指定，而非定義 Feature 上 DeltaTableSource。實體決定：

資料分組方式：特徵依據獨特的實體值組合彙整（類似 GROUP BY SQL 中的 SQL）
主要鍵結構：每個獨特的實體組合會產生一列計算出的特徵

範例：客戶層級功能

以下程式碼彙整客戶層級的功能（每位客戶一列）：

from databricks.feature_engineering.entities import DeltaTableSource

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="user_events",
)

Feature(
    source=source,
    entity=["user_id"],                # Features aggregated per user
    timeseries_column="event_time",    # Timestamp for time windows
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

範例：客戶-門市層級功能

若要更細緻地彙整功能（每個客戶與商店組合一列），可使用多個實體欄位：

source = DeltaTableSource(
    catalog_name="main",
    schema_name="retail",
    table_name="transactions",
)

Feature(
    source=source,
    entity=["user_id", "store_id"],  # Features aggregated per user-store pair
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

當你需要在不同層級的聚合（例如客戶層級和客戶店級）提供功能時，在功能定義中使用不同的 entity 值。同樣 DeltaTableSource 的特性可以在不同實體配置的特徵間共享。

`StreamSource`

StreamSource 提及一條溪流。串流包含串流來源的連線、認證、結構及擷取設定。對於 Kafka，特徵定義中的欄位引用必須以或作為前綴value.key.，以指示要閱讀的訊息部分。

StreamSource(
    full_name: str,                       # Required: Three-part Stream name (catalog.schema.stream)
    filter_condition: Optional[str],      # Optional: SQL WHERE clause applied before aggregation
)

參數：

full_name：串流的完整三部分名稱（例如， "my_catalog.my_schema.my_stream"）。
filter_condition （可選）：在彙整前應用於串流資料的 SQL WHERE 子句，使用點前的欄位參考（例如， "value.event_type = 'purchase'"）。

from databricks.feature_engineering.entities import StreamSource

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

`RequestSource`

RequestSource 定義了在請求有效載荷推論時提供的資料結構，而非從預先實體化的資料表中查詢。在訓練過程中，這些欄位會從傳遞給 create_training_set的標記 DataFrame 中擷取。在模型服務過程中，呼叫者必須將這些資料納入 HTTP 請求有效載荷中。

RequestSource 與（直接通過一個值）一起使用 ColumnSelection 。它不支援聚合函數或時間窗。

定義模式

將結構定義為一個物件清單FieldDefinition，每個物件指定欄位名稱及：ScalarDataType

from databricks.feature_engineering.entities import (
    FieldDefinition, RequestSource, ScalarDataType,
)

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

支援的數據類型

RequestSource支援定義於 ScalarDataType的純量類型：INTEGER， FLOAT， BOOLEANSTRINGDOUBLELONGTIMESTAMPDATESHORT。複雜型態如陣列、映射和結構體不被支援。

請求資料如何被水合

背景	行為
訓練（`create_training_set`）	欄位是從標記為 DataFrame 的中擷取的。類型會根據宣告的架構進行驗證。不匹配會產生錯誤（無隱性投擲）。
服務（模型端點）	欄位是從 `dataframe_records` HTTP 請求中拉取或 `dataframe_split` 包含在 HTTP 請求中。 JSON 值會被鑄造成宣告的類型（例如 JSON 編號 → `DOUBLE`）。

模型簽章

當模型以包含log_model特徵的訓練集記錄RequestSource時，欄位RequestSource會作為必要的輸入加入 MLflow 模型簽名。這表示服務端點的 API 架構反映了呼叫者在推論時必須提供的欄位。

訓練與推論 API

`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

`log_model()`

記錄模型中包含特徵元資料，用於譜系追蹤及推論時的自動特徵查詢。詳情請參見具備特徵視圖的火車模型。

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
)

`score_batch()`

執行離線批次推論並自動查找功能。利用模型中儲存的特徵元資料計算出即時正確的特徵，確保與訓練一致。

FeatureEngineeringClient.score_batch(
    model_uri: str,                           # URI of logged model (e.g., "models:/catalog.schema.model/1")
    df: DataFrame,                            # DataFrame with entity keys and timestamps
) -> DataFrame

輸入資料框架必須包含訓練中使用的實體欄與時間序列欄位。特徵會自動從來源資料中計算。

fe = FeatureEngineeringClient()

# Batch scoring with automatic feature lookup
predictions = fe.score_batch(
    model_uri="models:/main.ecommerce.fraud_model/1",
    df=inference_df,
)
predictions.display()

時間範圍

功能檢視支援三種不同類型的視窗來控制基於時間窗的聚合的回視行為：滾動、翻滾與滑動。

搖動窗從活動時間回望過去。持續時間與延遲有明確定義。
翻滾窗是固定且不重疊的時間窗。每個資料點只屬於一個視窗。
滑動視窗是重疊的滾動時間視窗，具有可設定的滑動間隔。

以下圖示展示了它們的運作方式。

翻滾、翻滾、滑動的回望視窗。

捲動窗

Note

RollingWindow 之前稱為 ContinuousWindow。如果你是從較舊的 SDK 版本遷移過來，請相應地更新匯入。

滾動視窗是 up-to日期和即時聚合，通常用於串流資料。在串流管線中，滾動視窗只有在固定長度視窗內容改變時才會發出新列，例如事件進入或離開時。當訓練管線中使用滾動視窗特徵時，會利用特定事件時間戳前的固定長度視窗長度，對來源資料進行精確的點點計算。這有助於防止線上離線偏差或資料外洩。時間點 T 的功能會彙整事件範圍 [T − 持續時間, T)。

class RollingWindow(TimeWindow):
    window_duration: datetime.timedelta
    delay: Optional[datetime.timedelta] = None

下表列出滾動視窗的參數。視窗的開始與結束時間基於以下參數：

開賽時間： evaluation_time - window_duration - delay （含）
終結時間： evaluation_time - delay （獨家）

參數	Constraints
`delay` (選用)	必須是 ≥ 0（將視窗從評估時間戳回溯）。用 `delay` 來考慮事件建立與事件時間戳記之間的系統延遲，以防止未來事件洩漏到訓練資料集中。例如，如果事件建立與這些事件最終被放入來源表並分配時間戳記之間有一分鐘的延遲，那麼延遲時間會是 `timedelta(minutes=1)`。
`window_duration`	必須為 > 0

from databricks.feature_engineering.entities import RollingWindow
from datetime import timedelta

# Look back 7 days from evaluation time
window = RollingWindow(window_duration=timedelta(days=7))

請用下方程式碼定義一個有延遲的滾動視窗。

# Look back 7 days, offset by 1 minute to account for data ingestion delay
window = RollingWindow(
    window_duration=timedelta(days=7),
    delay=timedelta(minutes=1)
)

捲動窗範例

window_duration=timedelta(days=7)：這會產生一個7天的回溯窗口，直到當前評估時間結束。第7天下午2點的活動，包含從第0天下午2點開始到第7天下午2點為止（但不包括）的所有活動。
window_duration=timedelta(hours=1), delay=timedelta(minutes=30)：這會產生一個為期1小時的回顧視窗，結束於評估時間前30分鐘。下午3點的活動涵蓋下午1點30分至2點半（但不包括）下午3點30分的所有活動。這有助於考慮資料擷取延遲。

翻轉視窗

對於使用翻滾視窗定義的特徵，聚合會透過預先設定的固定長度視窗計算，該視窗會依滑動間隔前進，產生不重疊且完全分割時間的視窗。因此，來源中的每個事件恰好貢獻於一個視窗。功能在 Time t 上彙整從 Windows 停止於或更早 t 的時間（排他）。 Windows 從 Unix 時代開始。

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

下表列出翻滾窗的參數。

參數	Constraints
`window_duration`	必須為 > 0

from databricks.feature_engineering.entities import TumblingWindow
from datetime import timedelta

window = TumblingWindow(
    window_duration=timedelta(days=7)
)

翻滾窗戶範例

window_duration=timedelta(days=5)：這會產生預先設定的固定長度窗口，每個時段為5天。舉例來說：視窗 #1 從第 0 天到第 4 天，視窗 #2 從第 5 天到第 9 天，視窗 #3 從第 10 天到 14 天，依此類推。具體來說，視窗 #1 包含所有從第 0 天開始 00:00:00.00 的時間戳事件，直到（但不包括）任何第 5 天有時間戳 00:00:00.00 的事件。每個事件只屬於一個視窗。

滑動視窗

對於使用滑動視窗定義的特徵，聚合會透過預先設定的固定長度視窗計算，該視窗會依滑動間隔前進，產生重疊視窗。來源中的每個事件都能參與多個視窗的功能聚合。功能在 Time t 上彙整從 Windows 停止於或更早 t 的時間（排他）。 Windows 從 Unix 時代開始。

class SlidingWindow(TimeWindow):
    window_duration: datetime.timedelta
    slide_duration: datetime.timedelta

下表列出滑動視窗的參數。

參數	Constraints
`window_duration`	必須為 > 0
`slide_duration`	必須為 > 0，且 <`window_duration`

from databricks.feature_engineering.entities import SlidingWindow
from datetime import timedelta

window = SlidingWindow(
    window_duration=timedelta(days=7),
    slide_duration=timedelta(days=1)
)

滑動視窗範例

window_duration=timedelta(days=5), slide_duration=timedelta(days=1)：這會產生重疊的5天窗口，每次提前1天。範例：視窗 #1 涵蓋第 0 天至第 4 天，視窗 #2 涵蓋第 1 天至第 5 天，視窗 #3 涵蓋第 2 天至第 6 天，依此類推。每個視窗包含從 00:00:00.00 起始日到結束日（但不包括） 00:00:00.00 的事件。由於視窗重疊，單一事件可能屬於多個視窗（在此範例中，每個事件最多屬於五個不同的視窗）。

物質化觸發

觸發器控制物質化管線何時運行。觸發類型取決於功能類型。

`CronSchedule`

用於 CronSchedule 聚合特徵（AggregationFunction）。管線依固定排程運行，該排程由 Quartz cron 表達式定義。

from databricks.feature_engineering.entities import CronSchedule

trigger = CronSchedule(
    quartz_cron_expression="0 0 * * * ?",  # Hourly
    timezone_id="UTC",
)

`TableTrigger`

用於TableTriggerColumnSelection由 DeltaTableSource. 當上游 Delta 資料表收到新的提交時，管線就會執行。

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

`StreamingMode`

用於 StreamingMode 由 StreamSource. 該管線以連續串流管線形式運作。

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

fe = FeatureEngineeringClient()

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

streaming_feature = fe.create_feature(
    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)),
    ),
    catalog_name="my_catalog",
    schema_name="my_schema",
    name="user_purchase_sum",
)

fe.materialize_features(
    features=[streaming_feature],
    online_config=OnlineStoreConfig(
        catalog_name="my_catalog",
        schema_name="my_schema",
        table_name_prefix="streaming_features_serving",
        online_store_name="feature_store_online",
    ),
    trigger=StreamingMode(),
)

選擇觸發點

特徵類型	Trigger	當它運行時
聚合（`AggregationFunction`） `DeltaTableSource`	`CronSchedule`	在固定 cron 排程上
`ColumnSelection` （摘自 `DeltaTableSource`）	`TableTrigger`	在每個來源資料表上提交
特色來自 `StreamSource`	`StreamingMode`	連續串流

你無法在單一 materialize_features 通話中實現需要不同觸發類型的功能。改為分開來電。

將測試版功能遷移至公開預覽版

功能檢視公開預覽在 Unity 目錄中引入了一流 的功能 實體，受 CREATE FEATURE 和 READ FEATURE 權限管理，且需要 databricks-feature-engineering 版本 0.16.0 或更新版本。測試版（版本 0.15.0）期間創建的功能會以 Unity 目錄函式儲存，並不支援所有公開預覽功能。若要獲得長期公開預覽支援，請以 0.16.0 版本重建您的測試版功能。特徵必須被刪除並重新創造，而非僅僅是重新物質化。

欲了解更多功能資訊，請參閱功能檢視。

採取行動

升級到 0.16.0。 這是公開預覽功能（批次與串流）所需的客戶端版本。
重現你的特徵。 Beta 功能檢視必須刪除並重新建立，而非重新實體化，因為它們並不支援所有公開預覽功能。
在窗口關閉前遷移。 現有的測試版功能必須在 2026 年 7 月 22 日前完成遷移。

識別測試版與公開預覽功能

公開預覽功能會以功能物件的形式出現在 Unity 目錄中，例如在目錄檔案總管中。 Beta 功能以帶有 YAML 定義的函式形式出現。任何以函式形式表示的功能都是你需要遷移的測試版功能。

遷移測試版功能

遷移測試版功能有三個階段：

重新建立此功能作為公開預覽功能。
重新實體化該功能，使其離線與線上資料表在新功能下重建。
驗證遷移功能後，刪除測試版功能及其實體化。

重現這些特徵

可使用 list_beta_feature_views 來尋找你的測試版功能、 Feature.clone() 建立未註冊的副本，以及 register_feature 重新註冊每份副本作為公開預覽功能。克隆會清除註冊、目錄和結構，以便重新註冊該功能。

為避免名稱衝突，請將遷移的功能註冊為與 beta 功能不同的名稱或結構。以下範例將每個特徵重新註冊為其原始結構，並加上 _migrated 名稱後綴。

# Update this to the catalog whose beta Feature Views you want to migrate.
CATALOG_TO_MIGRATE = "main"

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# 1. Find every beta Feature View in the catalog. Returns Feature objects,
#    scanned across all schemas in the catalog.
beta_features = fe.list_beta_feature_views(catalog_name=CATALOG_TO_MIGRATE)

# Keep each beta feature paired with its migrated counterpart for the next steps.
migrations = []
for beta_feature in beta_features:
    catalog_name, schema_name, leaf_name = beta_feature.full_name.split(".")
    # 2. Clone the feature as an unregistered copy, renamed with a "_migrated" suffix.
    cloned = beta_feature.clone(new_name=f"{leaf_name}_migrated")
    # 3. Re-register the clone as a Public Preview feature.
    migrated = fe.register_feature(
        feature=cloned,
        catalog_name=catalog_name,
        schema_name=schema_name,
    )
    migrations.append((beta_feature, migrated))

重新實體化遷移的特徵

如果有測試版功能實現，請重新實體化其公開預覽版，使其離線與線上表格在新功能下重建。提供遷移功能中的離線與線上商店設定，並從測試版功能的現有實體化中重建觸發條件。

from databricks.feature_engineering.entities import (
    CronSchedule,
    OfflineStoreConfig,
    OnlineStoreConfig,
    TableTrigger,
)

for beta_feature, migrated in migrations:
    # Inspect the beta feature's existing materializations to see what to rebuild and
    # to reconstruct the same trigger.
    trigger = None
    needs_offline = needs_online = False
    for mf in fe.list_materialized_features(feature_name=beta_feature.full_name):
        needs_online = needs_online or bool(mf.is_online)
        needs_offline = needs_offline or not mf.is_online
        # Rebuild the trigger from the materialized feature.
        if mf.cron_schedule_trigger is not None:
            trigger = CronSchedule(
                quartz_cron_expression=mf.cron_schedule_trigger.cron_expression,
                timezone_id="UTC",  # Materialized schedules run in UTC.
            )
        elif mf.table_trigger is not None:
            trigger = TableTrigger()
        elif mf.streaming_mode is not None:
            # Streaming features use StreamingMode, which can be reused as-is.
            trigger = mf.streaming_mode
    if not (needs_offline or needs_online):
        continue  # The beta feature was never materialized.

    catalog_name, schema_name, _ = migrated.full_name.split(".")
    fe.materialize_features(
        features=[migrated],
        offline_config=OfflineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
        )
        if needs_offline
        else None,
        online_config=OnlineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
            online_store_name="my_online_store",
        )
        if needs_online
        else None,
        trigger=trigger,
    )

Note

將每個特徵實體化為其獨立 materialize_features 呼叫，會建立獨立的管線。為了降低計算成本，將共享離線與線上目的地並觸發的功能，透過將它們合併materialize_features為單一features呼叫。

刪除測試版功能

Warning

只有在確認遷移功能及其實體化資料正確後，才刪除測試版功能及其實體化。刪除是不可逆的。

驗證遷移功能後，刪除每個測試版特徵的實體化，然後刪除測試版特性本身。

for beta_feature, _ in migrations:
    # Delete the beta feature's materializations first.
    mfs = list(fe.list_materialized_features(feature_name=beta_feature.full_name))
    offline_mfs = [mf for mf in mfs if not mf.is_online]
    if offline_mfs:
        # Aggregation features pair an offline and online table; deleting the offline
        # materialized feature removes its paired online table too.
        for mf in offline_mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    else:
        # Online-only features (ColumnSelection, streaming) have no offline pair; delete
        # the online materialized feature directly.
        for mf in mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    # Then delete the beta feature definition.
    fe.delete_feature(full_name=beta_feature.full_name)

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-14

特徵檢視 API 參考資料

存取控制

功能檢視 API

Feature 構造者與 register_feature()

create_feature()

delete_feature()

自動產生名稱

支援的功能

彙總函數

ColumnSelection （穿透聲）

範例：聚合與欄位選擇功能

具有篩選條件的特徵

數據源

DeltaTableSource

DeltaTableSource.from_sql()

迭代 to_dataframe()

理解實體

StreamSource

RequestSource

定義模式

支援的數據類型

請求資料如何被水合

模型簽章

訓練與推論 API

create_training_set()

log_model()

score_batch()

時間範圍

捲動窗

捲動窗範例

翻轉視窗

翻滾窗戶範例

滑動視窗

滑動視窗範例

物質化觸發

CronSchedule

TableTrigger

StreamingMode

選擇觸發點

將測試版功能遷移至公開預覽版

採取行動

識別測試版與公開預覽功能

遷移測試版功能

重現這些特徵

重新實體化遷移的特徵

刪除測試版功能

意見反應

其他資源

`Feature` 構造者與 `register_feature()`

`create_feature()`

`delete_feature()`

`ColumnSelection` （穿透聲）

`DeltaTableSource`

`DeltaTableSource.from_sql()`

迭代 `to_dataframe()`

`StreamSource`

`RequestSource`

`create_training_set()`

`log_model()`

`score_batch()`

`CronSchedule`

`TableTrigger`

`StreamingMode`