Feature Store 구축 완전 가이드: Feast 아키텍처·온라인/오프라인 서빙·ML 파이프라인 통합

# Feast 설치 (Redis, PostgreSQL 지원 포함)
pip install 'feast[redis,postgres]'

# 프로젝트 초기화
feast init feature_repo
cd feature_repo

# 디렉토리 구조 확인
# feature_repo/
#   feature_store.yaml    -- 프로젝트 설정
#   definitions.py        -- Entity, Feature View 정의
#   data/                 -- 샘플 데이터

project: my_ml_platform
provider: gcp
registry:
  registry_type: sql
  path: postgresql://feast:feast@db-host:5432/feast_registry
  cache_ttl_seconds: 60
online_store:
  type: redis
  connection_string: redis-host:6379,password=secret
offline_store:
  type: bigquery
  dataset: feast_offline
entity_key_serialization_version: 2

from datetime import timedelta
from feast import Entity, FeatureView, Field, FileSource, BigQuerySource
from feast.types import Float32, Float64, Int64, String

# Entity 정의
customer = Entity(
    name="customer_id",
    description="고객의 고유 식별자",
)

driver = Entity(
    name="driver_id",
    description="드라이버의 고유 식별자",
)

# BigQuery 소스 정의
customer_stats_source = BigQuerySource(
    name="customer_stats_source",
    table="my_project.feast_dataset.customer_stats",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_timestamp",
)

# Feature View 정의
customer_stats_fv = FeatureView(
    name="customer_stats",
    entities=[customer],
    ttl=timedelta(days=3),
    schema=[
        Field(name="total_orders", dtype=Int64),
        Field(name="avg_order_amount", dtype=Float64),
        Field(name="lifetime_value", dtype=Float64),
        Field(name="preferred_category", dtype=String),
        Field(name="churn_risk_score", dtype=Float32),
    ],
    source=customer_stats_source,
    online=True,
    tags={
        "team": "growth",
        "version": "v2",
    },
)

driver_stats_source = BigQuerySource(
    name="driver_stats_source",
    table="my_project.feast_dataset.driver_stats",
    timestamp_field="event_timestamp",
)

driver_stats_fv = FeatureView(
    name="driver_stats",
    entities=[driver],
    ttl=timedelta(hours=6),
    schema=[
        Field(name="avg_rating", dtype=Float64),
        Field(name="total_trips", dtype=Int64),
        Field(name="acceptance_rate", dtype=Float64),
        Field(name="avg_delivery_time_min", dtype=Float32),
    ],
    source=driver_stats_source,
    online=True,
)

from feast import FeatureService

# 이탈 예측 모델용 Feature Service
churn_prediction_svc = FeatureService(
    name="churn_prediction_service",
    features=[
        customer_stats_fv[["total_orders", "avg_order_amount", "lifetime_value", "churn_risk_score"]],
    ],
    tags={
        "model": "churn_prediction_v3",
        "owner": "growth-team",
    },
)

# 드라이버 매칭 모델용 Feature Service
driver_matching_svc = FeatureService(
    name="driver_matching_service",
    features=[
        driver_stats_fv[["avg_rating", "acceptance_rate", "avg_delivery_time_min"]],
        customer_stats_fv[["preferred_category"]],
    ],
)

from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path="feature_repo/")

# 학습용 엔티티 데이터프레임 (라벨 포함)
entity_df = pd.DataFrame({
    "customer_id": [1001, 1002, 1003, 1001, 1002],
    "event_timestamp": pd.to_datetime([
        "2026-01-15 10:00:00",
        "2026-01-15 11:00:00",
        "2026-01-16 09:00:00",
        "2026-02-01 10:00:00",
        "2026-02-01 11:00:00",
    ]),
    "churned": [0, 1, 0, 1, 0],  # 라벨
})

# Point-in-Time Join으로 학습 데이터 생성
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "customer_stats:total_orders",
        "customer_stats:avg_order_amount",
        "customer_stats:lifetime_value",
        "customer_stats:churn_risk_score",
    ],
).to_df()

print(training_df.head())
# customer_id | event_timestamp     | churned | total_orders | avg_order_amount | ...
# 1001        | 2026-01-15 10:00:00 | 0       | 42           | 35.50            | ...

# 온라인 피처 조회
online_features = store.get_online_features(
    features=[
        "customer_stats:total_orders",
        "customer_stats:avg_order_amount",
        "customer_stats:churn_risk_score",
    ],
    entity_rows=[
        {"customer_id": 1001},
        {"customer_id": 1002},
    ],
).to_dict()

