Feast와 MongoDB 통합

온라인 및 오프라인 스토어

• 온라인 저장 는 단일 MongoDB 컬렉션 으로 지원되는 키-값 저장 로, 온라인 추론 중에 엔터티별 최신 기능을 짧은 지연 시간으로 검색할 수 있도록 최적화되어 있습니다.

• 오프라인 저장 는 교육 데이터 세트, 점수 산정 및 구체화(데이터를 온라인 저장 로 승격)를 위해 MongoDB 컬렉션 (일반적으로 이라고 함)에 저장된 기록 기능 데이터 행을 쿼리하는 컴퓨팅 및 번역 계층입니다.feature_history

일반적인 워크플로 패턴

일반적인 엔드 투 엔드 워크플로는 다음과 같습니다.

1. MongoDB 지원 컬렉션을 점 엔터티, 기능 뷰 및 데이터 소스를 정의합니다.

2. offline_write_batch를 통해 오프라인 저장 에 기능 데이터를 수집하면 PyArrow 테이블을 입력으로 받고 오프라인 저장 스키마 에 따라 데이터를 feature_history MongoDB 컬렉션 에 삽입합니다.

3. get_historical_features을(를) 사용하여 교육 데이터를 생성하면 MongoDB 에 저장된 기록 기능 행에 대해 효율적인 특정 시점 조인을 실행합니다.

4. pull_latest_from_table_or_query 및 online_write_batch를 사용하여 오프라인 저장 의 최신 기능 값을 온라인 저장 로 구체화합니다.

5. 직렬화된 엔터티 키로 입력되는 단일 MongoDB 컬렉션 에서 읽는 Feast의 온라인 API를 통해 온라인 기능을 제공합니다.

개념 매핑

데이터 모델

MongoDB 오프라인 저장 모든 기능 보기에 대한 추가 전용 기록 기능 행을 저장하는 단일 공유 컬렉션 ( 기본값 으로 feature_history)을 사용합니다.

각 문서 특정 이벤트 타임스탬프에서 하나의 FeatureView에 대한 하나의 엔터티에 대한 하나의 관찰을 나타냅니다.

{
  "entity_id": "Binary(...)",
  "feature_view": "driver_stats",
  "event_timestamp": "ISODate(2024-01-15T12:00:00Z)",
  "created_at": "ISODate(2024-01-15T12:01:00Z)",
  "features": {
    "conv_rate": 0.72,
    "acc_rate": 0.91,
    "avg_daily_trips": 14
  }
}

주요 속성:

• 추가 전용: 기록 데이터는 변경할 수 없는 것으로 처리됩니다. 수정 사항은 현재 위치 업데이트가 아닌 최신 created_at 타임스탬프가 있는 새 행으로 기록됩니다.

• Time Series 친숙한: event_timestamp 은 기능 값이 관찰된 시점을 나타냅니다. created_at 은(는) 여러 관찰이 동일한 이벤트 타임스탬프를 주식 경우 타이 브레이커(tie-breaker)로 사용됩니다.

• FeatureView별 기능 그룹화: feature_view 는 행이 속한 FeatureView를 식별하므로 단일 컬렉션 여러 FV를 호스팅하다 할 수 있습니다.

단일 복합 인덱스 모든 주요 쿼리 패턴을 지원합니다.

(entity_id ASC, feature_view ASC, event_timestamp DESC, created_at DESC)

이 인덱스 사용하면 엔티티 및 기능 뷰에 대한 범위 스캔을 효율적으로 수행할 수 있으며 집계 중에 (entity_id, feature_view) 당 가장 최근의 관찰이 먼저 표시됩니다.

핵심 오프라인 운영

MongoDB 오프라인 저장 표준 Feast 오프라인 저장 인터페이스를 구현합니다.

• offline_write_batch - 구성된 MongoDBSource 메타데이터 사용하여 connection_string, database 및 collection를 결정하여 기능 데이터의 pyarrow.Table 를 기본 MongoDB 컬렉션 에 기록합니다.

• get_historical_features - 엔티티 및 이벤트 타임스탬프의 entity_df 와 FeatureViews 설정하다 가 주어지면 각 행 기능 특정 시점의 올바른 기능 값이 포함된 확장된 테이블을 반환합니다: 각 (entity_id, event_timestamp) 쌍에 대해 event_timestamp <= entity_event_timestamp 그리고 into TTL 이 선택되었습니다.

