Feature Views로 모델 학습

중요합니다

이 기능은 공개 미리보기 단계에 있습니다. 작업 영역 관리자는 미리 보기 페이지에서 이 기능에 대한 액세스를 제어할 수 있습니다. Azure Databricks 미리 보기 관리를 참조하세요.

기능 보기를 사용하면 유추 시 특정 시점의 올바른 기능 계산 및 자동 기능 조회를 사용하여 모델을 학습시킬 수 있습니다. 기능 뷰 정의에 대한 자세한 내용은 기능 보기를 참조하세요.

요구 사항

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 다음 요구 사항을 충족해야 합니다.

  • 기능 정의에서 참조하는 모든 엔터티 열을 포함해야 합니다.
  • 기능 정의에서 참조하는 timeseries 열을 포함해야 합니다.
  • 모든 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
)

매개 변수는 flavor 사용할 MLflow 모델 버전 모듈(예: mlflow.sklearn 또는 mlflow.xgboost.)을 지정합니다.

학습에 TrainingSet 사용되는 기능에 대한 계보를 자동으로 추적하여 기록된 모델입니다. 학습 집합에 RequestSource 특징이 포함되면 필요한 입력으로 MLflow 모델 RequestSource 열이 사양에 추가됩니다. 이렇게 하면 서비스 엔드포인트의 API 스키마가 호출자가 유추 시간에 제공해야 하는 필드를 반영합니다. 자세한 내용은 기능 테이블을 사용하여 모델 학습을 참조하세요.

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

스트리밍 기능을 사용한 교육

Stream을 정의하면 Databricks는 델타 테이블에 스트림 데이터를 쓰는 수집 파이프라인을 관리합니다. create_training_set는 이 수집 테이블에서 읽어 와서 DeltaTableSource의 배치 피처와 마찬가지로 레이블이 지정된 데이터프레임에 대해 시점 조인을 수행합니다. 수집 구성, 백필 및 중복 제거에 대한 자세한 내용은 수집 및 백필을 참조하세요.

예시

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() 깅된 모델은 온라인 저장소에서 기능 조회를 수행하고 두 소스 형식 모두에 대한 모델 서명을 구성합니다.

운영 시점에 원시 모델에 도달하는 데이터

기능 저장소 모델 래퍼는 열을 원시 모델에 전달하기 전에 필터링합니다.

열 형식 내부 모델에 도달합니까?
명시적 기능 출력(ColumnSelection, 집계) 예
RequestSource 기능으로 선언된 열 예
엔티티 열(조회 키) 아니요(기능으로 명시적으로 선언되지 않은 경우)
시계열 열 아니요(기능으로 명시적으로 선언되지 않은 경우)
  • Last updated on