特徴量ビューAPIリファレンス

備考

プレビュー

この機能はパブリックプレビュー段階です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

アクセス制御

特徴量は、ガバナンス可能な Unity Catalog オブジェクトです。特徴量へのアクセスは、CREATE FEATURE、READ FEATURE、および MANAGE の Unity Catalog 権限によって制御されます。詳細については、Unity Catalog 権限のリファレンスを参照してください。

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も必要です。MANAGEとREAD FEATUREがマテリアライズに適用される方法については、権限を参照してください。

特徴量ビューAPI

`Feature` コンストラクタと `register_feature()`

推奨されるアプローチは、Featureオブジェクトをローカルで構築し、register_featureを使用してUnity Catalogに永続化することです。この2段階のワークフローでは、特徴量（create_training_setを含む）を登録する前に試すことができます。

Python
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() ローカルで構築された Feature を Unity Catalog に登録する。

Python
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

Python
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 Catalog内の機能を検証、構築し、すぐに登録します。最初にローカルで機能のエクスペリメントをする必要がない場合、これを使用します。

Python
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演算子（例：Sum(input="amount")）、入力列、および時間枠をまとめたAggregationFunctionです。または、パススルー機能の場合はColumnSelection("column_name")を使用します。
catalog_name特徴量に対するUnity Catalogのカタログ名。
schema_name：特徴量のUnity Catalogスキーマ名です。
entity：集約レベルを定義する列名（主キー）のリストです。集約機能に必要です。たとえば、["user_id"]はユーザーごとに集約されます。
timeseries_column時間ウィンドウ集約に使用されるTimestamp列です。集計機能に必須です。
name：任意の機能名。省略した場合、入力列、関数、およびウィンドウから自動生成されます（例：amount_avg_rolling_7d）。
description：特徴量のオプションの説明。

戻り値： 検証済みの特徴量インスタンス

発生： いずれかの検証が失敗した場合、ValueErrorが発生します。

`delete_feature()`

Unity Catalogから、その完全修飾名で特徴量を削除します。

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

Python
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 （イベントの Timestamp からの 1 日遅延を伴う 30 日間のトランザクション数）

サポートされている関数

集計関数

注記

集計関数は、時間ウィンドウに記載されているように、時間ウィンドウと共に AggregationFunction でラップされています。各関数は、集計するソース列を指定する input パラメーターを受け取ります。

関数	説明	使用例
`Sum(input="column")`	値の合計	ユーザーごとの1日のアプリ使用量 (単位: 分)
`Avg(input="column")`	平均値	平均トランザクション額
`Count(input="column")`	レコード数	ユーザーあたりのログイン数
`Min(input="column")`	最小値	ウェアラブルデバイスによって記録された最低心拍数
`Max(input="column")`	最大値	セッションあたりの最大トランザクション量
`StddevPop(input="column")`	母集団標準偏差	すべての顧客における日次取引額の変動
`StddevSamp(input="column")`	サンプル標準偏差	広告キャンペーンのクリック率の変動性
`VarPop(input="column")`	母集団分散	工場におけるIoTデバイスのセンサー読み取り値の分散
`VarSamp(input="column")`	サンプルバリアンス	サンプリングされたグループ全体での映画の評価の広がり
`ApproxCountDistinct(input="column", relativeSD=0.05)`	おおよそのユニークカウント	購入されたアイテムの個別カウント
`ApproxPercentile(input="column", percentile=0.95, accuracy=100)`	近似パーセンタイル	p95 応答レイテンシ
`First(input="column")`	最初の値	最初のログインTimestamp
`Last(input="column")`	最後の値	最新の購入額

`ColumnSelection` （パススルー）

ColumnSelection 集計を適用せずにソースから単一の列を選択します。それは直接 function パラメーターにラップされています（AggregationFunction の中ではありません）。戻り値の型はソーススキーマから推論されます。

関数	説明	使用例
`ColumnSelection("col")`	列の最新値（集計なし）	最新のベンダーカテゴリ、リクエストフィールドのパススルー

ColumnSelection 任意のデータソースで使用できます：

DeltaTableSource ：時点結合によりエンティティキーごとに最新の値を返します（ルックバックウィンドウ集計なし）。
StreamSource ** **：ストリームからエンティティキーごとの最新値を返します（ルックバックウィンドウ集計なし）。
RequestSource ** **：推論時に提供された値（またはトレーニング時にラベル付けされたDataFrameから抽出された値）を渡します。

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

