Feast: 오픈소스 피처 스토어 서브세컨드 특성 서빙 — 2026 설치 가이드

# feature_store.yaml — Feast 프로젝트 구성
project: fraud_detection
provider: local
registry: 
  path: s3://my-bucket/registry.db  # 프로덕션용 SQL 레지스트리
online_store:
  type: redis
  connection_string: "redis://localhost:6379"
offline_store:
  type: bigquery
  project: my-gcp-project
  dataset: feast_offline
entity_key_serialization_version: 2

# 학습을 위한 역사적 피처 검색
from feast import FeatureStore

store = FeatureStore(repo_path=".")

historical_df = store.get_historical_features(
    entity_df=entity_df,  # entity ID와 타임스탬프가 있는 DataFrame
    features=[
        "user_features:avg_order_amount_30d",
        "user_features:total_transactions_90d",
        "user_features:days_since_last_order",
    ],
).to_df()

# 실시간 추론을 위한 온라인 피처 검색
features = store.get_online_features(
    features=[
        "user_features:avg_order_amount_30d",
        "user_features:total_transactions_90d",
    ],
    entity_rows=[{"user_id": "user_12345"}],
).to_dict()

# 반환: {'avg_order_amount_30d': [245.50], 'total_transactions_90d': [12]}

# 피처 서버 시작
feast serve --port 6566

# 피처 검색을 위한 REST API 엔드포인트
curl -X POST "http://localhost:6566/get-online-features" \
  -H "Content-Type: application/json" \
  -d '{
    "features": ["user_features:avg_order_amount_30d"],
    "entities": {"user_id": ["user_12345"]}
  }'

# 코어 Feast (최소 설치)
pip install feast

# BigQuery 오프라인 스토어 포함
pip install "feast[bigquery]"

# Redis 온라인 스토어 포함
pip install "feast[redis]"

# Snowflake 포함
pip install "feast[snowflake]"

# PostgreSQL 포함
pip install "feast[postgres]"

# 모든 일반적인 백엔드가 포함된 전체 설치
pip install "feast[gcp,redis,postgres,snowflake]"

feast version
# Feast SDK Version: 0.63.0

# 프로젝트 디렉토리 생성 및 진입
mkdir fraud_detection_feature_store
cd fraud_detection_feature_store

# Feast 초기화 (feature_store.yaml 및 example/ 생성)
feast init

# 프로젝트 구조:
# .
# ├── feature_store.yaml    # 메인 설정
# ├── example/
# │   ├── repo/
# │   │   ├── example.py    # 피처 정의
# │   │   └── test_workflow.py

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

# 사기 모델의 주요 엔터티 정의
user = Entity(
    name="user_id",
    value_type=ValueType.STRING,
    description="각 사용자의 고유 식별자",
    join_key="user_id",
)

# features/data_sources.py
from feast import BigQuerySource

# 오프라인 스토어의 역사적 데이터 소스
transaction_stats_source = BigQuerySource(
    name="transaction_stats",
    query="""
        SELECT 
            user_id,
            event_timestamp,
            avg_order_amount_30d,
            total_transactions_90d,
            days_since_last_order,
            unique_merchants_30d,
            avg_transaction_amount_7d,
            failed_transaction_rate_30d,
            created
        FROM `my-gcp-project.featds.transaction_aggregates`
    """,
    timestamp_field="event_timestamp",
    created_timestamp_column="created",
)

# features/feature_views.py
from feast import FeatureView, Field
from feast.types import Float32, Int64, Float64
from datetime import timedelta
from features.entities import user
from features.data_sources import transaction_stats_source

# 슬라이딩 윈도우 집계가 있는 피처 뷰
user_transaction_features = FeatureView(
    name="user_transaction_features",
    entities=[user],
    ttl=timedelta(days=90),  # 90일간 유효한 피처
    schema=[
        Field(name="avg_order_amount_30d", dtype=Float64),
        Field(name="total_transactions_90d", dtype=Int64),
        Field(name="days_since_last_order", dtype=Int64),
        Field(name="unique_merchants_30d", dtype=Int64),
        Field(name="avg_transaction_amount_7d", dtype=Float64),
        Field(name="failed_transaction_rate_30d", dtype=Float32),
    ],
    online=True,   # 온라인 스토어(Redis)에서 제공
    source=transaction_stats_source,
    tags={"team": "fraud", "domain": "transactions"},
    owner="ml-team@company.com",
)

# features/feature_services.py
from feast import FeatureService
from features.feature_views import user_transaction_features

# 피처 서비스 — 모델이 사용하는 인터페이스
fraud_detection_v1 = FeatureService(
    name="fraud_detection_v1",
    features=[user_transaction_features],
    tags={"version": "1.0", "model": "fraud_xgboost"},
    owner="ml-team@company.com",
)

# 레지스트리에 피처 정의 적용
feast apply

# 오프라인에서 온라인 스토어로 피처 머티리얼라이즈
# (최신 피처 값으로 Redis 채우기)
feast materialize-incremental $(date -u +"%Y-%m-%dT%H:%M:%S")

# 또는 특정 시간 범위 머티리얼라이즈
feast materialize 2026-01-01T00:00:00 2026-05-19T00:00:00

# feature_store.yaml — 프로덕션 설정
project: fraud_detection
provider: gcp
registry:
  registry_store_type: sql
  path: "postgresql://user:pass@pg-host:5432/feast_registry"
  cache_ttl_seconds: 60

online_store:
  type: redis
  connection_string: "redis://:password@redis-cluster.internal:6379"
  key_ttl_seconds: 604800  # 피처 키의 7일 TTL
  
offline_store:
  type: bigquery
  project: my-gcp-project
  dataset: feast_offline
  location: US

