Feature Views API リファレンス

アクセス制御

機能は、管理可能な Unity カタログ オブジェクトです。 機能へのアクセスは、 CREATE FEATURE、 READ FEATURE、および Unity カタログの MANAGE 権限によって制御されます。 完全な説明については、 Unity カタログの特権リファレンスを参照してください。

• CREATE FEATURE — スキーマで機能を作成するために必要です。 create_feature親スキーマregister_featureCREATE FEATUREが必要です。 最小特権の原則に従って、スキーマ レベルで CREATE FEATURE を付与します。カタログに付与して、そのカタログ内の任意のスキーマに機能を作成できるようにすることもできます。
• READ FEATURE — 機能とそのデータを読み取るために必要です。 get_feature、 create_training_set、トレーニングまたはサービスのために具体化された特徴データを読み取る場合は、機能に READ FEATURE が必要です。 READ FEATURE スキーマまたはカタログに付与された機能は、そのスキーマまたはカタログに含まれるすべての現在および将来の機能に適用されます。
• MANAGE — 機能のライフサイクルと許可を管理するために必要です。 delete_featureを使用してフィーチャーを削除し、materialize_featuresまたはdelete_materialized_featureを使用してフィーチャーを具体化するには、フィーチャーにMANAGEが必要です。

すべての機能操作では、親カタログと親スキーマのUSE SCHEMAにUSE CATALOGも必要です。 具体化に MANAGE と READ FEATURE 適用する方法については、「 アクセス許可」を参照してください。

機能ビュー API

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

推奨される方法は、 Feature オブジェクトをローカルに構築し、 register_feature を使用して Unity カタログに保持することです。 この 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 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 カタログの機能を 1 つのステップで検証、構築、およびすぐに登録します。 これは、最初に機能をローカルで試す必要がない場合に使用します。

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 (イベント タイムスタンプからの 1d 遅延を含むトランザクションの 30 日間の数)

サポートされている関数

集計関数

ColumnSelection (パススルー)

ColumnSelection は、集計を適用せずにソースから 1 つの列を選択します。 これは、(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 句として機能します。

フィルターは、特徴の計算に必要なデータのスーパーセットを含む大きなソース テーブルを操作する場合に便利であり、これらのテーブルの上に個別のビューを作成する必要性を最小限に抑えます。

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: 変換後の結果の DataFrame のスキーマ 。Spark StructType JSON 形式 ( df.schema.json())。 transformation_sqlが指定されている場合は必須です。 これにより、変換の結果として得られた列名と型がシステムに通知されます。

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

例: 列変換に 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(),
)

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_sqlを使用して直接filter_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_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))),
)

例: Customer-Store レベルの機能

より詳細なレベル (顧客ストアの組み合わせごとに 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 は Stream を参照 します。 Stream には、ストリーミング ソースの接続、認証、スキーマ、およびインジェストの構成が含まれています。 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: Stream の完全な 3 部構成の名前 (たとえば、 "my_catalog.my_schema.my_stream")。
• filter_condition(省略可能): ドットプレフィックス付きの列参照 ("value.event_type = 'purchase'" など) を使用して、集計前にストリーム データに適用される SQL WHERE句。

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) がサポートされます。 配列、マップ、構造体などの複合型はサポートされていません。

要求データのハイドレート方法

モデル シグネチャ

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

入力 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 つのウィンドウに属します。
• スライド ウィンドウは、構成可能なスライド間隔で重なり合うローリング タイム ウィンドウです。

次の図は、その動作を示しています。

ローリング ウィンドウ

ローリング ウィンドウは-date とリアルタイムの集計 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 (を含まない)

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:00 のイベントの場合、これには、0 日目の午後 2 時から 7 日目の午後 2:00 までのすべてのイベントが含まれます。

• window_duration=timedelta(hours=1), delay=timedelta(minutes=30): 評価時間の 30 分前に終了する 1 時間のルックバック ウィンドウが作成されます。 午後 3 時 00 分のイベントの場合、これには午後 1 時 30 分から午後 2 時 30 分までの (ただし含まない) すべてのイベントが含まれます。 これは、データ インジェストの遅延を考慮するのに役立ちます。

