Melatih model dengan Feature Views

Feature Views memungkinkan Anda melatih model dengan perhitungan fitur yang akurat sesuai titik waktu dan pencarian fitur otomatis saat inferensi. Untuk informasi tentang menentukan Tampilan Fitur, lihat Tampilan Fitur.

Persyaratan

Metode API

create_training_set()

Setelah Anda membuat Tampilan Fitur, langkah selanjutnya adalah membuat data pelatihan untuk model Anda. Untuk melakukan ini, teruskan himpunan data berlabel ke create_training_set, yang secara otomatis memastikan komputasi akurat titik waktu dari setiap nilai fitur.

Contohnya:

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

Panggil TrainingSet.load_df untuk menggabungkan data pelatihan asli dengan fitur yang dihitung secara dinamis pada titik waktu tertentu.

Argumen df harus memenuhi persyaratan berikut:

  • Harus berisi semua kolom entitas yang dirujuk oleh definisi fitur.
  • Harus berisi kolom timeseries yang dirujuk oleh definisi fitur.
  • Harus berisi semua kolom yang dideklarasikan dalam skema apa pun RequestSource . Jenis divalidasi terhadap skema yang dideklarasikan. Ketidakcocokan menimbulkan kesalahan (tidak ada transmisi implisit).
  • Harus berisi kolom label.
  • Kumpulan nama kolom entitas, nama kolom timeseries, dan nama kolom fitur permintaan harus unik secara global di semua sumber.

Ketepatan berdasarkan waktu tertentu: Untuk penggabungan dan fitur yang didukung oleh sumber tabel, fitur dihitung hanya menggunakan data sumber yang tersedia sebelum waktu setiap baris, untuk mencegah kebocoran data masa depan ke dalam pelatihan model. Untuk fitur RequestSource, nilai langsung diambil dari baris DataFrame yang berlabel.

log_model()

Gunakan MLflow untuk mencatat model dengan metadata fitur untuk pelacakan silsilah data dan pencarian fitur otomatis selama inferensi:

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
)

Parameter flavor menentukan modul ragam model MLflow untuk digunakan, seperti mlflow.sklearn atau mlflow.xgboost.

Model yang dicatat menggunakan TrainingSet secara otomatis melacak silsilah ke fitur yang digunakan dalam pelatihan. Saat set pelatihan menyertakan RequestSource fitur, kolom RequestSource ditambahkan ke signature model MLflow sebagai masukan yang diperlukan. Ini memastikan skema API dari titik akhir penyajian mencerminkan bidang-bidang yang harus disediakan oleh pemanggil pada waktu inferensi. Untuk informasi lebih lanjut, lihat Melatih model dengan tabel fitur.

Untuk FeatureViewSource, daftarkan fitur turunan dan fitur upstream-nya sebelum mencatat model. Input permintaan yang dibutuhkan oleh dependensi transitif juga diperlukan pada waktu inferensi. Lihat Dependensi UDF kustom untuk persyaratan paket model.

score_batch()

Lakukan inferensi batch dengan pencarian fitur otomatis:

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

score_batch menggunakan metadata fitur yang disimpan dengan model untuk secara otomatis menghitung fitur yang tepat pada titik waktu tertentu untuk inferensi, agar konsisten dengan pelatihan. Untuk informasi lebih lanjut, lihat Melatih model dengan tabel fitur.

Contoh alur kerja

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

Pelatihan dengan fitur RequestSource

Saat model Anda memerlukan data yang disediakan pada waktu inferensi (seperti detail transaksi dari panggilan API), gunakan fitur RequestSource dan fitur berbasis tabel. Selama pelatihan, kolom RequestSource diekstraksi dari DataFrame yang diberi label.

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

Transformasi nilai permintaan dengan CustomUDF

Untuk mengubah data permintaan, gunakan CustomUDF sebagai pengganti ColumnSelection. Contoh ini menggunakan fitur transaction log-transform:

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

Hasilnya mencakup kolom transaction_amount dan label asli, serta log_transaction_amount. UDF membaca transaction_amount dari setiap baris. Kolom permintaan yang diperlukan untuk komputasi tidak dapat dicantumkan di exclude_columns.

Berlatih dengan fitur FeatureViewSource

FeatureViewSource memungkinkan a CustomUDF mengonsumsi output fitur lain, termasuk fitur turunan. Oper output yang Anda inginkan ke create_training_set. Anda tidak perlu mencantumkan dependensi perantaranya.

Untuk setiap baris input, Azure Databricks menyelesaikan grafik ketergantungan penuh:

  1. Menghitung fitur upstream yang didukung tabel menggunakan kunci entitas, timestamp, dan definisi jendela. Perangkat ini menggunakan materialisasi offline yang kompatibel jika tersedia.
  2. Membaca kolom permintaan yang diperlukan dari DataFrame input dan mengevaluasi fitur yang didukung permintaan.
  3. Mengevaluasi fitur turunan berdasarkan urutan ketergantungan, sehingga setiap UDF menerima hasil hulunya.

