フィーチャー ビューを使用してモデルをトレーニングする

特徴ビューを使用すると、ポイントインタイムの正しい特徴計算と推論時の自動特徴検索を使用してモデルをトレーニングできます。 フィーチャー ビューの定義については、「フィーチャ ビュー」を参照してください。

必要条件

• フィーチャは Feature View として作成する必要があります。 「 フィーチャー ビュー」を参照してください。
• CustomUDFおよびFeatureViewSource要件については、Feature Views APIリファレンスを参照してください。

API メソッド

create_training_set()

フィーチャー ビューを作成した後の次の手順は、モデルのトレーニング データを作成することです。 これを行うには、ラベル付きデータセットを 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

TrainingSet.load_dfを呼び出して、元のトレーニング データをポイントインタイムで動的に計算された特徴と結合します。

df引数は、次の要件を満たしている必要があります。

• 機能定義によって参照されるすべてのエンティティ列を含める必要があります。
• 機能定義によって参照される時系列列を含める必要があります。
• 任意の RequestSource スキーマで宣言されているすべての列を含む必要があります。 型は、宣言されたスキーマに対して検証されます。 型不一致が発生するとエラーとなります (暗黙の型変換は行われません)。
• ラベル列を含める必要があります。
• エンティティ列名、時系列列名、および要求機能列名のセットは、すべてのソースでグローバルに一意である必要があります。

ポイントインタイムの正確性: テーブル ソースによってサポートされる集計および ColumnSelection 機能の場合、特徴は、モデル トレーニングへの将来のデータ漏洩を防ぐために、各行のタイムスタンプの前に使用可能なソース データのみを使用して計算されます。 RequestSource機能の場合、値はラベル付けされた DataFrame 行から直接取得されます。

log_model()

MLflow を使用して、系列追跡と推論中の自動特徴検索の特徴メタデータを含むモデルをログに記録します。

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
    extra_pip_requirements: Optional[List[str]] = None, # Optional: Additional serving dependencies
)

flavor パラメーターは、やmlflow.sklearnなど、使用する mlflow.xgboost フレーバー モジュールを指定します。

TrainingSetでログに記録されたモデルは、トレーニングで使用される機能の系列を自動的に追跡します。 トレーニング セットに RequestSource 機能が含まれている場合、 RequestSource 列は必要な入力として MLflow モデルシグネチャに追加されます。 これにより、呼び出し元が推論時に指定する必要があるフィールドが、サービス エンドポイントの API スキーマに反映されるようになります。 詳細については、「 特徴テーブルを使用してモデルをトレーニングする」を参照してください。

FeatureViewSourceの場合、モデルをログする前に派生した特徴とその上流の特徴を登録してください。 推移依存関係に必要な要求入力も推論時に必要とされます。 モデルパッケージの要件については 「カスタムUDF依存関係 」を参照してください。

score_batch()

自動機能検索を使用してバッチ推論を実行します。

FeatureEngineeringClient.score_batch(
    model_uri: str,                           # URI of logged model
    df: DataFrame,                            # DataFrame with entity keys and timestamps
) -> DataFrame

score_batch では、モデルと共に格納された特徴メタデータを使用して、推論のためのポイントインタイムの正しい特徴を自動的に計算し、トレーニングとの一貫性を確保します。 詳細については、「 特徴テーブルを使用してモデルをトレーニングする」を参照してください。

ワークフローの例

import mlflow
from databricks.feature_engineering import FeatureEngineeringClient
from sklearn.ensemble import RandomForestClassifier

fe = FeatureEngineeringClient()

# Assume features are registered in UC
# labeled_df should have columns "user_id", "transaction_time", and "is_fraud"

# 1. Create training set using Feature Views
training_set = fe.create_training_set(
    df=labeled_df,
    features=features,
    label="is_fraud",
)

# 2. Load training data with computed features
training_df = training_set.load_df()
X = training_df.drop("is_fraud").toPandas()
y = training_df.select("is_fraud").toPandas().values.ravel()

# 3. Train model
model = RandomForestClassifier().fit(X, y)

# 4. Log model with feature metadata
with mlflow.start_run():
    fe.log_model(
        model=model,
        artifact_path="fraud_model",
        flavor=mlflow.sklearn,
        training_set=training_set,
        registered_model_name="main.ecommerce.fraud_model",
    )