タンブリング ウィンドウ

タンブリング ウィンドウを使用して定義された機能の場合、集計は、定められた固定長ウィンドウで計算され、ウィンドウはスライド間隔によって進み、非重複となる時間の完全パーティションを生成します。 その結果、ソース内の各イベントは、1 つのウィンドウに正確に影響します。 時間 t 機能は、 t 以前に終了するウィンドウからのデータを集計します (排他的)。 Windows は Unix エポックから開始します。

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

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

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 日目までです。 Window #1 には、0 日目の 00:00:00.00 から始まるイベントで、5 日目の 00:00:00.00 を含まないすべてのイベントが含まれます。 各イベントは、1 つのウィンドウに属します。

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

スライディング ウィンドウを使用して定義されたフィーチャの場合、集計は、スライド間隔で進む事前に決定された固定長ウィンドウに対して計算され、重なり合うウィンドウが生成されます。 ソース内の各イベントは、複数のウィンドウの機能集約に貢献できます。 時間 t 機能は、 t 以前に終了するウィンドウからのデータを集計します (排他的)。 Windows は Unix エポックから開始します。

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

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

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 日間のウィンドウが重なって作成されます。 例: Window #1 は 0 日目から 4 日目まで、Window #2 は 1 日目から 5 日目まで、Window #3 は 2 日目から 6 日目までです。 各ウィンドウには、開始日の 00:00:00.00 から、終了日に 00:00:00.00 までのイベントが含まれます (含まれません)。 ウィンドウは重なっているため、1 つのイベントが複数のウィンドウに属している可能性があります (この例では、各イベントは最大 5 つの異なるウィンドウに属しています)。

具体化トリガー

具体化パイプラインを実行するタイミングを制御します。 トリガーの種類は、機能の種類によって異なります。

CronSchedule

集計機能 (CronSchedule) にAggregationFunctionを使用します。 パイプラインは、Quartz cron 式によって定義された固定スケジュールで実行されます。

from databricks.feature_engineering.entities import CronSchedule

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

TableTrigger

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

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

トリガーの選択

1 回の materialize_features 呼び出しで異なるトリガーの種類を必要とする機能を具体化することはできません。 代わりに個別の呼び出しを発行します。

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

Feature Views パブリック プレビューでは、CREATE FEATUREとREAD FEATUREの特権によって管理される Unity カタログの最上位のフィーチャー エンティティが導入されており、バージョン 0.16.0 以降databricks-feature-engineering必要です。 ベータ期間中に作成された機能 (バージョン 0.15.0) は Unity カタログ関数として格納され、すべてのパブリック プレビュー機能をサポートしているわけではありません。 長期的なパブリック プレビュー のサポートを受けるために、バージョン 0.16.0 でベータ機能を再作成します。 フィーチャーは、再具体化するだけでなく、削除して再作成する必要があります。

フィーチャの詳細については、「フィーチャ ビュー」を参照してください。

必要な操作

• 0.16.0 にアップグレードします。 これは、パブリック プレビュー機能 (バッチとストリーミング) に必要なクライアント バージョンです。
• 機能を再作成します。 ベータ機能ビューは、すべてのパブリック プレビュー機能をサポートしていないので、再具体化されずに削除および再作成する必要があります。
• ウィンドウが閉じる前に移行します。 既存のベータ機能は、2026 年 7 月 22 日より前に移行する必要があります。

ベータ機能とパブリック プレビュー機能を特定する

パブリック プレビュー機能は、たとえばカタログ エクスプローラーで Unity カタログの フィーチャー オブジェクトとして表示されます。 ベータ機能は、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))

移行された機能を再具体化する

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

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

ベータ機能を削除する

移行された機能を確認したら、各ベータ機能の具体化を削除してから、ベータ機能自体を削除します。

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)