Feature Store 설계와 운영 가이드: Feast 기반 Online/Offline Store 구축·ML 피처 파이프라인 자동화

# feature_store.yaml - Feast 프로젝트 설정
project: fraud_detection
registry: gs://ml-feature-store/registry.db
provider: gcp

offline_store:
  type: bigquery
  dataset: feature_store

online_store:
  type: redis
  connection_string: 'redis-cluster.internal:6379'
  redis_type: redis_cluster

entity_key_serialization_version: 2

# features/fraud_detection.py
from datetime import timedelta
from feast import Entity, FeatureView, Field, FileSource, BigQuerySource
from feast.types import Float32, Int64, String

# Entity 정의 - 피처가 연결되는 대상
user_entity = Entity(
    name="user_id",
    join_keys=["user_id"],
    description="Unique identifier for a user",
)

merchant_entity = Entity(
    name="merchant_id",
    join_keys=["merchant_id"],
    description="Unique identifier for a merchant",
)

# Data Source 정의
user_transactions_source = BigQuerySource(
    name="user_transactions",
    table="ml_data.user_transaction_features",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_timestamp",
)

merchant_stats_source = BigQuerySource(
    name="merchant_stats",
    table="ml_data.merchant_statistics",
    timestamp_field="event_timestamp",
)

# Feature View 정의 - 피처 그룹
user_transaction_features = FeatureView(
    name="user_transaction_features",
    entities=[user_entity],
    ttl=timedelta(days=7),
    schema=[
        Field(name="transaction_count_7d", dtype=Int64),
        Field(name="transaction_amount_avg_7d", dtype=Float32),
        Field(name="transaction_amount_max_7d", dtype=Float32),
        Field(name="unique_merchants_7d", dtype=Int64),
        Field(name="avg_time_between_transactions", dtype=Float32),
    ],
    source=user_transactions_source,
    online=True,
    tags={
        "team": "fraud-detection",
        "version": "v2",
    },
)

merchant_risk_features = FeatureView(
    name="merchant_risk_features",
    entities=[merchant_entity],
    ttl=timedelta(days=30),
    schema=[
        Field(name="chargeback_rate_30d", dtype=Float32),
        Field(name="avg_transaction_amount", dtype=Float32),
        Field(name="total_transactions_30d", dtype=Int64),
        Field(name="risk_score", dtype=Float32),
    ],
    source=merchant_stats_source,
    online=True,
    tags={
        "team": "fraud-detection",
        "version": "v1",
    },
)

# features/services.py
from feast import FeatureService

fraud_detection_service = FeatureService(
    name="fraud_detection_v2",
    features=[
        user_transaction_features,
        merchant_risk_features,
    ],
    tags={
        "model": "fraud_detector_v2",
        "owner": "ml-team",
    },
)

# Feast CLI로 Feature Registry 적용
feast apply

# 배치 Materialization 실행
feast materialize 2026-03-01T00:00:00 2026-03-12T00:00:00

# 증분 Materialization (마지막 실행 이후)
feast materialize-incremental 2026-03-12T00:00:00

# dags/feast_materialization.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

default_args = {
    "owner": "ml-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    "feast_materialization",
    default_args=default_args,
    description="Daily feature materialization pipeline",
    schedule_interval="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["feast", "mlops"],
)

# Feature 소스 데이터 검증
validate_sources = PythonOperator(
    task_id="validate_sources",
    python_callable=lambda: __import__("feast").FeatureStore(
        repo_path="/opt/feast/feature_repo"
    ),
    dag=dag,
)

# Materialization 실행
materialize = BashOperator(
    task_id="materialize_features",
    bash_command="""
        cd /opt/feast/feature_repo && \
        feast materialize-incremental $(date -u +%Y-%m-%dT%H:%M:%S)
    """,
    dag=dag,
)

# Online Store 정합성 검증
validate_online = PythonOperator(
    task_id="validate_online_store",
    python_callable=lambda: print("Validating online store consistency..."),
    dag=dag,
)

validate_sources >> materialize >> validate_online

# feature_store.yaml - Redis 설정
project: fraud_detection
registry: gs://ml-feature-store/registry.db
provider: gcp

online_store:
  type: redis
  connection_string: 'redis-cluster.internal:6379,redis-cluster.internal:6380,redis-cluster.internal:6381'
  redis_type: redis_cluster
  key_ttl_seconds: 604800 # 7일

# feature_store.yaml - DynamoDB 설정
project: fraud_detection
registry: s3://ml-feature-store/registry.db
provider: aws

online_store:
  type: dynamodb
  region: ap-northeast-2
  table_name_template: 'feast_online_{project}_{table}'

# feature_store.yaml - BigQuery 설정
project: fraud_detection
registry: gs://ml-feature-store/registry.db
provider: gcp

offline_store:
  type: bigquery
  dataset: feature_store
  location: asia-northeast3

# training_data.py
from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path="./feature_repo")

# Entity DataFrame - 학습에 사용할 시점과 엔티티
entity_df = pd.DataFrame({
    "user_id": ["user_001", "user_002", "user_003", "user_001"],
    "merchant_id": ["merch_100", "merch_200", "merch_100", "merch_300"],
    "event_timestamp": pd.to_datetime([
        "2026-03-01 10:00:00",
        "2026-03-02 14:30:00",
        "2026-03-03 09:15:00",
        "2026-03-05 16:45:00",
    ]),
    "label": [0, 1, 0, 1],  # 사기 여부
})

# Point-in-Time 정확한 피처 추출
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_transaction_features:transaction_count_7d",
        "user_transaction_features:transaction_amount_avg_7d",
        "user_transaction_features:transaction_amount_max_7d",
        "user_transaction_features:unique_merchants_7d",
        "merchant_risk_features:chargeback_rate_30d",
        "merchant_risk_features:risk_score",
    ],
).to_df()

