실시간 피처 서빙 (Online Feature Serving)

-- CDF 활성화 (TRIGGERED/CONTINUOUS 모드에 필수)
ALTER TABLE ml_prod.features.customer_features
SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');

-- 가장 간단한 방법: SQL DDL
CREATE ONLINE TABLE ml_prod.features.customer_features_online
FROM ml_prod.features.customer_features;

-- 동기화 모드를 지정하는 경우
CREATE ONLINE TABLE ml_prod.features.customer_features_online
FROM ml_prod.features.customer_features
WITH (
    SYNC_MODE = 'TRIGGERED'  -- TRIGGERED | CONTINUOUS | SNAPSHOT
);

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Online Table 생성
online_table = w.online_tables.create(
    name="ml_prod.features.customer_features_online",
    spec={
        "source_table_full_name": "ml_prod.features.customer_features",
        "primary_key_columns": ["customer_id"],
        "run_triggered": {}  # TRIGGERED 모드
        # "run_continuously": {}  # CONTINUOUS 모드를 원하는 경우
    }
)

print(f"Online Table 생성 완료: {online_table.name}")
print(f"상태: {online_table.status}")

-- Online Table 동기화 상태 확인
DESCRIBE ONLINE TABLE ml_prod.features.customer_features_online;

# Python으로 상태 확인
status = w.online_tables.get("ml_prod.features.customer_features_online")
print(f"상태: {status.status.detailed_state}")
print(f"마지막 동기화: {status.status.message}")

from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup

fe = FeatureEngineeringClient()

# 피처 조회 정의
feature_lookups = [
    FeatureLookup(
        table_name="ml_prod.features.customer_features",
        lookup_key=["customer_id"],
        feature_names=["avg_transaction_amount", "transaction_count_30d", "account_age_days"]
    ),
    FeatureLookup(
        table_name="ml_prod.features.merchant_features",
        lookup_key=["merchant_id"],
        feature_names=["avg_fraud_rate", "category"]
    ),
]

# 피처 조회가 포함된 학습 데이터셋 생성
training_set = fe.create_training_set(
    df=raw_transactions_df,
    feature_lookups=feature_lookups,
    label="is_fraud",
)

# 모델 학습 후 로깅 (피처 조회 메타데이터가 함께 저장됩니다)
fe.log_model(
    model=trained_model,
    artifact_path="fraud_model",
    flavor=mlflow.sklearn,
    training_set=training_set,
    registered_model_name="ml_prod.models.fraud_detector",
)

# 클라이언트 코드: Primary Key만 전송하면 됩니다
import requests

response = requests.post(
    f"https://{workspace_url}/serving-endpoints/fraud-detector/invocations",
    json={
        "dataframe_records": [
            {
                "customer_id": 12345,
                "merchant_id": "M-9876",
                "transaction_amount": 450.00,
                # ↑ 이 값들만 전송하면 됩니다
                # avg_transaction_amount, transaction_count_30d 등은
                # Online Table에서 자동으로 조회됩니다
            }
        ]
    },
    headers={"Authorization": f"Bearer {token}"}
)

print(response.json())
# {"predictions": [{"is_fraud": 0.87}]}

-- Feature Table 생성
CREATE TABLE IF NOT EXISTS ml_prod.features.customer_features (
    customer_id BIGINT,
    avg_transaction_amount DOUBLE,
    transaction_count_30d INT,
    account_age_days INT,
    preferred_category STRING,
    update_timestamp TIMESTAMP
)
TBLPROPERTIES (
    'delta.enableChangeDataFeed' = 'true'  -- CDF 활성화
);

-- 피처 데이터 적재
INSERT INTO ml_prod.features.customer_features
SELECT
    customer_id,
    AVG(amount) AS avg_transaction_amount,
    COUNT(*) AS transaction_count_30d,
    DATEDIFF(CURRENT_DATE(), MIN(signup_date)) AS account_age_days,
    MODE(category) AS preferred_category,
    CURRENT_TIMESTAMP() AS update_timestamp
FROM ml_prod.gold.transactions
WHERE transaction_date >= CURRENT_DATE() - INTERVAL 30 DAYS
GROUP BY customer_id;

CREATE ONLINE TABLE ml_prod.features.customer_features_online
FROM ml_prod.features.customer_features;

from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup
import mlflow.sklearn
from sklearn.ensemble import GradientBoostingClassifier

fe = FeatureEngineeringClient()

feature_lookups = [
    FeatureLookup(
        table_name="ml_prod.features.customer_features",
        lookup_key=["customer_id"],
    ),
]

training_set = fe.create_training_set(
    df=labeled_transactions_df,   # customer_id + is_fraud 레이블
    feature_lookups=feature_lookups,
    label="is_fraud",
)

training_df = training_set.load_df().toPandas()
X = training_df.drop(columns=["is_fraud", "customer_id"])
y = training_df["is_fraud"]

model = GradientBoostingClassifier(n_estimators=100)
model.fit(X, y)

fe.log_model(
    model=model,
    artifact_path="fraud_model",
    flavor=mlflow.sklearn,
    training_set=training_set,
    registered_model_name="ml_prod.models.fraud_detector",
)

-- 필요한 피처만 포함하여 Online Table 크기를 최소화합니다
CREATE ONLINE TABLE ml_prod.features.customer_features_slim_online
FROM (
    SELECT customer_id, avg_transaction_amount, transaction_count_30d
    FROM ml_prod.features.customer_features
);