print(online_features)
# 출력 예시:
# {
#     "customer_id": [1001, 1002],
#     "total_orders": [45, 12],
#     "avg_order_amount": [35.50, 28.00],
#     "churn_risk_score": [0.15, 0.82],
# }

# 전체 Feature View Materialization
feast materialize 2026-01-01T00:00:00 2026-03-10T00:00:00

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

from airflow import DAG
from airflow.decorators import task
from datetime import datetime, timedelta

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

with DAG(
    dag_id="feast_materialization_pipeline",
    default_args=default_args,
    schedule_interval="0 */4 * * *",  # 4시간마다 실행
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    @task()
    def validate_source_data():
        """소스 데이터 품질 검증"""
        from great_expectations import get_context
        context = get_context()
        result = context.run_checkpoint(checkpoint_name="feature_source_check")
        if not result.success:
            raise ValueError("소스 데이터 품질 검증 실패")
        return True

    @task()
    def materialize_features():
        """오프라인 스토어에서 온라인 스토어로 Materialization"""
        from feast import RepoConfig, FeatureStore
        from feast.infra.online_stores.redis import RedisOnlineStoreConfig
        from feast.repo_config import RegistryConfig

        repo_config = RepoConfig(
            project="my_ml_platform",
            provider="gcp",
            registry=RegistryConfig(
                registry_type="sql",
                path="postgresql://feast:feast@db-host:5432/feast_registry",
            ),
            online_store=RedisOnlineStoreConfig(
                connection_string="redis-host:6379",
            ),
        )
        store = FeatureStore(config=repo_config)
        store.materialize_incremental(end_date=datetime.utcnow())
        return True

    @task()
    def validate_online_store():
        """온라인 스토어 피처 값 검증"""
        from feast import FeatureStore
        store = FeatureStore(repo_path="feature_repo/")

        # 샘플 엔티티로 온라인 피처 조회
        result = store.get_online_features(
            features=["customer_stats:total_orders"],
            entity_rows=[{"customer_id": 1001}],
        ).to_dict()

        if result["total_orders"][0] is None:
            raise ValueError("온라인 스토어에 피처가 로드되지 않음")
        return True

    @task()
    def notify_completion():
        """Slack 알림 전송"""
        import requests
        requests.post(
            "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
            json={"text": "Feast Materialization 완료"},
        )

    validate_source_data() >> materialize_features() >> validate_online_store() >> notify_completion()

from kfp import dsl
from kfp.dsl import component, Output, Dataset

@component(
    base_image="python:3.10",
    packages_to_install=["feast[redis,postgres]>=0.40.0"],
)
def feast_materialize_op(
    project_name: str,
    registry_path: str,
    redis_connection: str,
):
    from feast import RepoConfig, FeatureStore
    from feast.infra.online_stores.redis import RedisOnlineStoreConfig
    from feast.repo_config import RegistryConfig
    from datetime import datetime

    config = RepoConfig(
        project=project_name,
        provider="gcp",
        registry=RegistryConfig(
            registry_type="sql",
            path=registry_path,
        ),
        online_store=RedisOnlineStoreConfig(
            connection_string=redis_connection,
        ),
    )
    store = FeatureStore(config=config)
    store.materialize_incremental(end_date=datetime.utcnow())

@component(
    base_image="python:3.10",
    packages_to_install=["feast[redis,postgres]>=0.40.0", "scikit-learn"],
)
def train_model_op(
    project_name: str,
    model_output: Output[Dataset],
):
    from feast import FeatureStore
    import pandas as pd
    from sklearn.ensemble import GradientBoostingClassifier
    import pickle

    store = FeatureStore(repo_path="feature_repo/")
    entity_df = pd.read_parquet("gs://my-bucket/training_entities.parquet")

    training_df = store.get_historical_features(
        entity_df=entity_df,
        features=[
            "customer_stats:total_orders",
            "customer_stats:avg_order_amount",
            "customer_stats:churn_risk_score",
        ],
    ).to_df()

    X = training_df.drop(columns=["customer_id", "event_timestamp", "churned"])
    y = training_df["churned"]

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

    with open(model_output.path, "wb") as f:
        pickle.dump(model, f)

@dsl.pipeline(name="feast-ml-training-pipeline")
def feast_training_pipeline():
    materialize_task = feast_materialize_op(
        project_name="my_ml_platform",
        registry_path="postgresql://feast:feast@db-host:5432/feast_registry",
        redis_connection="redis-host:6379",
    )
    train_task = train_model_op(
        project_name="my_ml_platform",
    )
    train_task.after(materialize_task)