機能ビュー

Important

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

フィーチャービューを使用すると、データソースから特徴を定義および計算できます。特徴は、さまざまなソース (Delta テーブル、Kafka Stream、要求時データ) と計算 (時間枠の集計、単純な列の選択など) を使用して定義できます。このガイドでは、次のワークフローについて説明します。

機能開発 ワークフロー
- create_featureを使用して、モデルのトレーニングとワークフローの提供で使用できる Unity カタログ機能オブジェクトを定義します。
- または、 Feature オブジェクトをローカルで構築し、 register_feature を使用して後で Unity カタログに保持します。ローカルで構築された機能は、登録前に create_training_set で使用できます。
モデルトレーニング ワークフロー
- create_training_setを使用して、機械学習のポイントインタイム集計機能を計算します。フィーチャービューを使用したトレーニングの詳細なドキュメントについては、「フィーチャービューを使用してモデルをトレーニングする」を参照してください。
特徴の具体化と提供の ワークフロー
- create_featureを使用して機能を定義するか、get_featureを使用して取得した後、materialize_featuresを使用して、機能または機能のセットをオフラインストアに具体化して効率的に再利用したり、オンラインサービス用のオンラインストアにしたりすることができます。
- create_training_setを具体化されたビューと共に使用して、オフラインバッチトレーニングデータセットを準備します。

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

必要条件

Databricks Runtime 17.0 ML 以上を実行しているサーバーレスコンピューティングまたはクラシックコンピューティングクラスター。
カスタム Python パッケージをインストールする必要があります。ノートブックを実行するたびに、次のコード行を実行します。
```
%pip install databricks-feature-engineering>=0.16.0
dbutils.library.restartPython()
```

クイックスタートの例

実行可能なクイックスタートノートブックについては、「ノートブックの例」を参照してください。

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    CronSchedule, DeltaTableSource, Feature, AggregationFunction,
    Sum, Avg, ColumnSelection, TableTrigger,
    TumblingWindow, SlidingWindow,
    OfflineStoreConfig, OnlineStoreConfig,
)
from datetime import timedelta

CATALOG_NAME = "main"
SCHEMA_NAME = "feature_store"
TABLE_NAME = "transactions"

# 1. Create data source
source = DeltaTableSource(
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    table_name=TABLE_NAME,
)

# 2. Define features locally (no catalog/schema needed yet)
avg_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), TumblingWindow(window_duration=timedelta(days=30))),
    name="avg_transaction_30d",
)

sum_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), SlidingWindow(window_duration=timedelta(days=7), slide_duration=timedelta(days=1))),
    # name auto-generated: "amount_sum_sliding_7d_1d"
)

fe = FeatureEngineeringClient()

# 3. Explore features with compute_features
feature_df = fe.compute_features(features=[avg_feature, sum_feature])
feature_df.display()

# 4. Create training set using local features
# `labeled_df` should have columns "user_id", "transaction_time", and "target".
training_set = fe.create_training_set(
    df=labeled_df,
    features=[avg_feature, sum_feature],
    label="target",
)
training_set.load_df().display()

# 5. Register features in Unity Catalog
avg_feature = fe.register_feature(
    feature=avg_feature,
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
)
sum_feature = fe.register_feature(
    feature=sum_feature,
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
)

# 6. Or use create_feature for a one-step define-and-register workflow
latest_amount = fe.create_feature(
    source=source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    name="latest_amount",
)

# 7. Train model
with mlflow.start_run():
    training_df = training_set.load_df()

    # training code

    fe.log_model(
        model=model,
        artifact_path="recommendation_model",
        flavor=mlflow.sklearn,
        training_set=training_set,
        registered_model_name=f"{CATALOG_NAME}.{SCHEMA_NAME}.recommendation_model",
    )

# 8. (Optional) Materialize features for serving
# Features must be registered in UC before calling materialize_features
online_config = OnlineStoreConfig(
    catalog_name=CATALOG_NAME,
    schema_name=SCHEMA_NAME,
    table_name_prefix="customer_features_serving",
    online_store_name="customer_features_store",
)