# 5. Batch scoring with automatic feature lookup
# inference_df must contain the same entity and timeseries columns
# used during training. Features are automatically computed.
predictions = fe.score_batch(
    model_uri="models:/main.ecommerce.fraud_model/1",
    df=inference_df,
)
predictions.display()

RequestSource 機能を使用したトレーニング

モデルで推論時に提供されるデータ (API 呼び出しからのトランザクションの詳細など) が必要な場合は、テーブルに基づく機能と共に RequestSource 機能を使用します。 トレーニング中、 RequestSource 列はラベル付けされた DataFrame から抽出されます。

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

fe = FeatureEngineeringClient()

# RequestSource provides transaction data at inference time
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),
    ]
)

delta_source = DeltaTableSource(
    catalog_name="catalog",
    schema_name="schema",
    table_name="vendor_data",
)

# A column selection feature from the request source (pass-through)
latest_transaction_amount = Feature(
    source=request_source,
    function=ColumnSelection("transaction_amount"),
    name="latest_transaction_amount",
)

# A lookup feature from a delta table
vendor_category = Feature(
    source=delta_source,
    function=ColumnSelection("vendor_category"),
    entity=["vendor_id"],
    timeseries_column="transaction_time",
    name="vendor_category",
)

# labels_df must contain: transaction_id, transaction_time, vendor_id,
# transaction_amount, and the label column.
ts = fe.create_training_set(
    df=labels_df,
    features=[latest_transaction_amount, vendor_category],
    label="is_fraud",
    exclude_columns=["card_id"],
)

import mlflow
from sklearn.ensemble import RandomForestClassifier

with mlflow.start_run():
    training_df = ts.load_df().toPandas()
    X = training_df.drop(columns=["is_fraud"])
    y = training_df["is_fraud"]
    model = RandomForestClassifier().fit(X, y)

    # log_model() adds RequestSource columns to the MLflow model signature
    fe.log_model(
        model=model,
        artifact_path="fraud_model",
        flavor=mlflow.sklearn,
        training_set=ts,
        registered_model_name="catalog.schema.fraud_model",
    )

CustomUDFで要求値を変換する

リクエストデータを変換するには、ColumnSelectionではなくCustomUDFを使いましょう。 この例では トランザクションのログ変換機能を使います:

log_transaction_amount = fe.get_feature(
    full_name="main.ecommerce.log_transaction_amount"
)
request_df = spark.createDataFrame(
    [(0.0, 0), (99.0, 1)],
    "transaction_amount DOUBLE, label INT",
)

transaction_training_set = fe.create_training_set(
    df=request_df,
    features=[log_transaction_amount],
    label="label",
)
transaction_training_set.load_df().show()

結果には元の transaction_amount 列と label 列に加え log_transaction_amountが含まれています。 UDFは各行から transaction_amount を読み取ります。 計算に必要なリクエスト列は exclude_columnsで記載できません。

FeatureViewSource 機能を使ってトレーニングする

FeatureViewSource CustomUDFが派生特徴を含む他の特徴出力を消費します。 必要な出力を create_training_set に渡します。 中間依存関係をリストアップする必要はありません。

各入力行に対して、Azure Databricksは完全な依存関係グラフを解決します:

1. エンティティキー、タイムスタンプ、ウィンドウ定義を用いてテーブルバック型のアップストリーム特徴を計算します。 利用可能な場合、互換性のあるオフラインの物質化を使用します。
2. 入力データフレームから必要なリクエストカラムを読み取り、リクエストバック機能を評価する。
3. 派生特徴を依存順で評価し、各UDFが上流の結果を受け取るようにします。

派生した特徴は別の時間窓や時点検索を導入しません。 その上流の特徴は独自の時間の意味を保持しています。 DataFrameは、最終派生機能のみが要求される場合でも、上流に必要なエンティティ、タイムスタンプ、リクエストカラムを含まなければなりません。

例えば、revenue_sum_7dとcost_sum_7dを組み合わせた登録マージン機能を使うことができます。

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()
margin = fe.get_feature(full_name="main.ecommerce.margin")