print(training_df.head())
print(f"Training data shape: {training_df.shape}")

# inference.py
from feast import FeatureStore

store = FeatureStore(repo_path="./feature_repo")

# 실시간 피처 조회
feature_vector = store.get_online_features(
    features=[
        "user_transaction_features:transaction_count_7d",
        "user_transaction_features:transaction_amount_avg_7d",
        "user_transaction_features:unique_merchants_7d",
        "merchant_risk_features:chargeback_rate_30d",
        "merchant_risk_features:risk_score",
    ],
    entity_rows=[
        {"user_id": "user_001", "merchant_id": "merch_100"},
    ],
).to_dict()

print(feature_vector)
# 출력 예시:
# {
#   "user_id": ["user_001"],
#   "merchant_id": ["merch_100"],
#   "transaction_count_7d": [23],
#   "transaction_amount_avg_7d": [45000.5],
#   "unique_merchants_7d": [8],
#   "chargeback_rate_30d": [0.02],
#   "risk_score": [0.15]
# }

# skew_detection.py
import pandas as pd
import numpy as np
from feast import FeatureStore
from scipy import stats

store = FeatureStore(repo_path="./feature_repo")

def detect_training_serving_skew(
    feature_name: str,
    training_values: pd.Series,
    sample_size: int = 1000,
):
    """학습 데이터와 Online Store의 피처 분포를 비교한다."""
    # Online Store에서 샘플링
    online_features = []
    entity_rows = [{"user_id": f"user_{i:04d}"} for i in range(sample_size)]

    online_result = store.get_online_features(
        features=[feature_name],
        entity_rows=entity_rows,
    ).to_df()

    serving_values = online_result[feature_name.split(":")[-1]].dropna()

    # KS 테스트로 분포 비교
    ks_stat, p_value = stats.ks_2samp(
        training_values.dropna(),
        serving_values,
    )

    # PSI (Population Stability Index) 계산
    psi = calculate_psi(training_values.dropna(), serving_values)

    return {
        "feature": feature_name,
        "ks_statistic": ks_stat,
        "p_value": p_value,
        "psi": psi,
        "skew_detected": psi > 0.2 or p_value < 0.05,
    }


def calculate_psi(expected, actual, bins=10):
    """PSI(Population Stability Index)를 계산한다."""
    breakpoints = np.linspace(
        min(expected.min(), actual.min()),
        max(expected.max(), actual.max()),
        bins + 1,
    )

    expected_counts = np.histogram(expected, breakpoints)[0] / len(expected)
    actual_counts = np.histogram(actual, breakpoints)[0] / len(actual)

    # 0 방지
    expected_counts = np.clip(expected_counts, 0.001, None)
    actual_counts = np.clip(actual_counts, 0.001, None)

    psi = np.sum(
        (actual_counts - expected_counts) * np.log(actual_counts / expected_counts)
    )
    return psi

# monitoring/feature_monitor.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import pandas as pd

@dataclass
class FeatureStats:
    feature_name: str
    timestamp: datetime
    mean: float
    std: float
    min_val: float
    max_val: float
    null_rate: float
    unique_count: int
    p99_latency_ms: Optional[float] = None

def compute_feature_stats(df: pd.DataFrame, feature_name: str) -> FeatureStats:
    """피처의 통계 정보를 계산한다."""
    series = df[feature_name]
    return FeatureStats(
        feature_name=feature_name,
        timestamp=datetime.utcnow(),
        mean=series.mean(),
        std=series.std(),
        min_val=series.min(),
        max_val=series.max(),
        null_rate=series.isnull().sum() / len(series),
        unique_count=series.nunique(),
    )

def check_drift_alerts(
    current: FeatureStats,
    baseline: FeatureStats,
    thresholds: dict,
) -> list:
    """피처 드리프트 알림을 확인한다."""
    alerts = []

    # 평균 변화율 확인
    if baseline.mean != 0:
        mean_change = abs(current.mean - baseline.mean) / abs(baseline.mean)
        if mean_change > thresholds.get("mean_change", 0.3):
            alerts.append(
                f"Mean drift detected: {baseline.mean:.4f} -> {current.mean:.4f} "
                f"(change: {mean_change:.2%})"
            )

    # NULL 비율 변화
    null_diff = abs(current.null_rate - baseline.null_rate)
    if null_diff > thresholds.get("null_rate_change", 0.05):
        alerts.append(
            f"Null rate change: {baseline.null_rate:.4f} -> {current.null_rate:.4f}"
        )

    # 범위 이상
    if current.max_val > baseline.max_val * thresholds.get("max_multiplier", 2.0):
        alerts.append(
            f"Max value anomaly: {current.max_val} (baseline max: {baseline.max_val})"
        )

    return alerts

# Materialization 상태 확인
feast materialize-incremental 2026-03-12T00:00:00 --verbose

# 특정 Feature View만 재실행
feast materialize-incremental 2026-03-12T00:00:00 \
  --feature-views user_transaction_features

# Online Store의 피처 신선도 확인
feast feature-views list

# 피처 신선도 확인 스크립트
from feast import FeatureStore
from datetime import datetime, timedelta

store = FeatureStore(repo_path="./feature_repo")

# Feature View별 최종 Materialization 시간 확인
for fv in store.list_feature_views():
    if fv.materialization_intervals:
        last_mat = fv.materialization_intervals[-1]
        staleness = datetime.utcnow() - last_mat.end_date
        if staleness > timedelta(hours=24):
            print(f"ALERT: {fv.name} is stale by {staleness}")
        else:
            print(f"OK: {fv.name} last materialized at {last_mat.end_date}")
    else:
        print(f"WARNING: {fv.name} has never been materialized")