# Aggregation features support CronSchedule or TableTrigger, and support both offline and online configs
fe.materialize_features(
    features=[avg_feature, sum_feature],
    offline_config=OfflineStoreConfig(
        catalog_name=CATALOG_NAME,
        schema_name=SCHEMA_NAME,
        table_name_prefix="customer_features",
    ),
    online_config=online_config,
    trigger=CronSchedule(
        quartz_cron_expression="0 0 * * * ?",  # Hourly
        timezone_id="UTC",
    ),
)

# ColumnSelection features use TableTrigger and only support online config
fe.materialize_features(
    features=[latest_amount],
    online_config=online_config,
    trigger=TableTrigger(),
)

ノートブックの例

Feature Views クイックスタートノートブック

ノートブックを入手

ストリーミング機能

機能値がバッチスケジュールで更新されるのではなく、継続的に更新する必要がある場合はストリーミング機能を活用しましょう。ストリーミング機能とバッチ機能は同じ Feature 構造体、集約関数、トレーニングおよびサービスワークフローを使用します。

ストリーミング機能はオフラインストアには現れません。トレーニングおよびバッチ推論では、Databricksはソースから特徴値を計算します。

ストリーミング機能には以下の要件があります:

online_configを提出しなければなりません。ストリーミング機能は offline_configをサポートしていません。
ストリーミング機能とバッチ機能を1つの通話で組み合わせることはできません materialize_features 。トリガータイプごとに別々のコールを行ってください。
transformation_sql ストリーミング機能には対応していません。
ストリーミングマテリアライゼーションは、パイプライン開始後に到着したレコードのみを処理し、過去のレコードをバックフィルしません。ローリングウィンドウ集計は、最初の完全なデータウィンドウが届いた後にのみ完全な結果を返します。

ストリーミング機能ソースの選択

新鮮度の要件と既存の摂取環境に基づいて供給源を選びましょう:

1秒未満の鮮度が最優先の場合は StreamSource を使いましょう。 StreamSource これらの機能により、P99のエンドツーエンド遅延は200ミリ秒です。まずストリームを設定し、その後 StreamSourceで参照します。ストリームソースは入力としてKafkaをサポートし、学習用にデータの履歴的コピーとして自動的に取り込みDeltaテーブルを維持します。
すでに低遅延のDeltaテーブルへの取り込み経路がある場合は DeltaTableSource を使いましょう。新鮮さは数十秒程度のものになると予想してください。
低遅延の取り込み経路がまだない場合は、Zerobusを使って DeltaTableSource を埋めてください。 Zerobus の取り込みには数十秒程度かかるため、特徴量の鮮度は 1 分未満になると見込まれます。

StreamSourceを使ってストリーミング機能を定義します

StreamSourceは、3 部構成の名前 (catalog.schema.stream_name) で Stream を参照します。 Stream は Unity カタログのセキュリティ保護可能なオブジェクトではありませんが、スコープは Unity カタログスキーマであり、アクセスは Stream のインジェストテーブルによって管理されます。エンティティ、時系列、および関数の定義内の列参照には、読み取る Kafka メッセージのどの部分を示すために、 value. または key. プレフィックスを付ける必要があります。入れ子になったフィールドは、ドット表記 ( value.user.address.city など) を使用してサポートされます。

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

client = FeatureEngineeringClient()

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

feature = Feature(
    name="user_purchase_sum",
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(
        operator=Sum(input="value.amount"),
        time_window=RollingWindow(window_duration=timedelta(hours=1)),
    ),
)

DeltaTableSourceを使ったストリーミング機能を定義する

DeltaTableSource上でストリーミング機能として定義された機能を具現化するには、StreamingModeをトリガーとして渡して、materialize_featuresにします。機能定義は、 DeltaTableSourceに裏付けられたバッチ機能と同じAPIを使用します。デルタテーブルソースは集約および列選択機能をサポートしています。

ソース Delta テーブルでは、delta.enableChangeDataFeed=trueを設定して変更データフィード（CDF）を有効にする必要があります。

以下の例は、デルタテーブルソースを用いた集約特徴を定義し、具体化します。

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

client = FeatureEngineeringClient()

source = DeltaTableSource(
    catalog_name="my_catalog",
    schema_name="my_schema",
    table_name="transactions",
)

