ML Feature Store 구축과 운영 가이드: Feast 기반 실시간 피처 서빙

# 학습 시 (오프라인) - SQL로 피처 계산
train_features = pd.read_sql("""
    SELECT user_id,
           AVG(purchase_amount) AS avg_purchase_30d,
           COUNT(*) AS purchase_count_30d,
           MAX(purchase_amount) AS max_purchase_30d
    FROM transactions
    WHERE event_timestamp BETWEEN '2026-01-01' AND '2026-02-01'
    GROUP BY user_id
""", conn)

# 서빙 시 (온라인) - 애플리케이션 코드로 피처 계산
# 30일 윈도우 계산 로직이 미묘하게 다를 수 있음!
def get_user_features(user_id):
    txns = redis_client.lrange(f"user_txns:{user_id}", 0, -1)
    amounts = [float(t['amount']) for t in txns]
    return {
        'avg_purchase_30d': sum(amounts) / len(amounts),
        'purchase_count_30d': len(amounts),
        'max_purchase_30d': max(amounts)
    }

+-------------------+       +--------------------+       +------------------+
|   Data Sources    |       |   Feature Store    |       |   Consumers      |
|                   |       |                    |       |                  |
|  - Kafka          | ----> |  Offline Store     | ----> |  Model Training  |
|  - Database       |       |  (BigQuery/S3)     |       |  (Batch)         |
|  - Stream         |       |                    |       |                  |
|  - Data Lake      | ----> |  Online Store      | ----> |  Model Serving   |
|                   |       |  (Redis/DynamoDB)  |       |  (Real-time)     |
|                   |       |                    |       |                  |
|                   |       |  Feature Registry  | ----> |  Feature         |
|                   |       |  (Metadata)        |       |  Discovery       |
+-------------------+       +--------------------+       +------------------+

# 기본 설치
pip install feast

# Redis 온라인 스토어 + PostgreSQL 오프라인 스토어
pip install 'feast[redis,postgres]'

# AWS 환경 (DynamoDB + Redshift)
pip install 'feast[aws]'

# GCP 환경 (Datastore + BigQuery)
pip install 'feast[gcp]'

# 프로젝트 생성
feast init my_feature_store
cd my_feature_store

# 프로젝트 구조 확인
tree .
# .
# |-- feature_store.yaml
# |-- features.py
# |-- __init__.py

project: ml_recommendation
provider: local
registry:
  registry_type: sql
  path: postgresql://feast:feast@localhost:5432/feast_registry
  cache_ttl_seconds: 60
online_store:
  type: redis
  connection_string: 'localhost:6379,password=feast_secret'
  redis_type: redis
offline_store:
  type: postgres
  host: localhost
  port: 5432
  database: feast_offline
  db_schema: public
  user: feast
  password: feast_secret
entity_key_serialization_version: 2

# features/entities.py
from feast import Entity, ValueType

# 사용자 엔티티
user = Entity(
    name="user_id",
    value_type=ValueType.INT64,
    description="서비스 사용자 고유 식별자",
    tags={"team": "recommendation", "owner": "ml-platform"},
)

# 상품 엔티티
item = Entity(
    name="item_id",
    value_type=ValueType.INT64,
    description="상품 고유 식별자",
    tags={"team": "recommendation", "owner": "ml-platform"},
)

# features/user_features.py
from datetime import timedelta
from feast import FeatureView, Field
from feast.types import Float32, Int64, String
from feast.infra.offline_stores.contrib.postgres_offline_store.postgres_source import (
    PostgreSQLSource,
)

# 데이터 소스 정의
user_activity_source = PostgreSQLSource(
    name="user_activity_source",
    query="""
        SELECT
            user_id,
            avg_purchase_amount_30d,
            purchase_count_30d,
            max_purchase_amount_30d,
            last_login_days_ago,
            preferred_category,
            event_timestamp,
            created_timestamp
        FROM user_activity_features
    """,
    timestamp_field="event_timestamp",
    created_timestamp_column="created_timestamp",
)

# 피처 뷰 정의
user_activity_fv = FeatureView(
    name="user_activity_features",
    entities=[user],
    ttl=timedelta(days=1),  # 온라인 스토어에서 1일 이상 된 피처는 stale 처리
    schema=[
        Field(name="avg_purchase_amount_30d", dtype=Float32),
        Field(name="purchase_count_30d", dtype=Int64),
        Field(name="max_purchase_amount_30d", dtype=Float32),
        Field(name="last_login_days_ago", dtype=Int64),
        Field(name="preferred_category", dtype=String),
    ],
    source=user_activity_source,
    online=True,  # 온라인 스토어에 동기화
    tags={"team": "recommendation", "version": "v2"},
)