• pull_latest_from_table_or_query - Feast의 구체화 엔진 이 온라인 저장 시드하는 데 사용하는 시간 창 최신 기능 값을 포함하는 엔터티당 하나의 행을 반환합니다.

• pull_all_from_table_or_query - 동일한 feature_history 스키마 및 인덱스 기반으로 내보내기 또는 검사를 위해 지정된 날짜 범위 의 데이터 소스 에서 모든 행을 검색합니다.

• persist (RetrievalJob.persist을(를) 통해) - 기록 기능 쿼리 의 결과를 feature_history와(과) 별개인 SavedDatasetStorage을(를) 통해 별도의 컬렉션 또는 외부 싱크에 씁니다.

FeatureStore.write_to_offline_store(feature_view_name, df)
  → provider.ingest_df_to_offline_store(feature_view, arrow_table)
    → OfflineStore.offline_write_batch(config, feature_view, table, progress)

$match { feature_view, event_timestamp: {$gte, $lte} }
→ $sort { entity_id, event_timestamp DESC, created_at DESC }
→ $group $first by entity_id
→ $project { entity_id, event_timestamp, features.* }

애그리게이션 구현

권장되는 오프라인 구현 MongoDBOfflineStore이라는 애그리게이션 기반 MongoDB 오프라인 저장 입니다.

주요 특성:

• feature_view(으)로 구별되는 모든 FeatureView에서 공유하는 단일 feature_history 컬렉션 사용합니다.

• 모든 쿼리에 대해 복합 인덱스 (entity_id, feature_view, event_timestamp, created_at) 에 의존하여 전체 컬렉션 스캔을 방지합니다.

• 서버 측 $group $first 을(를) 워크로드 "점수"(엔터티당 하나의 행)에 사용하고, 엔터티 ID가 반복되는 "교육" 워크로드에는 pd.merge_asof 을(를) 사용하여 정확성과 성능의 균형을 맞춥니다.

• 청킹을 통한 메모리 사용량이 제한되므로 RAM 소진하지 않고 큰 entity_df 값을 처리할 수 있습니다.

벤치마크에 따르면 이 구현 대체 MongoDB 오프라인 접근 방식에 비해 처리량 과 메모리 효율성 의 최상의 조합을 제공합니다.

scoring_path = (
    entity_df[all_entity_id_cols].drop_duplicates().shape[0]
    == len(entity_df)
)

$match  →  $sort  →  $group $first  →  $project

# Merge on entity_id (left = entity_df rows, right = server results)
merged = result[["_fv_entity_id", event_timestamp_col]].merge(
    fv_join, on="_fv_entity_id", how="left"
)
# Null out rows where the server doc is in the future or outside TTL
future_mask = merged["_fv_ts"] > merged[event_timestamp_col]
if fv.ttl:
    ttl_mask = merged["_fv_ts"] < (
        merged[event_timestamp_col] - fv.ttl
    )
    bad_mask = future_mask | ttl_mask
else:
    bad_mask = future_mask
for feat in features:
    vals = merged[feat].copy()
    vals[bad_mask | merged["_fv_ts"].isna()] = None
    result[col] = vals.values

$match  →  (no $group)

result = pd.merge_asof(
    result.sort_values(event_timestamp_col),
    fv_df_subset.sort_values("_fv_ts"),
    left_on=event_timestamp_col,
    right_on="_fv_ts",
    by="_fv_entity_id",
    direction="backward",
)

if len(working_df) <= CHUNK_SIZE:
    result_df = _run_single(working_df, coll)
else:
    chunks = [
        _run_single(chunk, coll)
        for chunk in _chunk_dataframe(working_df, CHUNK_SIZE)
    ]
    result_df = pd.concat(chunks, ignore_index=True)

if "features" in fv_df.columns:
    for feat in features:
        src_col = reverse_fm.get(feat, feat)
        fv_df[feat] = fv_df["features"].apply(
            lambda d, _s=src_col: (
                d.get(_s) if isinstance(d, dict) else None
            )
        )
    fv_df = fv_df.drop(columns=["features"])

오프라인 스토어 기능