例：集計と列選択機能

次の例では、同じデータソースに対して定義された機能を示します。

Python
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句として機能します。

注記

filter_condition 集計前にSQL WHERE句がGROUP BYの前に適用されるように、集計前に行をフィルタリングします。これは粒度を変更しません。粒度は、特徴量定義でentityによって常に定義されます。

フィルターは、特徴量コンピュテーションに必要なデータのスーパーセットを含む大規模なソーステーブルを扱う際に役立ち、これらのテーブル上に個別のビューを作成する必要性を最小限に抑えます。

Python
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オブジェクトです。新しいテーブルは作成されません。データの読み取りと特徴量の集計のための設定を指定します。

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 CatalogのソースDeltaテーブルを識別します。
filter_condition：集計前に適用されるSQL WHERE句。例："status = 'completed'"。
transformation_sql：ソーステーブルに適用される SQL SELECT 式。これを使用して、集計前に列の名前を変更したり、型をキャストしたり、派生列をコンピュートしたりします。省略された場合、すべての列が選択されます（*）。例："user_id, CAST(amount AS DOUBLE) AS amount, event_time"。
dataframe_schema：変換後の結果のDataFrameのスキーマ。Spark StructType JSON形式（df.schema.json()から）。 transformation_sqlが提供されている場合、必須です。 これは、変換によって生成される列名と型をシステムに伝えます。

filter_condition と transformation_sql の両方が設定されている場合、結果のクエリーは次のようになります：SELECT {transformation_sql} FROM {table} WHERE {filter_condition}。

注記

timeseries_column（DeltaTableSourceではなく特徴量定義で指定された）はタイプTimestampTypeまたはDateTypeである必要があります。Integer型は機能しますが、時間枠集計の精度が低下します。

例：列の変換に transformation_sql を使用する

Python
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 DataFrameからtransformation_sqlとdataframe_schemaを導出する

変換をPySparkクエリーとして記述し、その後、結果のDataFrameからスキーマを抽出できます。

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

注記

transformation_sql 行ごとの式 (列の名前変更、キャスト、算術演算) のみサポートします。COUNT(*)またはSUM()のような集計関数はサポートされていません。代わりに特徴量定義でAggregationFunctionを使用します。

`DeltaTableSource.from_sql()`

便宜上、SQLクエリーからDeltaTableSourceを作成できます。このメソッドはクエリーを解析し、テーブル名、transformation_sql、およびfilter_conditionを自動的に抽出します。

Python
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）は拒否されます。複雑なクエリーの場合は、transformation_sqlとfilter_conditionでDeltaTableSourceを直接構築します。

Python
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_conditionとtransformation_sqlを反復するのに役立ちます。

Python
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では指定されません。エンティティが決定するもの：

データのグループ化方法 ：特徴量は、エンティティ値の一意の組み合わせごとに集計されます（SQLのGROUP BYに類似）。
主キー構造 ：各一意のエンティティの組み合わせは、コンピュートされた特徴の1行になります。

例：顧客レベルの機能

次のコードは、顧客レベルで特徴量を集計します（顧客ごとに 1 行）。

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

例：顧客・ストアレベルの特徴量

より詳細なレベルで特徴を集約するには（顧客と店舗の組み合わせごとに1行）、複数のエンティティ列を使用します：

Python
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.をプレフィックスとして付ける必要があります。

Python
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：ストリームの完全な3部構成の名前（例："my_catalog.my_schema.my_stream"）。
filter_condition （オプション）：集計前にストリームデータに適用されるSQL WHERE句で、ドットプレフィックス付きの列参照（例えば、"value.event_type = 'purchase'"）を使用します。

Python
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を指定します。

Python
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、BOOLEAN、STRING、DOUBLE、LONG、TIMESTAMP、DATE、SHORT。配列、マップ、構造体などの複合型はサポートされていません。

リクエストデータがどのようにハイドレートされるか

コンテキスト	挙動
トレーニング（`create_training_set` ）	列はラベル付きDataFrameから抽出されます。型は宣言されたスキーマに対して検証されます。不一致があるとエラーが発生します（暗黙的なキャストなし）。
サービング (モデルEndpoint)	HTTPリクエストでは、`dataframe_records`または`dataframe_split`から列が取得されます。JSON値は、宣言された型にキャストされます（例：JSON数値 → `DOUBLE`）。

