Feature Engineering 개요

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# 피처 계산
customer_features_df = spark.sql("""
    SELECT
        customer_id,
        COUNT(*) AS total_orders,
        SUM(amount) AS lifetime_revenue,
        AVG(amount) AS avg_order_value,
        MAX(order_date) AS last_order_date,
        DATEDIFF(CURRENT_DATE(), MAX(order_date)) AS days_since_last_order,
        COUNT(DISTINCT product_category) AS unique_categories
    FROM silver_orders
    GROUP BY customer_id
""")

# Feature Table로 등록
fe.create_table(
    name="catalog.schema.customer_features",
    primary_keys=["customer_id"],
    df=customer_features_df,
    description="고객 행동 기반 피처"
)

# 새로운 피처 데이터로 업데이트
fe.write_table(
    name="catalog.schema.customer_features",
    df=updated_features_df,
    mode="merge"  # 기존 데이터와 병합 (upsert)
)

from databricks.feature_engineering import FeatureLookup

# 학습 데이터에 피처를 자동으로 결합
training_set = fe.create_training_set(
    df=labels_df,  # 라벨만 있는 데이터 (customer_id, is_churned)
    feature_lookups=[
        FeatureLookup(
            table_name="catalog.schema.customer_features",
            lookup_key="customer_id"
        ),
        FeatureLookup(
            table_name="catalog.schema.product_features",
            lookup_key="product_id",
            feature_names=["avg_rating", "return_rate"]  # 특정 피처만 선택
        )
    ],
    label="is_churned"
)

training_df = training_set.load_df()
# → customer_id, is_churned + total_orders, lifetime_revenue, avg_order_value, ... + avg_rating, return_rate

training_set = fe.create_training_set(
    df=labels_df,
    feature_lookups=[
        FeatureLookup(
            table_name="catalog.schema.customer_features",
            lookup_key="customer_id",
            timestamp_lookup_key="event_timestamp"  # 이 시점 기준으로 조회
        )
    ],
    label="is_churned"
)

# 기존 방식 (Workspace Feature Store) — deprecated
from databricks.feature_store import FeatureStoreClient
fs = FeatureStoreClient()  # ❌ 더 이상 권장하지 않음

# 현재 방식 (Unity Catalog Feature Engineering)
from databricks.feature_engineering import FeatureEngineeringClient
fe = FeatureEngineeringClient()  # ✅ 권장

labels_df:
  customer_id | event_timestamp | is_churned
  C001        | 2025-03-15      | True
  C001        | 2025-01-10      | False

Feature Table (customer_features):
  customer_id | feature_timestamp | total_orders | lifetime_revenue
  C001        | 2025-01-01       | 10           | 500,000
  C001        | 2025-02-01       | 15           | 750,000
  C001        | 2025-03-01       | 18           | 900,000

Point-in-Time Lookup 결과:
  C001 | 2025-03-15 → total_orders=18, lifetime_revenue=900,000  (3월 1일 기준)
  C001 | 2025-01-10 → total_orders=10, lifetime_revenue=500,000  (1월 1일 기준)
  ← 각 이벤트 시점 기준으로 "과거의 피처 값"만 사용

-- Feature Table 접근 권한 관리
GRANT SELECT ON TABLE catalog.ml.customer_features TO `ml_engineers`;
GRANT SELECT, MODIFY ON TABLE catalog.ml.customer_features TO `feature_team`;

-- 피처 태그 지정
ALTER TABLE catalog.ml.customer_features
SET TAGS ('domain' = 'customer', 'pii' = 'false', 'freshness_sla' = 'daily');

-- 컬럼별 태그 (민감 정보 표시)
ALTER TABLE catalog.ml.customer_features
ALTER COLUMN email SET TAGS ('pii' = 'true');

-- 피처 리니지 확인
SELECT * FROM system.lineage.table_lineage
WHERE target_table_full_name = 'catalog.ml.customer_features';

# Unity Catalog에서 피처 테이블 검색
from databricks.sdk import WorkspaceClient
w = WorkspaceClient()

# 태그 기반 검색: 'customer' 도메인의 피처 테이블 찾기
tables = w.tables.list(catalog_name="catalog", schema_name="ml")
for t in tables:
    if t.table_type == "MANAGED" and "feature" in t.name:
        print(f"{t.full_name}: {t.comment}")

# 피처 테이블의 마지막 업데이트 시간 확인
last_update = spark.sql("""
    DESCRIBE HISTORY catalog.ml.customer_features
    LIMIT 1
""").select("timestamp").collect()[0][0]

from datetime import datetime, timedelta
staleness = datetime.now() - last_update

if staleness > timedelta(hours=24):
    # Alert 발생: 피처가 24시간 이상 업데이트되지 않음
    print(f"⚠️ Feature staleness alert: {staleness}")

-- Feature Table에 모니터 설정
CREATE MONITOR catalog.ml.customer_features
WITH (
  SCHEDULE = CRON '0 0 * * *',  -- 매일 체크
  BASELINE_TABLE = 'catalog.ml.customer_features_baseline'
);

-- 모니터링 결과 확인: 데이터 드리프트 감지
SELECT * FROM catalog.ml.customer_features_profile_metrics
WHERE log_type = 'DATA_DRIFT'
  AND drift_detected = true;

# 고객의 모든 피처를 하나의 테이블에
fe.create_table(
    name="catalog.ml.customer_features_wide",
    primary_keys=["customer_id"],
    timestamp_keys=["feature_date"],  # Point-in-Time용
    df=wide_features_df,  # 200+ 컬럼
    description="고객 통합 피처 테이블 (행동, 인구통계, 거래, 상호작용 피처 포함)"
)

# 여러 도메인별 Feature Table에서 조합
training_set = fe.create_training_set(
    df=labels_df,
    feature_lookups=[
        FeatureLookup(table_name="catalog.ml.customer_demographics", lookup_key="customer_id"),
        FeatureLookup(table_name="catalog.ml.customer_transactions", lookup_key="customer_id",
                      timestamp_lookup_key="event_date"),
        FeatureLookup(table_name="catalog.ml.customer_interactions", lookup_key="customer_id",
                      timestamp_lookup_key="event_date"),
        FeatureLookup(table_name="catalog.ml.product_features", lookup_key="product_id"),
    ],
    label="is_churned"
)

# Liquid Clustering 적용
spark.sql("""
    ALTER TABLE catalog.ml.customer_features
    CLUSTER BY (customer_id)
""")

# 특정 피처만 선택 (Column Pruning)
FeatureLookup(
    table_name="catalog.ml.customer_features",
    lookup_key="customer_id",
    feature_names=["total_orders", "lifetime_revenue", "avg_order_value"]  # 필요한 것만
)

# Online Table 생성 (Continuous 동기화)
w.online_tables.create(
    name="catalog.ml.customer_features_online",
    spec={
        "source_table_full_name": "catalog.ml.customer_features",
        "primary_key_columns": ["customer_id"],
        "run_continuously": {
            "triggered": False  # True면 Triggered 모드
        }
    }
)