feature = client.create_feature(
    catalog_name="my_catalog",
    schema_name="my_schema",
    name="user_purchase_sum",
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(
        operator=Sum(input="amount"),
        time_window=RollingWindow(window_duration=timedelta(hours=1)),
    ),
)

online_config = OnlineStoreConfig(
    catalog_name="my_catalog",
    schema_name="my_schema",
    table_name_prefix="streaming_features",
    online_store_name="my_online_store",
)

client.materialize_features(
    features=[feature],
    online_config=online_config,
    trigger=StreamingMode(),
)

Zerobusで埋められたデルタテーブルを使います

Zerobusで埋められたDeltaテーブルはストリーミング機能ソースとして機能します。 Zerobusは自動的に delta.enableChangeDataFeed=true を設定しません。このプロパティは、ストリーミング機能ソースとして使う前にターゲットDeltaテーブルで手動で設定する必要があります。

ストリーミングソースのフィルター条件

filter_conditionまたはStreamSourceの集約前に行をフィルタリングするためにDeltaTableSourceを使いましょう。

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

ストリーミングソースからのコラム選択

ColumnSelection フィーチャは、各エンティティキーについて、ソース内の最新の値を集約せずにそのまま渡します。トレーニング時には、特徴値はポイントインタイム精度を満たします。

列選択機能にはTTLはありません。オンラインストアから選択された値を削除するには、ソースが選択した列に対してnull値を発行しなければなりません。

from databricks.feature_engineering.entities import ColumnSelection

passenger_count = Feature(
    name="passenger_count",
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=ColumnSelection(column="value.passenger_count"),
)

StreamSourceから入れ子状フィールドへのアクセス

StreamSourceでは、ドット表記(例えばvalue.nested_field.amount)を使ってネストされたJSONフィールドにアクセスできます。サービス時に、要求ペイロードと応答はリーフノード名 (たとえば、amountではなくvalue.amount) を使用します。リーフノード名は、モデルやフィーチャースペック内のすべてのエンティティ名、時系列名、特徴名に対して一意でなければなりません。なぜなら、サービスエンドポイントはリーフ名を使って値をルーティングしているからです。

ストリーミング機能の時間枠

ストリーミング機能では、集計の RollingWindow のみがサポートされます。ローリングウィンドウは、ストリーミングソースのリアルタイムの性質に合わせて、最新のデータを継続的に再計算します。 TumblingWindow および SlidingWindow は、固定履歴間隔でのバッチ計算用に設計されています。

ストリーミング機能のノートブックの例

ストリーミング機能ビューのクイックスタートノートブック

ノートブックを入手

モデルのトレーニングと推論

log_model()、score_batch()、create_training_set()など、フィーチャビューを使用してモデルをトレーニングし、バッチ推論を実行するには、「フィーチャビューを使用してモデルをトレーニングする」を参照してください。

特徴の具体化

機能を定義したら、それらをオフラインまたはオンラインストアに具体化して、トレーニングおよびサービスワークフローで効率的に再利用できます。特徴を具体化した後、CPU モデルサービスを使用してモデルを提供できます。詳細については、「フィーチャビューの具体化」を参照してください。

ベストプラクティス

機能の名前付け

ビジネスクリティカルな機能にはわかりやすい名前を使用します。
チーム間で一貫した名前付け規則に従います。
機能の開発を開始するときは、自動生成された名前を使用します。

時間枠

ウィンドウの境界をビジネスサイクル (毎日、毎週) に合わせます。
ウィンドウが短いほど、最近の傾向がキャプチャされますが、ノイズが多くなる可能性があります。ウィンドウが長いほど、より安定した機能分布が生成されますが、最近の動作の変化を見逃す可能性があります。ユースケースの基になるシグナルの変化速度に基づいて選択します。たとえば、7 日間のウィンドウでは、毎日の変動がスムーズになり、一貫性のあるモデル入力が生成されますが、1 時間のウィンドウは動作の変化にすばやく反応しますが、モデルのパフォーマンスを低下させる分散が生じる可能性があります。分布が変化したときにモデルの精度が低下する場合は、長いウィンドウを使用して入力を安定させます。
タンブリングウィンドウとスライディングウィンドウは、ローリング (連続) ウィンドウよりもスケーラブルです。ほとんどのユースケースのスライディングウィンドウから始めます。