# features/feature_services.py
from feast import FeatureService

# 추천 모델용 피처 서비스
recommendation_feature_service = FeatureService(
    name="recommendation_v2",
    features=[
        user_activity_fv[["avg_purchase_amount_30d", "purchase_count_30d"]],
        item_popularity_fv[["view_count_7d", "purchase_rate_7d"]],
        user_item_interaction_fv,
    ],
    tags={"model": "recommendation", "version": "v2"},
    description="추천 모델 v2에서 사용하는 피처 서비스",
)

# 피처 정의를 레지스트리에 반영
feast apply

# 등록된 피처 확인
feast feature-views list
feast feature-services list
feast entities list

# 특정 기간의 피처를 온라인 스토어로 동기화
feast materialize 2026-03-01T00:00:00 2026-03-14T00:00:00

# 마지막 동기화 시점부터 현재까지 증분 동기화
feast materialize-incremental $(date -u +"%Y-%m-%dT%H:%M:%S")

# serving/realtime_serving.py
from feast import FeatureStore
import pandas as pd

# Feature Store 초기화
store = FeatureStore(repo_path="./my_feature_store")

# 단일 엔티티 온라인 피처 조회
entity_rows = [
    {"user_id": 1001},
    {"user_id": 1002},
    {"user_id": 1003},
]

# Feature Service를 통한 조회 (권장)
online_features = store.get_online_features(
    features=store.get_feature_service("recommendation_v2"),
    entity_rows=entity_rows,
).to_dict()

print(online_features)
# 출력 예시:
# - user_id: [1001, 1002, 1003]
# - avg_purchase_amount_30d: [45000.0, 12000.0, 78000.0]
# - purchase_count_30d: [15, 3, 42]
# - view_count_7d: [120, 45, 230]
# - purchase_rate_7d: [0.12, 0.05, 0.18]

# HTTP Feature Server 구동 (기본 포트 6566)
feast serve --host 0.0.0.0 --port 6566

# gRPC Feature Server 구동
feast serve --type grpc --port 6567

# HTTP API를 통한 피처 조회
import requests

response = requests.post(
    "http://localhost:6566/get-online-features",
    json={
        "feature_service": "recommendation_v2",
        "entities": {
            "user_id": [1001, 1002]
        },
    },
)

features = response.json()
print(features)

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

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

# 학습 데이터의 엔티티와 타임스탬프
entity_df = pd.DataFrame({
    "user_id": [1001, 1002, 1003, 1001, 1002],
    "item_id": [501, 502, 503, 504, 505],
    "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",
        "2026-03-07 11:20:00",
    ]),
    "label": [1, 0, 1, 1, 0],
})

# Point-in-Time Join으로 학습 데이터 생성
# 각 이벤트 시점에서 유효한 피처 값을 정확히 조인
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=store.get_feature_service("recommendation_v2"),
).to_df()

print(f"학습 데이터 shape: {training_df.shape}")
print(training_df.head())

# 학습 데이터 저장
training_df.to_parquet("training_data_v2.parquet", index=False)

# dags/feast_materialization_dag.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from datetime import datetime, timedelta

default_args = {
    "owner": "ml-platform",
    "depends_on_past": False,
    "email_on_failure": True,
    "email": ["ml-platform@company.com"],
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    dag_id="feast_daily_materialization",
    default_args=default_args,
    description="Feast 일일 Materialization 파이프라인",
    schedule_interval="0 2 * * *",  # 매일 새벽 2시
    start_date=days_ago(1),
    catchup=False,
    tags=["feast", "feature-store", "materialization"],
)

# 증분 Materialization
materialize_task = BashOperator(
    task_id="materialize_incremental",
    bash_command=(
        "cd /opt/feast/my_feature_store && "
        "feast materialize-incremental $(date -u +%Y-%m-%dT%H:%M:%S)"
    ),
    dag=dag,
)