モデルシグネチャ

RequestSource特徴を含むトレーニングセットでlog_modelを使用してモデルがログに記録されると、必要な入力としてRequestSource列がMLflowモデル署名に追加されます。これは、サービングEndpointのAPIスキーマが、呼び出し元が推論時に提供する必要があるフィールドを反映していることを意味します。

トレーニングと推論API

`create_training_set()`

時点補正された特徴量計算を含むトレーニングデータセットを作成します。詳細については、フィーチャービューでモデルをトレーニングするを参照してください。

Python
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()`

Logs a model with feature metadata for リネージ tracking and automatic feature lookup during inference.詳細については、フィーチャービューでモデルをトレーニングするを参照してください。

Python
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()`

自動の特徴量ルックアップを使用して、オフラインバッチ推論を実行します。モデルと一緒に保存されている特徴量メタデータを使用して、時点に合った特徴量をコンピュートし、トレーニングとの一貫性を確保します。

Python
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

入力 DataFrame には、トレーニング中に使用されたエンティティ列と時系列列が含まれている必要があります。特徴はソースデータから自動的にコンピュートされます。

Python
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()

時間ウィンドウ

特徴量ビューは、時間ウィンドウベースの集計におけるルックバック動作を制御するために、ローリング、タンブリング、スライディングの3種類のウィンドウタイプをサポートしています。

ローリングウィンドウはイベント時間から遡って参照します。期間と遅延は明示的に定義されています。
タンブリングウィンドウは、固定された重複しない時間ウィンドウです。各データポイントは1つのウィンドウにのみ属します。
スライディングウィンドウは、設定可能なスライド間隔を持つ、重複するローリング時間ウィンドウです。

次の図は、それらがどのように機能するかを示しています。

ローリング、タンブリング、スライディングのルックバックウィンドウ。

ローリングウィンドウ

注記

RollingWindow 以前はContinuousWindowという名前でした。以前のSDKバージョンから移行している場合は、インポートをそれに応じて更新してください。

ローリングウィンドウは最新のリアルタイム集計であり、通常はストリーミングデータで使用されます。ストリーミングパイプラインでは、イベントの入力または出力など、固定長ウィンドウの内容が変更された場合にのみ、ローリングウィンドウは新しい行を出力します。トレーニングパイプラインでローリングウィンドウ特徴量が使用される場合、特定のイベントのTimestampの直前にある固定長ウィンドウ期間を使用して、ソースデータに対して正確な時点の特徴量計算が実行されます。これは、オンライン／オフラインのずれやデータ漏洩を防ぐのに役立ちます。T時点の特徴量は、[T − 期間, T)からイベントを集約します。

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

次の表に、ローリングウィンドウのパラメーターを示します。ウィンドウの開始時刻と終了時刻は、これらのパラメーターに基づいて次のように決まります。

開始時刻：evaluation_time - window_duration - delay（含む）
終了時刻：evaluation_time - delay (排他的)

パラメーター	制約
`delay` (任意)	0以上である必要があります（評価 Timestamp から時間枠を過去にシフトします）。`delay`を使用すると、イベントが作成された時刻とイベントのTimestampとの間のシステム遅延を考慮し、将来のイベントがトレーニングデータセットに漏洩するのを防ぐことができます。たとえば、イベントが作成されてから、それらのイベントがTimestampが割り当てられたソーステーブルに最終的に取り込まれるまでの間に1分の遅延がある場合、その遅延は`timedelta(minutes=1)`になります。
`window_duration`	0より大きい必要があります。

パラメーター

制約

delay (任意)

0以上である必要があります（評価 Timestamp から時間枠を過去にシフトします）。delayを使用すると、イベントが作成された時刻とイベントのTimestampとの間のシステム遅延を考慮し、将来のイベントがトレーニングデータセットに漏洩するのを防ぐことができます。たとえば、イベントが作成されてから、それらのイベントがTimestampが割り当てられたソーステーブルに最終的に取り込まれるまでの間に1分の遅延がある場合、その遅延はtimedelta(minutes=1)になります。

window_duration

0より大きい必要があります。

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

以下のコードを使用して、遅延を伴うローリングウィンドウを定義します。