데이터 레이크 또는 웨어하우스로 직접 내보내기와 같은 추가 편의는 특정 RetrievalJob 구현 에 따라 달라지며 오프라인 스토어에 대한 Feast의 표준 패턴을 따를 것으로 예상됩니다.

데이터 모델

MongoDB 온라인 저장 직렬화된 엔터티 키로 키로 지정되는 모든 FeatureViews에 대해 단일 컬렉션 사용합니다.

• _id: serialized_entity_key(entity_key), 엔터티 이름과 값을 정렬하고 바이트로 인코딩하는 Feast의 안정적인 인코딩 함수에서 생성합니다.

• features: 각 FeatureView가 자체 기능 네임스페이스 유지 관리하는 중첩된 하위 문서입니다.

• event_timestamps: 해당 FeatureView의 최신 값이 작성된 시점을 나타내는 FeatureView별 타임스탬프입니다.

• created_timestamp 또는 updated_at: TTL 인덱싱 및 진단에 유용한 부기 필드입니다.

예시(간체):

{
  "_id": "b\"<serialized_entity_key>\"",
  "features": {
    "driver_stats": {
      "rating": 4.91,
      "trips_last_7d": 132
    },
    "pricing": {
      "surge_multiplier": 1.2
    }
  },
  "event_timestamps": {
    "driver_stats": "ISODate(2026-01-01T12:00:00Z)",
    "pricing": "ISODate(2026-01-21T12:00:00Z)"
  },
  "created_timestamp": "ISODate(2026-01-21T12:00:00Z)"
}

설계 근거:

• 단일 컬렉션 각 엔터티의 상태 하나의 문서 에 보관하며, 이는 키 기반 조회에 대한 Feast의 기대치에 부합하고 FeatureView별 컬렉션에서 상태 조각화되는 것을 방지합니다.

• 직렬화된 엔터티 키를 _id 로 사용하면 Feast의 결정론적 인코딩을 재사용하고, 컬렉션 전체에서 프라이머리 키가 중복되는 것을 방지하고, 엔터티당 단일 키 조회로 계속 검색합니다.

핵심 온라인 운영

MongoDB 온라인 저장 Feast의 표준 온라인 저장 API 구현합니다.

• online_write_batch - 구체화하는 동안 Feast는 각 엔터티의 최신 기능 값을 MongoDB 문서에 씁니다. 각 배치 업서트 중첩된 관련 features.<feature_view> 하위 문서와 event_timestamps 의 해당 항목만 업데이트하여 엔터티 문서를 원자적이고 일관적인 유지합니다.

• online_read 및 get_online_features - 온라인 검색은 오프라인과 동일한 직렬화 로직을 사용하여 엔터티 키를 _id 값으로 해석한 다음 키 조회를 수행합니다. 각 조회는 중첩된 features 구조를 활용하여 엔터티에 대해 요청된 모든 기능을 한 번의 왕복으로 반환합니다.

• TTL 및 최신 상태 - 기능 TTL FeatureView에서 구성되며 주로 오프라인 PIT 조인에 사용됩니다. 온라인 TTL updated_at 또는 유사한 타임스탬프에 대한 인덱스 사용하여 구현할 수 있으며, 이는 오프라인 스토어는 추가 전용이고 온라인 스토어는 최신 상태 보유한다는 Feast의 개념과 일관적인 .

오프라인 스토어 구성

오프라인 저장 MongoDBOfflineStoreConfig을 사용하여 구성됩니다.

class MongoDBOfflineStoreConfig(FeastConfigBaseModel):
    type: str = "...MongoDBOfflineStore"
    connection_string: str = "mongodb://localhost:27017"
    database: str = "feast"
    collection: str = "feature_history"

예 feature_store.yaml :

offline_store:
  type: feast.infra.offline_stores.contrib.mongodb_offline_store.mongodb.MongoDBOfflineStore
  connection_string: "mongodb+srv://user:pass@cluster.mongodb.net"
  database: feast
  collection: feature_history

MongoDBSource 는 해당 DataSource 입니다. 해당 name 필드 는 feature_view 모든 문서 에 저장된 판별자가 됩니다. 전체 구성 옵션은 Feast 문서에서 MongoDB 데이터 소스 참조를 참조하세요.

source = MongoDBSource(
    name="driver_stats",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_at",
)