Performance

1 回の materialize_features 呼び出しで同じデータソースの特徴を具体化し、データスキャンを最小限に抑えます。
同じデータソース上の機能に対して同じ粒度 (たとえば、すべての 1 時間またはすべての 1 日のスライド期間) を使用して、具体化時のグループ化を向上させます。

エンティティ列とフィルター条件

同じソーステーブルの機能を使用する場合は、次の決定ガイドを使用します。

異なる集計レベルが必要な場合は、 entity ( create_feature) を使用します。

顧客レベルの機能 (顧客 ごとに 1 行): entity=["customer_id"]
顧客・販売者機能（顧客ごとの複数行）: entity=["customer_id", "merchant_id"]
異なる集計レベルで同じDeltaTableSourceを共有できます。各機能定義で異なるentity値を指定します

同じ集計レベルで行をフィルター処理する必要がある場合は、 filter_condition ( DeltaTableSource) を使用します。

価値の高いトランザクションのみ: filter_condition="amount > 100" (引き続き顧客ごとに集計)
完了した注文のみ: filter_condition="status = 'completed'" (引き続き顧客ごとに集計)

おおまかなやり方： エンティティ値ごとに異なる数の行が変更される場合は、フィーチャ定義で異なる entity 値を使用します。同じ集計に関係する行をフィルター処理するだけの場合は、ソースで filter_condition を使用します。

一般的なパターン

顧客分析

from databricks.feature_engineering.entities import AggregationFunction, Sum, Count, RollingWindow

fe = FeatureEngineeringClient()
features = [
    # Recency: Number of transactions in the last day
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Count(input="transaction_id"), RollingWindow(window_duration=timedelta(days=1)))),

    # Frequency: transaction count over the last 90 days
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Count(input="transaction_id"), RollingWindow(window_duration=timedelta(days=90)))),

    # Monetary: total spend in the last month
    fe.create_feature(catalog_name="main", schema_name="ecommerce", source=transactions,
            entity=["user_id"], timeseries_column="transaction_time",
            function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=30)))),
]

傾向分析

# Compare recent vs. historical behavior
fe = FeatureEngineeringClient()
recent_avg = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

historical_avg = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=7), delay=timedelta(days=7))),
)

季節パターン

# Same day of week, 4 weeks ago
fe = FeatureEngineeringClient()
weekly_pattern = fe.create_feature(
    catalog_name="main", schema_name="ecommerce",
    source=transactions, entity=["user_id"], timeseries_column="transaction_time",
    function=AggregationFunction(Avg(input="amount"), RollingWindow(window_duration=timedelta(days=1), delay=timedelta(weeks=4))),
)

制限事項

エンティティ列と時系列列の名前は、トレーニング (ラベル付き) データセットと、 create_training_set API で使用する場合の特徴の定義の間で一致する必要があります。
トレーニングデータセットの label 列として使用される列名は、 Featureの定義に使用されるソーステーブルには存在しません。
create_feature API では、限られた関数の一覧 (UDF) がサポートされています。サポートされている関数を参照してください。
エンティティ列は、 DATE 型または TIMESTAMP型にすることはできません。
RequestSource では、 ScalarDataType で定義されているスカラーデータ型 (INTEGER、 FLOAT、 BOOLEAN、 STRING、 DOUBLE、 LONG、 TIMESTAMP、 DATE、 SHORT) のみがサポートされます。配列、マップ、構造体などの複合型はサポートされていません。
RequestSource では、集計関数またはタイムウィンドウはサポートされていません。 ColumnSelection関数のみを使用できます。
エンティティ列名、時系列列名、および要求機能列名のセットは、トレーニングセットまたはサービスエンドポイント内のすべてのソースでグローバルに一意である必要があります。

score_batch は、サーバーレスコンピューティングでは成功しない可能性があります。これを回避する場合は、Databricks Runtime 17.0 ML 以降を実行しているクラシックコンピューティングクラスターを使用します。

具体化固有の制限事項については、「制限事項」を参照してください。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-08-14