Fitur turunan tidak menambahkan jendela waktu lain atau pencarian pada titik waktu tertentu. Fitur hulu mempertahankan semantik waktunya masing-masing. DataFrame harus berisi kolom entitas, timestamp, dan permintaan yang dibutuhkan oleh upstream tersebut, bahkan ketika hanya fitur turunan akhir yang diminta.

Misalnya, gunakan fitur margin terdaftar, yang menggabungkan revenue_sum_7d dan 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()

Hasilnya mengandung label dan margin. Fitur pendapatan dan biaya dihitung tetapi tidak dikembalikan sebagai kolom tambahan. Untuk menyertakan pendapatan dalam data pelatihan, ambil dengan revenue = fe.get_feature(full_name="main.ecommerce.revenue_sum_7d") dan teruskan features=[margin, revenue]. Hal ini juga berlaku untuk rantai bertingkat: meminta fitur akhir tidak akan mengembalikan setiap keluaran antara.

Anda dapat menggabungkan fitur yang didukung permintaan, didukung tabel, dan fitur turunan dalam daftar yang sama features . Untuk menggabungkan nilai-nilai mereka di dalam satu UDF, representasikan nilai permintaan sebagai fitur dan referensikan bersama fitur yang didukung tabel dalam sebuah FeatureViewSource.

Untuk eksperimen, konstruksi objek lokal Feature , termasuk graf hulunya, tanpa mendaftarkannya. Gunakan create_training_set, opsional dengan label=None, untuk memeriksa hasilnya. compute_features tidak mendukung RequestSource atau FeatureViewSource.

Note

Batas lima panggilan UDF Unity Catalog per kueri juga berlaku untuk kueri pelatihan. Menghitung panggilan UDF yang dibutuhkan oleh graf ketergantungan penuh, bukan hanya fitur yang diminta sebagai output. Batas kueri ini terpisah dari batas kedalaman grafik.

Ketergantungan UDF khusus

Untuk komputasi offline, deklarasikan paket Python dalam klausa UDF ENVIRONMENTUnity Catalog. Menginstal paket di notebook saja tidak menginstalnya di lingkungan UDF.

Untuk penyajian model, berikan juga paket yang diperlukan secara eksplisit ke log_model. Baik UDF ENVIRONMENT maupun spesifikasi fitur bernama yang berisi Tampilan Fitur tidak secara otomatis menyediakan persyaratan model ini. Sertakan dependensi yang dibutuhkan oleh UDF hulu serta output fitur yang diminta.

Setelah melatih model scikit-learn pada transaction_training_set.load_df(), catat model tersebut menggunakan data pelatihan yang sama. Sertakan NumPy dan paket pencarian yang kompatibel:

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

Endpoint yang melayani seharusnya secara otomatis mengambil databricks-feature-lookup versi 1.15.0 atau lebih baru, yang mendukung komputasi UDF Unity Catalog sesuai permintaan untuk fitur RequestSource. Jaga agar versi paket UDF Anda konsisten antara lingkungan offline dan lingkungan penyajian untuk menghindari perbedaan nilai yang dihitung. Untuk endpoint Feature Serving tanpa model, deklarasikan paket di create_feature_spec sebagai gantinya. Lihat Menambahkan dependensi Python.

Pelatihan dengan fitur streaming

Saat Anda menentukan Aliran, Databricks mengelola alur penyerapan yang menulis aliran data ke dalam tabel Delta. create_training_set membaca data dari tabel ingesti ini dan melakukan join point-in-time terhadap DataFrame berlabel Anda, sama seperti fitur batch dari DeltaTableSource. Untuk detail mengenai konfigurasi ingesti, backfill, dan deduplikasi, lihat Ingestion and backfill.

Example

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

Mencampur fitur batch dan streaming

Fitur batch dan streaming dapat digunakan bersama-sama dalam set pelatihan dan model yang sama. Pada waktu penyajian, fitur batch dicari dari toko offline atau online, dan fitur streaming dicari dari toko online.

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

Model yang dicatat dengan log_model() melakukan pencarian fitur dari toko online dan mengonfigurasi tanda tangan model untuk kedua jenis sumber.

Apa yang mencapai model mentah pada waktu penyajian

Pembungkus model Feature Store memfilter kolom sebelum meneruskannya ke model mentah:

Tipe kolom Apakah sudah mencapai model inti?
Keluaran fitur eksplisit (ColumnSelection, agregasi) Yes
RequestSource kolom dideklarasikan sebagai fitur Yes
Kolom entitas (kunci pencarian) Tidak (kecuali secara eksplisit dinyatakan sebagai fitur)
Kolom deret waktu Tidak (kecuali secara eksplisit dinyatakan sebagai fitur)
  • Terakhir diperbarui pada