# 온라인 스토어 헬스 체크
def check_online_store_health():
    from feast import FeatureStore
    store = FeatureStore(repo_path="/opt/feast/my_feature_store")
    # 샘플 엔티티로 조회 테스트
    result = store.get_online_features(
        features=["user_activity_features:avg_purchase_amount_30d"],
        entity_rows=[{"user_id": 1}],
    ).to_dict()
    assert result is not None, "온라인 스토어 조회 실패"
    print("온라인 스토어 헬스 체크 통과")

health_check = PythonOperator(
    task_id="online_store_health_check",
    python_callable=check_online_store_health,
    dag=dag,
)

materialize_task >> health_check

# monitoring/drift_detection.py
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metrics import (
    DataDriftTable,
    DatasetDriftMetric,
    ColumnDriftMetric,
)
from feast import FeatureStore
import pandas as pd
from datetime import datetime, timedelta

def detect_feature_drift(
    store: FeatureStore,
    feature_service_name: str,
    reference_start: str,
    reference_end: str,
    current_start: str,
    current_end: str,
    drift_threshold: float = 0.25,
):
    """피처 드리프트를 감지하고 리포트를 생성한다."""

    # 기준 데이터 (학습 시점)
    reference_entity_df = pd.read_sql(
        f"""
        SELECT user_id, event_timestamp
        FROM user_events
        WHERE event_timestamp BETWEEN '{reference_start}' AND '{reference_end}'
        LIMIT 10000
        """,
        conn,
    )

    # 현재 데이터 (서빙 시점)
    current_entity_df = pd.read_sql(
        f"""
        SELECT user_id, event_timestamp
        FROM user_events
        WHERE event_timestamp BETWEEN '{current_start}' AND '{current_end}'
        LIMIT 10000
        """,
        conn,
    )

    feature_service = store.get_feature_service(feature_service_name)

    reference_df = store.get_historical_features(
        entity_df=reference_entity_df,
        features=feature_service,
    ).to_df()

    current_df = store.get_historical_features(
        entity_df=current_entity_df,
        features=feature_service,
    ).to_df()

    # Evidently 드리프트 리포트 생성
    report = Report(metrics=[
        DatasetDriftMetric(),
        DataDriftTable(),
    ])

    report.run(
        reference_data=reference_df,
        current_data=current_df,
    )

    # 리포트 저장
    report.save_html("drift_report.html")

    # 드리프트 결과 확인
    result = report.as_dict()
    dataset_drift = result["metrics"][0]["result"]["dataset_drift"]

    if dataset_drift:
        print("WARNING: 데이터 드리프트가 감지되었습니다!")
        # 알림 전송 로직
        send_alert(
            channel="ml-alerts",
            message=f"Feature drift detected in {feature_service_name}",
        )

    return report


# 사용 예시
store = FeatureStore(repo_path="./my_feature_store")
report = detect_feature_drift(
    store=store,
    feature_service_name="recommendation_v2",
    reference_start="2026-02-01",
    reference_end="2026-02-28",
    current_start="2026-03-01",
    current_end="2026-03-14",
)

# prometheus/feast_alerts.yaml
groups:
  - name: feast_feature_store
    rules:
      - alert: FeatureServingHighLatency
        expr: histogram_quantile(0.99, rate(feast_serving_request_latency_seconds_bucket[5m])) > 0.05
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: 'Feast 피처 서빙 p99 지연 50ms 초과'
          description: '피처 서빙 p99 지연이 5분 이상 50ms를 초과했습니다.'

      - alert: FeatureServingErrors
        expr: rate(feast_serving_request_errors_total[5m]) > 0.01
        for: 3m
        labels:
          severity: critical
        annotations:
          summary: 'Feast 피처 서빙 에러율 1% 초과'
          description: '피처 서빙 에러율이 3분 이상 1%를 초과했습니다.'

      - alert: FeatureFreshnessStale
        expr: feast_feature_freshness_seconds > 86400
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: '피처 Freshness 24시간 초과'
          description: '피처가 24시간 이상 업데이트되지 않았습니다. Materialization 파이프라인을 확인하세요.'

# Redis 메모리 상태 확인
redis-cli INFO memory

# 특정 피처 뷰의 키 수 확인
redis-cli DBSIZE

# maxmemory 임시 증가
redis-cli CONFIG SET maxmemory "8gb"

# 누락 기간 수동 Materialization
feast materialize 2026-03-11T00:00:00 2026-03-14T00:00:00

# Materialization 상태 확인
feast materialize-incremental $(date -u +"%Y-%m-%dT%H:%M:%S") --verbose