特徴量ビュー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を含む）を登録する前に試すことができます。
Feature(
source: DataSource, # Required: DeltaTableSource, StreamSource, or RequestSource
function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
entity: Optional[List[str]] = None, # Required for all sources except RequestSource: entity columns
timeseries_column: Optional[str] = None, # Required for all sources except RequestSource: 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 に登録する。
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 Catalog内の機能を検証、構築し、すぐに登録します。最初にローカルで機能のエクスペリメントをする必要がない場合、これを使用します。
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 all sources except RequestSource: entity columns
timeseries_column: Optional[str] = None, # Required for all sources except RequestSource: 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集計またはルックアップキー（主キー）を定義する列名のリストです。RequestSourceを除くすべてのソースタイプに必須です。たとえば、["user_id"]はユーザーごとに集計またはルックアップします。timeseries_column：時間ウィンドウ集計または最新値選択に使用されるTimestamp列。RequestSourceを除くすべてのソースタイプに必須です。name：任意の機能名。省略した場合、入力列、関数、およびウィンドウから自動生成されます（例：amount_avg_rolling_7d）。description：特徴量のオプションの説明。
戻り値： 検証済みの特徴量インスタンス
発生： いずれかの検証が失敗した場合、ValueErrorが発生します。
delete_feature()
Unity Catalogから、その完全修飾名で特徴量を削除します。
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（イベントの Timestamp からの 1 日遅延を伴う 30 日間のトランザクション数）
サポートされている関数
集計関数
集計関数は、時間ウィンドウに記載されているように、時間ウィンドウと共に AggregationFunction でラップされています。各関数は、集計するソース列を指定する input パラメーターを受け取ります。
関数 | 説明 | 使用例 |
|---|---|---|
| 値の合計 | ユーザーごとの1日のアプリ使用量 (単位: 分) |
| 平均値 | 平均トランザクション額 |
| レコード数 | ユーザーあたりのログイン数 |
| 最小値 | ウェアラブルデバイスによって記録された最低心拍数 |
| 最大値 | セッションあたりの最大トランザクション量 |
| 母集団標準偏差 | すべての顧客における日次取引額の変動 |
| サンプル標準偏差 | 広告キャンペーンのクリック率の変動性 |
| 母集団分散 | 工場におけるIoTデバイスのセンサー読み取り値の分散 |
| サンプルバリアンス | サンプリングされたグループ全体での映画の評価の広がり |
| おおよそのユニークカウント | 購入されたアイテムの個別カウント |
| 近似パーセンタイル | p95 応答レイテンシ |
| 最初の値 | 最初のログインTimestamp |
| 最後の値 | 最新の購入額 |
First そしてLastはdefaultで空値を含みます。空値をスキップするには、nullである入力列を明示的に除外するfilter_conditionを追加します。
ColumnSelection （パススルー）
ColumnSelection 集計を適用せずにソースから単一の列を選択します。それは直接 function パラメーターにラップされています（AggregationFunction の中ではありません）。戻り値の型はソーススキーマから推論されます。
関数 | 説明 | 使用例 |
|---|---|---|
| 列の最新値（集計なし） | 最新のベンダーカテゴリ、リクエストフィールドのパススルー |
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句として機能します。
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 CatalogのソースDeltaテーブルを識別します。filter_condition：集計前に適用されるSQLWHERE句。例："status = 'completed'"。transformation_sql：ソーステーブルに適用される SQLSELECT式。これを使用して、集計前に列の名前を変更したり、型をキャストしたり、派生列をコンピュートしたりします。省略された場合、すべての列が選択されます（*）。例："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 を使用する
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からスキーマを抽出できます。
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を自動的に抽出します。
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を直接構築します。
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を反復するのに役立ちます。
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 行）。
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行）、複数のエンティティ列を使用します：
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：ストリームの完全な3部構成の名前（例："my_catalog.my_schema.my_stream"）。filter_condition（オプション）：集計前にストリームデータに適用されるSQLWHERE句で、ドットプレフィックス付きの列参照（例えば、"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、BOOLEAN、STRING、DOUBLE、LONG、TIMESTAMP、DATE、SHORT。配列、マップ、構造体などの複合型はサポートされていません。
リクエストデータがどのようにハイドレートされるか
コンテキスト | 挙動 |
|---|---|
**トレーニング**（ | 列はラベル付きDataFrameから抽出されます。型は宣言されたスキーマに対して検証されます。不一致があるとエラーが発生します（暗黙的なキャストなし）。 |
サービング (モデルEndpoint) | HTTPリクエストでは、 |
モデルシグネチャ
RequestSource特徴を含むトレーニングセットでlog_modelを使用してモデルがログに記録されると、必要な入力としてRequestSource列がMLflowモデル署名に追加されます。これは、サービングEndpointのAPIスキーマが、呼び出し元が推論時に提供する必要があるフィールドを反映していることを意味します。
トレーニングと推論API
create_training_set およびscore_batchが、ソースデータから特定の時点に正しい特徴量値をオンデマンドでコンピュートします。オフラインマテリアライゼーションをサポートする機能（Deltaテーブルソースでのスライディングウィンドウ集計など）の場合、まず機能をオフラインストアにマテリアライズすると、両方の操作のパフォーマンスが向上します。マテリアライズされたオフラインの特徴量が利用可能な場合、操作はソースから特徴量値を再計算する代わりに、事前に計算されたオフラインデータを読み取ります。オフラインストアに特徴量をマテリアライズするには、特徴量ビューをマテリアライズするを参照してください。
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()
Logs a model with feature metadata for リネージ tracking and automatic feature lookup during inference.詳細については、フィーチャービューでモデルをトレーニングするを参照してください。
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
入力 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()
時間ウィンドウ
特徴量ビューは、時間ウィンドウベースの集計におけるルックバック動作を制御するために、ローリング、タンブリング、スライディングの3種類のウィンドウタイプをサポートしています。
- ローリングウィンドウはイベント時間から遡って参照します。期間と遅延は明示的に定義されています。
- タンブリングウィンドウは、固定された重複しない時間ウィンドウです。各データポイントは1つのウィンドウにのみ属します。
- スライディングウィンドウは、設定可能なスライド間隔を持つ、重複するローリング時間ウィンドウです。
次の図は、それらがどのように機能するかを示しています。
ローリングウィンドウ
RollingWindow 以前はContinuousWindowという名前でした。以前のSDKバージョンから移行している場合は、インポートをそれに応じて更新してください。
ローリング ウィンドウは最新のリアルタイム集計であり、通常はストリーミング データで使用されます。ストリーミングパイプラインでは、イベントの入力または出力など、固定長ウィンドウの内容が変更された場合にのみ、ローリングウィンドウは新しい行を出力します。トレーニングパイプラインでローリングウィンドウ特徴量が使用される場合、特定のイベントのTimestampの直前にある固定長ウィンドウ期間を使用して、ソースデータに対して正確な時点の特徴量計算が実行されます。これは、オンライン／オフラインのずれやデータ漏洩を防ぐのに役立ちます。T時点の特徴量は、[T − 期間, T)からイベントを集約します。
class RollingWindow(TimeWindow):
window_duration: datetime.timedelta
delay: Optional[datetime.timedelta] = None
次の表に、ローリングウィンドウのパラメーターを示します。ウィンドウの開始時刻と終了時刻は、これらのパラメーターに基づいて次のように決まります。
- 開始時刻：
evaluation_time - window_duration - delay（含む） - 終了時刻：
evaluation_time - delay(排他的)
パラメーター | 制約 |
|---|---|
| 0以上である必要があります（評価 Timestamp から時間枠を過去にシフトします）。 |
| 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時までの（7日目の午後2時は含まず）すべてのイベントを含みます。 -
window_duration=timedelta(hours=1), delay=timedelta(minutes=30)：これにより、評価時刻の30分前で終わる1時間のルックバックウィンドウが作成されます。午後3時のイベントの場合、これは午後1時30分から午後2時30分までの（午後2時30分は含まず）すべてのイベントを含みます。これは、データ取り込みの遅延に対応するために役立ちます。
タンブリングウィンドウ
タンブリングウィンドウを使用して定義された特徴量の場合、集計は、スライド間隔で進む所定の固定長ウィンドウで計算され、時間を完全に分割する重複しないウィンドウが生成されます。その結果、ソース内の各イベントは正確に1つのウィンドウに寄与します。時刻tの特徴量は、t (排他的) 以前に終了するウィンドウからデータを集計します。ウィンドウはUnixエポックで開始します。
class TumblingWindow(TimeWindow):
window_duration: datetime.timedelta
次の表に、タンブリングウィンドウのパラメーターを示します。
パラメーター | 制約 |
|---|---|
| 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 は Day 0 から Day 4 にわたり、ウィンドウ #2 は Day 5 から Day 9 にわたり、ウィンドウ #3 は Day 10 から Day 14 にわたり、などです。具体的には、ウィンドウ #1 は、Day 0 の00:00:00.00に開始する Timestamp を持つすべてのイベントから、Day 5 の Timestamp00:00:00.00を持つイベントまで（ただし、含まず）を含みます。各イベントは厳密に1つのウィンドウに属します。
スライディングウィンドウ
スライディングウィンドウを使用して定義された特徴量の場合、集計は、スライド間隔で進む事前に決定された固定長ウィンドウで計算され、重複するウィンドウを生成します。ソース内の各イベントは、複数のウィンドウの特徴量集計に寄与できます。時刻tの特徴量は、t (排他的) 以前に終了するウィンドウからデータを集計します。ウィンドウはUnixエポックで開始します。
class SlidingWindow(TimeWindow):
window_duration: datetime.timedelta
slide_duration: datetime.timedelta
次の表に、スライディングウィンドウのパラメーターを示します。
パラメーター | 制約 |
|---|---|
| 0より大きい必要があります。 |
| > 0 かつ < である必要があります。 |
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式で定義された固定スケジュールで実行されます。
from databricks.feature_engineering.entities import CronSchedule
trigger = CronSchedule(
quartz_cron_expression="0 0 * * * ?", # Hourly
timezone_id="UTC",
)
TableTrigger
DeltaTableSourceによってサポートされるColumnSelection特徴量にはTableTriggerを使用します。アップストリームのDeltaテーブルが新しいcommitを受信するたびに、パイプラインが実行されます。
from databricks.feature_engineering.entities import TableTrigger
trigger = TableTrigger()
StreamingMode
StreamSourceによってサポートされる特徴量にはStreamingModeを使用します。パイプラインは、継続的なストリーミングパイプラインとして実行されます。
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の選択
特徴量タイプ | トリガー | 実行するとき |
|---|---|---|
集計（ |
| 固定のcronスケジュールで |
|
| 各ソーステーブルcommit時 |
からの特徴量 |
| 連続ストリーミング |
単一の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 という名前のサフィックスを付けて各フィーチャーを再登録します。
# 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を再構築します。
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します。
ベータ版の特徴量を削除する
移行された機能とそのマテリアライズされたデータが正しいことを確認した後でのみ、ベータ機能とそのマテリアライズを削除してください。削除は元に戻すことはできません。
移行された特徴量を検証した後、各ベータ機能のマテリアライズを削除し、次にベータ機能自体を削除します。
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)