Feature Engineering 개요 — 심화

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 모드
        }
    }
)