[Feature Store & Feast - Feature Store-lite]

이 글에서 다루는 것

Feast 풀도입 이전 단계로, GitOps + Airflow를 사용해 피처 생성/버전화/재현성의 최소 요건(계약/메타/버전화)을 고정하는 Feature Store-lite 설계

선수지식

• Triton 서빙 플랫폼 - dynamic_batching + instance_group

이 단계에서 해결하려는 문제

Feature Store는 “ML 성능"이 아니라 운영 안정성/재현성에서 먼저 갈립니다.

“오늘 만든 feature.csv"가 아니라, 아래가 반드시 남아야 운영이 됩니다.

• 언제 생성됐는지 (generated_at)
• 어떤 스키마(계약)로 생성됐는지 (schema + schema_hash)
• 어떤 소스에서 생성됐는지 (source)
• 어떤 버전으로 저장됐는지 (version)
• 결과물이 어디 있는지 (feature_uri)

Feature Store 도입을 고민하면 흔히 “Feast부터 써야 하나?“가 먼저 나오는데, 도구보다 먼저 피처 생성/버전화/재현 방식이 고정돼야 합니다.

이 글은 Feast 풀도입 이전 단계로, **Feature Store-lite 최소 요건(계약/메타/버전화/재현성)**을 GitOps + Airflow로 먼저 고정하는 구축입니다.

• Online Store / Serving 연동은 의도적으로 제외했습니다.
• 대신 운영에서 통하는 뼈대(Contract -> Pipeline -> Versioned Storage)부터 만듭니다.

🎯 이 글에서 달성하는 것 (완료 기준)

GitOps로 계약 배포 + Airflow로 스키마 기반 피처 생성 + S3 버전화 저장 + 재현성 메타 남김

1️⃣ 전체 구조

이 글에서의 Feature Store-lite 범위는 아래까지입니다.

• GitOps로 배포된 계약 리소스(schema/metadata template)
• Airflow 파이프라인 실행
• S3에 버전화 저장 + 메타데이터로 재현성 확보

Online Store / Serving / Feast 연동은 다음 글에서 다룹니다.

2️⃣ 코드/리소스 트리

(A) GitOps 리소스

mlops-infra-gitops/
  envs/
    dev/feature-store/feature-store-cm.yaml
    prod/feature-store/feature-store-cm.yaml

(B) ArgoCD Application

mlops-infra/
  apps/
    feature-store-dev.yaml
    feature-store-prod.yaml

(C) Airflow DAG/라이브러리

airflow-dags-dev/dags/
  dag_data_pipeline_daily_v4.py       # DAG는 연결만
  mlops_lib/
    dp/
      config.py
      s3.py
      feature_schema.py
      build.py
      store.py
      tasks.py
  .airflowignore                      # 라이브러리 스캔 차단

3️⃣ 설계 포인트 (운영 관점 핵심)

(1) “계약(Contract)“을 GitOps로 배포한다

스키마/메타 템플릿은 코드가 아니라 운영 표준 리소스로 다뤘습니다.

• ConfigMap으로 관리하고 ArgoCD로 dev/prod에 동일 배포
• Airflow는 파일을 읽기만 하도록 구성 -> 코드 변경 최소화

(2) DAG는 얇게, 로직은 패키지로 분리한다

DAG가 두꺼워지는 순간 유지보수/테스트/재사용이 바로 깨집니다.