entity_key_serialization_version: 2

flags:
  alpha_features: true
  on_demand_transforms: true

# Redis Cluster 구성
online_store:
  type: redis
  redis_type: redis_cluster
  connection_string: "redis://redis-node-1:6379,redis-node-2:6379,redis-node-3:6379"
  key_ttl_seconds: 604800

# Ubuntu 22.04에 Redis 배포 (DigitalOcean Droplet)
sudo apt update
sudo apt install redis-server

# 프로덕션용 구성
sudo tee -a /etc/redis/redis.conf <<EOF
maxmemory 2gb
maxmemory-policy allkeys-lru
bind 0.0.0.0
protected-mode yes
requirepass your_secure_password
EOF

sudo systemctl restart redis

# 확인
redis-cli ping
# PONG

# training_pipeline.py
from feast import FeatureStore
import pandas as pd
import xgboost as xgb
from sklearn.model_selection import train_test_split

# 피처 스토어 초기화
store = FeatureStore(repo_path=".")

# 레이블이 있는 엔터티 DataFrame 로드 (user_id + 타겟 타임스탬프 + 레이블)
entity_df = pd.read_parquet("s3://training-data/labeled_users.parquet")

# 포인트인타임 정확한 피처 검색
feature_service = store.get_feature_service("fraud_detection_v1")
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=feature_service,
).to_df()

# 분할 및 학습
X = training_df.drop(columns=["user_id", "event_timestamp", "is_fraud"])
y = training_df["is_fraud"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

model = xgb.XGBClassifier(
    max_depth=6,
    learning_rate=0.1,
    n_estimators=200,
    subsample=0.8,
)
model.fit(X_train, y_train)

# 평가
from sklearn.metrics import roc_auc_score
print(f"AUC-ROC: {roc_auc_score(y_test, model.predict_proba(X_test)[:, 1]):.4f}")

# inference_service.py
from feast import FeatureStore
from fastapi import FastAPI
import xgboost as xgb
import joblib

app = FastAPI()
store = FeatureStore(repo_path=".")
model = joblib.load("models/fraud_xgboost.pkl")

@app.post("/predict")
async def predict(user_id: str, transaction_amount: float):
    # Redis에서 온라인 피처 검색 (< 5ms)
    features = store.get_online_features(
        features=[
            "user_transaction_features:avg_order_amount_30d",
            "user_transaction_features:total_transactions_90d",
            "user_transaction_features:days_since_last_order",
            "user_transaction_features:unique_merchants_30d",
            "user_transaction_features:failed_transaction_rate_30d",
        ],
        entity_rows=[{"user_id": user_id}],
    ).to_dict()
    
    # 피처 벡터 구축
    feature_vector = [
        transaction_amount,
        features["avg_order_amount_30d"][0],
        features["total_transactions_90d"][0],
        features["days_since_last_order"][0],
        features["unique_merchants_30d"][0],
        features["failed_transaction_rate_30d"][0],
    ]
    
    # 예측
    fraud_probability = model.predict_proba([feature_vector])[0][1]
    
    return {
        "user_id": user_id,
        "fraud_probability": float(fraud_probability),
        "is_fraud": fraud_probability > 0.7,
        "features_retrieved": {k: v[0] for k, v in features.items()},
    }

# dags/feast_materialize.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "ml-team",
    "depends_on_past": False,
    "email_on_failure": True,
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    "feast_materialize",
    default_args=default_args,
    schedule_interval="@hourly",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:
    
    materialize = BashOperator(
        task_id="materialize_features",
        bash_command="""
            cd /opt/feast/fraud_detection_feature_store && \
            feast materialize-incremental {{ ds }}T{{ ts_nodash_with_tz }}
        """,
    )
    
    validate = BashOperator(
        task_id="validate_online_store",
        bash_command="""
            cd /opt/feast/fraud_detection_feature_store && \
            python scripts/validate_online_features.py
        """,
    )
    
    materialize >> validate

from feast import on_demand_feature_view
from feast.types import Float64

# 온디맨드 변환 정의
@on_demand_feature_view(
    sources=[user_transaction_features],
    schema=[
        Field(name="transaction_amount_ratio", dtype=Float64),
    ],
    mode="python",
)
def transaction_transforms(inputs):
    import pandas as pd
    df = pd.DataFrame()
    df["transaction_amount_ratio"] = (
        inputs["transaction_amount"] / inputs["avg_order_amount_30d"]
    ).fillna(0)
    return df

from feast.dqm.profilers.ge_profiler import GeProfiler

# 데이터 품질 기대치를 피처 뷰에 첨부
user_transaction_features_with_validation = FeatureView(
    name="user_transaction_features",
    entities=[user],
    schema=[
        Field(name="avg_order_amount_30d", dtype=Float64),
        Field(name="total_transactions_90d", dtype=Int64),
    ],
    source=transaction_stats_source,
    profiler=GeProfiler(
        expectations=[
            {
                "expectation_type": "expect_column_mean_to_be_between",
                "kwargs": {
                    "column": "avg_order_amount_30d",
                    "min_value": 10.0,
                    "max_value": 10000.0,
                },
            },
        ]
    ),
)

# feature_store_team_a.yaml
project: team_a_fraud
registry:
  path: s3://shared-bucket/registry_team_a.db
online_store:
  type: redis
  connection_string: "redis://shared-redis:6379/0"
offline_store:
  type: bigquery
  project: my-gcp-project
  dataset: team_a_features

pip install feast[redis,bigquery]
feast init
# 엔터티, 피처 뷰, 피처 서비스 정의
feast apply
feast materialize-incremental $(date -u +"%Y-%m-%dT%H:%M:%S")