# labeled_df contains customer_id, event_time, and label.
training_set = fe.create_training_set(
    df=labeled_df,
    features=[margin],
    label="label",
    exclude_columns=["customer_id", "event_time"],
)
training_df = training_set.load_df()

結果は label と marginを含みます。 収益およびコストの特徴は計算されますが、追加の列として返されることはありません。 収益をトレーニングデータに含めるには、revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") で取得し、features=[margin, revenue] を渡します。 これは多層チェーンにも当てはまります。最終的な特徴を要求しても、すべての中間出力が返されるわけではありません。

リクエストバック、テーブルバック、派生機能を同じ features リスト内で組み合わせることができます。 1つのUDF内でそれらの値を組み合わせるには、リクエスト値を特徴として表現し、テーブル裏面の特徴と共に FeatureViewSourceで参照します。

実験のためには、局所 Feature オブジェクトを登録せずにその上流のグラフを含めて構築します。 create_training_setを使い、オプションでlabel=Noneを付けて結果を確認できます。 compute_features RequestSourceやFeatureViewSourceをサポートしていません。

カスタムUDF依存関係

オフライン計算の場合は、Unity Catalog UDFの ENVIRONMENT clauseでPythonパッケージを宣言してください。 ノートブックにパッケージをインストールするだけではUDF環境にはインストールされません。

モデルサービングのために、必要なパッケージも明示的に log_modelに渡してください。 UDFの ENVIRONMENT も、フィーチャービューを含む名前付き特徴仕様も、これらのモデル要件を自動的に提供するものではありません。 上流のUDFに必要な依存関係や、要求された機能出力も含めてください。

transaction_training_set.load_df()でscikit-learnモデルをトレーニングした後、同じトレーニングセットでログを記録します。 NumPyと対応するルックアップパッケージを含めてください:

import mlflow

fe.log_model(
    model=model,
    artifact_path="transaction_model",
    flavor=mlflow.sklearn,
    training_set=transaction_training_set,
    registered_model_name="main.ecommerce.transaction_model",
    extra_pip_requirements=[
        "numpy==1.26.4",
        "databricks-feature-lookup>=1.15.0",
    ],
)

エンドポイントの提供元は、databricks-feature-lookup バージョン 1.15.0 以降を自動的に採用する必要があります。このバージョンでは、RequestSource 機能に対するオンデマンドの Unity Catalog UDF コンピューティングがサポートされています。 UDFパッケージのバージョンはオフライン環境とサービス環境間で一貫性を保ち、計算値の差を避けましょう。 モデルを持たないFeature Servingエンドポイントの場合は、代わりにパッケージを宣言 create_feature_spec 。 詳細は「Python依存関係の追加」を参照してください。

ストリーミング機能を使用したトレーニング

Stream を定義すると、Databricks は、ストリーム データを Delta テーブルに書き込むインジェスト パイプラインを管理します。 create_training_set は、このインジェスト テーブルから読み取り、 DeltaTableSourceのバッチ機能と同様に、ラベル付けされた DataFrame に対してポイントインタイム結合を実行します。 インジェス設定、バックフィル、重複除去の詳細については、「 インジェスションとバックフィル」を参照してください。

例

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

fe = FeatureEngineeringClient()

# Define a streaming feature
stream_source = StreamSource(full_name="my_catalog.my_schema.my_stream")

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

# Create training set — reads from the ingestion table
# labeled_df must contain "user_id", "event_time", and label columns.
# Entity and timeseries columns use leaf node names (not value. prefixes).
training_set = fe.create_training_set(
    df=labeled_df,
    features=[streaming_feature],
    label="is_fraud",
)

training_df = training_set.load_df()

バッチ機能とストリーミング機能の混在

バッチ機能とストリーミング機能は、同じトレーニング セットとモデルで一緒に使用できます。 サービス時には、バッチ機能はオフラインまたはオンライン ストアから検索され、ストリーミング機能はオンライン ストアから検索されます。

training_set = fe.create_training_set(
    df=labeled_df,
    features=[batch_feature, streaming_feature],
    label="is_fraud",
)

log_model()でログに記録されたモデルは、オンライン ストアから機能検索を実行し、両方のソースの種類のモデル署名を構成します。

サービング時に RAW モデル 渡されるもの

Feature Store モデル ラッパーは、列を未加工モデルに渡す前にフィルター処理します。