Python
# 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時までの（7日目の午後2時は含まず）すべてのイベントを含みます。
window_duration=timedelta(hours=1), delay=timedelta(minutes=30)：これにより、評価時刻の30分前で終わる1時間のルックバックウィンドウが作成されます。午後3時のイベントの場合、これは午後1時30分から午後2時30分までの（午後2時30分は含まず）すべてのイベントを含みます。これは、データ取り込みの遅延に対応するために役立ちます。

タンブリングウィンドウ

タンブリングウィンドウを使用して定義された特徴量の場合、集計は、スライド間隔で進む所定の固定長ウィンドウで計算され、時間を完全に分割する重複しないウィンドウが生成されます。その結果、ソース内の各イベントは正確に1つのウィンドウに寄与します。時刻tの特徴量は、t (排他的) 以前に終了するウィンドウからデータを集計します。ウィンドウはUnixエポックで開始します。

Python
class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

次の表に、タンブリングウィンドウのパラメーターを示します。

パラメーター	制約
`window_duration`	0より大きい必要があります。

Python
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 は Day 0 から Day 4 にわたり、ウィンドウ #2 は Day 5 から Day 9 にわたり、ウィンドウ #3 は Day 10 から Day 14 にわたり、などです。具体的には、ウィンドウ #1 は、Day 0 の 00:00:00.00 に開始する Timestamp を持つすべてのイベントから、Day 5 の Timestamp 00:00:00.00 を持つイベントまで（ただし、含まず）を含みます。各イベントは厳密に1つのウィンドウに属します。

スライディングウィンドウ

スライディングウィンドウを使用して定義された特徴量の場合、集計は、スライド間隔で進む事前に決定された固定長ウィンドウで計算され、重複するウィンドウを生成します。ソース内の各イベントは、複数のウィンドウの特徴量集計に寄与できます。時刻tの特徴量は、t (排他的) 以前に終了するウィンドウからデータを集計します。ウィンドウはUnixエポックで開始します。

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

次の表に、スライディングウィンドウのパラメーターを示します。

パラメーター	制約
`window_duration`	0より大きい必要があります。
`slide_duration`	> 0 かつ < である必要があります。 `window_duration`

Python
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)これにより、毎回1日ずつ進む重複する5日間のウィンドウが作成されます。例: ウィンドウ #1は0日目から4日目まで、ウィンドウ #2は1日目から5日目まで、ウィンドウ #3は2日目から6日目までといった具合です。各ウィンドウには、開始日の00:00:00.00から終了日の00:00:00.00まで (ただしは含まない) のイベントが含まれます。ウィンドウが重複しているため、単一のイベントが複数のウィンドウに属することができます (この例では、各イベントは最大5つの異なるウィンドウに属します)。

Materialization Trigger

トリガーは、実体化パイプラインが実行されるタイミングを制御します。Triggerタイプはフィーチャータイプによって異なります。

`CronSchedule`

集計特徴量（AggregationFunction）にはCronScheduleを使用します。パイプラインは、Quartz cron式で定義された固定スケジュールで実行されます。

Python
from databricks.feature_engineering.entities import CronSchedule

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

`TableTrigger`

DeltaTableSourceによってサポートされるColumnSelection特徴量にはTableTriggerを使用します。アップストリームのDeltaテーブルが新しいcommitを受信するたびに、パイプラインが実行されます。

Python
from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

`StreamingMode`

StreamSourceによってサポートされる特徴量にはStreamingModeを使用します。パイプラインは、継続的なストリーミングパイプラインとして実行されます。