• 로직을 mlops_lib/dp/*로 분리하면
  • 로직 단위 수정/재사용 가능
  • 테스트/리팩터링이 쉬움
  • DAG는 흐름만 유지

(3) 버전화 저장은 “깊어지는 것"이 아니라 “표준화"다

버전 디렉토리 depth는 1단 고정이 운영에서 가장 안전합니다.

• 권장: s3://<feature_base>/<feature_set>/<version>/
• 흔한 실패: 날짜/실행ID/태스크ID를 섞어 depth 폭발 -> “최신이 무엇인지"가 애매해짐

4️⃣ Contract 리소스 GitOps 배포 + Airflow 마운트

ConfigMap으로 계약 리소스를 분리해 GitOps로 배포합니다.

apiVersion: v1
kind: ConfigMap
metadata:
  name: feature-store-resources
  namespace: airflow-dev
data:
  user_features.schema.json: |
    {
      "feature_set": "user_features",
      "version": "v1",
      "columns": [
        {"name": "user_id", "type": "int64"},
        {"name": "f_total_events_7d", "type": "int64"}
      ],
      "primary_keys": ["user_id"]
    }

  metadata.json.j2: |
    {
      "feature_set": "{{ feature_set }}",
      "version": "{{ version }}",
      "generated_at": "{{ generated_at }}",
      # ... (이하 생략)
    }

Airflow 구성요소에 동일하게 마운트하고, mountPath를 표준화합니다.

• /opt/airflow/feature-store/
  • user_features.schema.json
  • metadata.json.j2

5️⃣ 재현성의 핵심: schema_hash

스키마는 “계약서"이고, schema_hash는 “서명"입니다. 이 해시가 있어야 “그때 그 스키마로 만든 피처"를 시스템적으로 증명할 수 있습니다.

# dags/mlops_lib/dp/feature_schema.py
import json, hashlib

def load_schema(schema_path: str, expected_feature_set: str) -> tuple[dict, str]:
    with open(schema_path, "r", encoding="utf-8") as f:
        schema = json.load(f)

    if schema.get("feature_set") != expected_feature_set:
        raise ValueError(f"schema mismatch: {schema.get('feature_set')} != {expected_feature_set}")

    canonical = json.dumps(schema, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    schema_hash = hashlib.sha256(canonical).hexdigest()
    return schema, schema_hash

6️⃣ Airflow 파이프라인 흐름

원칙: DAG는 연결만 담당한다

# dags/dag_data_pipeline_daily_v4.py (요약)
with DAG(
    dag_id="data_pipeline_daily_dev_v4",
    schedule=None,
    catchup=False,
    max_active_runs=1,
    tags=["data-pipeline", "dev", "mlops"],
    on_failure_callback=alert_slack,
) as dag:
    t1 = PythonOperator(task_id="extract_raw_data", python_callable=task_extract_raw_data)
    t2 = PythonOperator(task_id="validate_data", python_callable=task_validate_data)
    t3 = PythonOperator(task_id="build_features", python_callable=task_build_features)
    t4 = PythonOperator(task_id="store_features", python_callable=task_store_features)
    # ... (이하 생략)

    t1 >> t2 >> t3 >> t4

실행 단계

• extract_raw_data: RAW S3 객체 존재 확인 + source 경로 XCom 기록
• validate_data: 최소 검증(빈 데이터 방지)
• build_features: schema 로드 + schema_hash 생성 + CSV 생성(스키마 순서 준수)
• store_features: 버전 생성 후 S3에 3파일 세트 저장

7️⃣ S3 저장 규칙: “버전 1단 + 3파일 세트”

실무에서 가장 중요한 고정 규칙입니다.

• features.csv
• schema.json
• metadata.json

# dags/mlops_lib/dp/store.py (핵심)
def store_features(feature_base, pipeline_name, feature_set, metadata_tpl_path, ti):
    ver = _version_id(getattr(ti, "execution_date", None))
    prefix = prefix.rstrip("/") + f"/{feature_set}/{ver}/"

    feature_uri = f"s3://{bkt}/{prefix}features.csv"

    s3.put_object(... Key=f"{prefix}features.csv", Body=features_csv.encode("utf-8"))
    s3.put_object(... Key=f"{prefix}schema.json", Body=json.dumps(schema, ...).encode("utf-8"))
    s3.put_object(... Key=f"{prefix}metadata.json", Body=meta.encode("utf-8"))
    # ... (이하 생략)

8️⃣ 검증 체크리스트 (운영형)

• ArgoCD에서 feature-store-dev, feature-store-prod Synced/Healthy
• Airflow Pod 내부 파일 존재 확인
  • /opt/airflow/feature-store/user_features.schema.json
  • /opt/airflow/feature-store/metadata.json.j2
• DAG 실행 성공
• S3에 버전화 prefix 생성 확인
• 동일 prefix에 3개 파일 존재: features.csv, schema.json, metadata.json
• metadata.json 값 채워짐: schema_hash, feature_uri, generated_at, source
• KubernetesExecutor Task Pod에서 AWS credential chain 정상 동작

9️⃣ 트러블슈팅 (이번 구축에서 실제로 걸린 지점)

(1) ArgoCD Application apply 에러: spec.orphanedResources

• CRD 버전에 따라 필드가 없어서 strict decode 실패
• 해결: 해당 필드 제거(또는 ArgoCD 버전과 스펙 정합 맞추기)

(2) Task Pod에서 Variable.get(default_var=...) 에러

• 에러: unexpected keyword argument 'default_var'
• 원인: Airflow 버전/SDK 차이
• 해결: 예외 처리로 안전화

def _get_var(key: str, default: str) -> str:
    try:
        return Variable.get(key)
    except Exception:
        return default

(3) mlops_lib 분리 후 import 에러

• Airflow가 라이브러리 파일까지 DAG처럼 스캔하면서 import가 깨짐
• 해결: .airflowignore로 스캔 차단

(4) KubernetesExecutor 로그 추적 난이도

• Task Pod가 짧게 올라왔다가 사라짐
• 실무 팁: Airflow UI 로그 + remote logging(권장), 또는 빠르게 kubectl logs -f로 추적

설계 판단 (Why This Way?)

Feast 도입 전에 스키마·버전화·재현성 계약을 먼저 확립해야 도구 위에 얹기만 하면 되고, 스키마는 ConfigMap으로 분리하여 ArgoCD sync 기반 GitOps 배포와 롤백을 가능하게 했습니다. 버전 디렉토리를 1단 구조로 고정하여 최신 버전 탐색과 latest 포인터 도입을 단순화했습니다.

다음에 읽을 글

→ Feature Store & Feast - Feast — Feast 도입과 운영