MongoDBと Feast の統合

オンラインストアとオフラインストア

• オンライン ストアは、オンライン推論中にエンティティごとの最新機能を低レイテンシで取得するために最適化された、単一の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. Feast のオンライン API 経由で機能をオンラインで提供します。この API は、シリアル化されたエンティティ キーでキー付けされた単一のMongoDBコレクションから読み取ります。

概念マッピング

データモデル

MongoDBオフライン ストアは、すべての機能ビューに対する追加のみの履歴機能行を保存する単一の共有コレクション（デフォルトでは feature_history）を使用します。

各ドキュメントは、特定のイベントタイムスタンプにおける 1 つの FeatureView の 1 つのエンティティの 1 つの観察を表します。

{
  "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 履歴データは不変として扱われます。修正は、インプレース アップデートではなく、新しい行として書き込まれます。

• 時系列に適しています: event_timestamp は機能値が観察された時間を表します。 created_at は、複数の観察が同じイベントタイムスタンプを共有する場合に、タイブレークとして使用されます。

• 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 を決定し、基礎のMongoDBコレクションに機能データの pyarrow.Table を書き込みます。

• get_historical_features - エンティティとイベントタイムスタンプの entity_df と FeatureView のセットを指定すると、 は、各行に特定の時点の正しい機能値が含まれる拡大テーブルを返します。それぞれの (entity_id, event_timestamp) ペアについて、event_timestamp <= entity_event_timestamp を持つ最新の機能値TTL 内の と が選択されます。

• pull_latest_from_table_or_query - オンライン ストアをシードするために Feast のマテリアライズドエンジンによって使用される 時間ウィンドウの最新の機能値を含むエンティティごとに 1 行を返します。

• pull_all_from_table_or_query - 同じ feature_historyスキーマとインデックスに基づき、エクスポートまたは検査のために指定された日付範囲内のすべての行をデータソースから取得します。

• persist （RetrievalJob.persist経由）- 履歴機能クエリの結果を、SavedDatasetStorage経由で別のコレクションまたは外部シンクに書き込みます。feature_history とは異なります。

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オフライン ストアです。

主な特徴：

• すべての FeatureView で共有される単一の feature_historyコレクションを使用します（feature_view で区別されます）。

• すべてのクエリで複合インデックス(entity_id, feature_view, event_timestamp, created_at) に依存し、コレクションのフル スキャンを回避します。

• 「スコアリング」ワークロード（エンティティごとに 1 行）にはサーバーサイドの $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オンライン ストアでは、シリアル化されたエンティティ キーをキーとするすべての FeatureView に対して単一のコレクションが使用されます。

• _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)"
}

設計理由:

• 単一のコレクションは各エンティティの状態を 1 つのドキュメントで保持するため、キーベースの検索に関する Feast の予想と一致し、FeatureView ごとのコレクション間での状態の断片化を回避します。

• シリアル化されたエンティティキーを _id として使用すると、Feast の決定的エンコーディングが再利用され、コレクション間で重複するプライマリキーが回避され、エンティティごとに単一のキー検索が取得されます。

主要オンライン操作

MongoDBオンライン ストアには、Feast の標準オンライン ストアAPIが実装されています。

• online_write_batch - マテリアライズ中に、Feast は各エンティティの最新の機能値をMongoDBドキュメントに書込みます。各バッチするアップサートでは、関連するネストされた features.<feature_view> サブドキュメントと、event_timestamps 内の対応するエントリのみが更新され、エンティティ ドキュメントはアトミックで一貫性が保たれます。

• online_read と get_online_features - オンラインで提供すると、オフラインと同じ直列化ロジックを使用してエンティティキーを _id 値に解決し、キー検索を実行します。各検索は、ネストされた features 構造を活用して、エンティティに要求された機能を 1 回のラウンド トリップですべて返します。

• 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",
)