Python
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`	各ソーステーブルcommit時
からの特徴量 `StreamSource`	`StreamingMode`	連続ストリーミング

単一のmaterialize_features呼び出しで、異なるTriggerタイプを必要とする特徴量をマテリアライズすることはできません。代わりに、個別の呼び出しを発行します。

ベータ機能をパブリックプレビューに移行する

特徴量ビューのパブリックプレビューでは、Unity Catalog にファーストクラスの**特徴量**エンティティが導入されており、CREATE FEATURE READ FEATUREおよびの権限によって管理され、databricks-feature-engineering バージョン 0.16.0 以降が必要です。ベータ版（バージョン 0.15.0）で作成された特徴量は Unity Catalog の関数として保存され、すべてのパブリックプレビュー機能には対応していません。長期のパブリックプレビューサポートを受けるには、ベータ版の特徴量をバージョン 0.16.0 で再作成してください。特徴量は、再マテリアライズするだけでなく、削除して再作成する必要があります。

特徴量に関する詳細情報は、特徴量ビューをご覧ください。

実施事項

0.16.0にアップグレードします。 これは、パブリックプレビュー機能（バッチおよびストリーミング）に必要なクライアントバージョンです。
**特徴量を再作成してください。** ベータ版の特徴量ビューは、すべてのパブリックプレビュー機能に対応していないため、再マテリアライズするのではなく、削除して再作成する必要があります。
**期間が終了する前に移行してください。** 既存のベータ機能は、2026年7月22日より前に移行する必要があります。

ベータ版およびパブリックプレビューのフィーチャーを識別する

Public Preview 機能は、Unity Catalog において、例えばカタログエクスプローラで Feature オブジェクトとして表示されます。ベータ機能は、YAML定義を持つ関数として表示されます。関数として表現されるすべての特徴量は、移行する必要があるベータ版の特徴量です。

ベータ機能の移行

ベータ機能の移行には3つの部分があります：

機能をパブリックプレビュー機能として再作成します。
特徴量を再マテリアライズして、オフラインおよびオンラインテーブルが新しい特徴量のもとで再構築されるようにします。
移行された機能を検証後、ベータ機能とそのマテリアライズを削除します。

フィーチャーを再作成する

ベータ機能を見つけるにはlist_beta_feature_viewsを使用し、未登録のコピーを作成するにはFeature.clone()を使用し、各コピーをパブリックプレビュー機能として再登録するにはregister_featureを使用します。クローニングにより登録、カタログ、スキーマがクリアされるため、特徴量を再登録できます。

名前の競合を避けるため、移行された特徴量をベータ版の特徴量とは異なる名前で、または異なるスキーマに登録してください。以下の例は、元のスキーマに _migrated という名前のサフィックスを付けて各フィーチャーを再登録します。

Python
# 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))

移行された特徴量を再マテリアライズ

ベータ機能がマテリアライズされた場合は、そのパブリックプレビュー版を再マテリアライズし、オフラインおよびオンラインのテーブルが新しい機能の下で再構築されるようにします。移行された特徴量のためにオフラインおよびオンラインストアの構成を提供し、ベータ機能の既存のマテリアライズからTriggerを再構築します。

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

注記

それぞれの特徴を独自のmaterialize_features呼び出しで具体化すると、個別のパイプラインが作成されます。コンピュートコストを削減するために、オフラインおよびオンラインの保存先を共有する特徴をグループ化し、featuresでそれらを一緒に渡すことによって、単一のmaterialize_features呼び出しにTriggerします。

ベータ版の特徴量を削除する

警告

移行された機能とそのマテリアライズされたデータが正しいことを確認した後でのみ、ベータ機能とそのマテリアライズを削除してください。削除は元に戻すことはできません。

移行された特徴量を検証した後、各ベータ機能のマテリアライズを削除し、次にベータ機能自体を削除します。

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

アクセス制御​

特徴量ビュー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()​

時間ウィンドウ​

ローリングウィンドウ​

ローリングウィンドウの例​

タンブリングウィンドウ​

タンブリングウィンドウの例​

スライディングウィンドウ​

スライディングウィンドウの例​

Materialization Trigger​

CronSchedule​

TableTrigger​

StreamingMode​

Triggerの選択​

ベータ機能をパブリックプレビューに移行する​

実施事項​

ベータ版およびパブリックプレビューのフィーチャーを識別する​

ベータ機能の移行​

フィーチャーを再作成する​

移行された特徴量を再マテリアライズ​

ベータ版の特徴量を削除する​

アクセス制御

特徴量ビュー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()`

時間ウィンドウ

ローリングウィンドウ

ローリングウィンドウの例

タンブリングウィンドウ

タンブリングウィンドウの例

スライディングウィンドウ

スライディングウィンドウの例

Materialization Trigger

`CronSchedule`

`TableTrigger`

`StreamingMode`

Triggerの選択

ベータ機能をパブリックプレビューに移行する

実施事項

ベータ版およびパブリックプレビューのフィーチャーを識別する

ベータ機能の移行

フィーチャーを再作成する

移行された特徴量を再マテリアライズ

ベータ版の特徴量を削除する