Integre o MongoDB com o Feast

Lojas online e offline

• A loja online é um armazenamento de valores-chave apoiado por uma única coleção do MongoDB , otimizado para recuperação de baixa latência dos recursos mais recentes por entidade durante a inferência online.

• A loja offline é uma camada de computação e tradução que query linhas de dados históricos de funcionalidades armazenados em uma collection do MongoDB (normalmente chamada) para conjuntos de dados feature_history de treinamento, pontuação e materialização (promoção de dados para o armazenamento online).

Padrões comuns de fluxo de trabalho

Um fluxo de trabalho típico de ponta a ponta é assim:

1. Defina entidades, visualizações de recursos e fontes de dados que ponto para coleções apoiadas MongoDB.

2. A ingestão de dados de funcionalidades no armazenamento offline por meio do offline_write_batch, que aceita uma tabela PyArrow como entrada e insere os dados na coleção feature_history MongoDB seguindo o esquema do armazenamento offline.

3. Gere dados de treinamento usando o get_historical_features, que executa uma conexão eficiente de ponto no tempo em linhas de recursos históricos armazenadas no MongoDB.

4. Materialize os valores de funcionalidades mais recentes da loja offline para a loja online usando pull_latest_from_table_or_query e online_write_batch.

5. Disponibilize recursos online por meio das APIs online do Feast, que leem de uma única coleção do MongoDB chaveada por uma chave de entidade serializada.

Mapeamento de conceitos

Modelo de dados

O armazenamento offline do MongoDB usa uma única coleção compartilhada (por padrão feature_history) que armazena linhas de recursos históricos somente anexados para todas as visualizações de recursos.

Cada documento representa uma observação de uma entidade para um FeatureView em um registro de data/hora de evento específico:

{
  "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
  }
}

Propriedades principais:

• Somente anexar: os dados históricos são tratados como imutáveis; as correções são escritas como novas linhas com registros de data/hora created_at mais recentes em vez de atualizações no local.

• Compatível com séries temporais: event_timestamp representa quando o valor do recurso foi observado; created_at é usado como desempate quando várias observações compartilham o mesmo carimbo de data/hora do evento .

• Agrupamento de recursos por FeatureView: feature_view identifica a qual FeatureView a linha pertence, de modo que uma única coleção possa hospedar vários FVs.

Um único índice composto suporta todos os principais padrões de query:

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

Esse índice permite varreduras de intervalo eficientes sobre entidades e visualizações de recursos, enquanto garante que a observação mais recente por (entity_id, feature_view) seja vista primeiro durante a agregação.

Principais operações offline

A loja offline do MongoDB implementa a interface de loja offline padrão do Feast:

• offline_write_batch - Grava um pyarrow.Table dos dados de funcionalidades na collection subjacente do MongoDB , usando os metadados MongoDBSource configurados para determinar connection_string, database e collection.

• get_historical_features - Dado um entity_df de entidades e registros de data/hora de evento , além de um conjunto de FeatureViews, retorna uma tabela ampliada em que cada linha inclui valores de recursos corretos point-in-time: para cada par (entity_id, event_timestamp), o valor de recurso mais recente cujo event_timestamp <= entity_event_timestamp e dentro de TTL estiver selecionado.

• pull_latest_from_table_or_query - Retorna uma linha por entidade contendo os valores de funcionalidade mais recentes em uma janela de tempo, usada pelo mecanismo de materialização do Feast para semear a loja online.

• pull_all_from_table_or_query - Recupera todas as linhas de uma fonte de dados em um intervalo de datas especificado para exportação ou inspeção, apoiado pelo mesmo esquema e índice do feature_history.

• persist (via RetrievalJob.persist) - Grava o resultado de uma query de recurso histórico em uma coleção separada ou coletor externo via SavedDatasetStorage, distinto de 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.* }

Implementação de agregação

A implementação offline recomendada é o armazenamento offline do MongoDB baseado em aggregation, denominado MongoDBOfflineStore.

Características principais:

• Utiliza uma única coleção feature_history compartilhada por todos os FeatureViews, distinguidos por feature_view.

• Confia no índice composto (entity_id, feature_view, event_timestamp, created_at) para todas as queries, evitando varreduras completas da coleção.

• Usa o $group $first do lado do servidor para cargas de trabalho de "pontuação" (uma linha por entidade) e pd.merge_asof para cargas de trabalho de "treinamento" com IDs de entidade repetidas, equilibrando correção e desempenho.

• Uso de memória limitada por meio de agrupamento, portanto, valores entity_df grandes podem ser processados sem esgotar a RAM.

Os benchmarks mostram que essa implementação fornece a melhor combinação de taxa de transferência e eficiência de memória em comparação com abordagens offline alternativas do 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"])

Recursos da loja offline

Conveniências adicionais, como exportar diretamente para data lakes ou armazéns, dependem da implementação específica do RetrievalJob e espera-se que siga os padrões padrão da Feast para armazenamentos offline.

Modelo de dados

A loja online MongoDB usa uma única coleção para todos os FeatureViews, codificados pela chave de entidade serializada.

• _id: serialized_entity_key(entity_key), produzido pela função de codificação estável do Feast que classifica os nomes e valores das entidades e os codifica em bytes.

• features: subdocumento aninhado onde cada FeatureView mantém seu próprio namespace de funcionalidade .

• event_timestamps: carimbos de data/hora por FeatureView que indicam quando o valor mais recente para esse FeatureView foi gravado.

• created_timestamp ou updated_at: campos de contabilidade úteis para indexação e diagnóstico de TTL.

Exemplo (simplificado):

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

Fundamento do design:

• Uma única collection mantém o estado de cada entidade em um documento, o que corresponde à expectativa do Feast de pesquisas baseadas em chaves e evita a fragmentação do estado nas collections por FeatureView.

• O uso da chave de entidade serializada como _id reutiliza a codificação determinística do Feast, evita chaves primárias duplicadas entre collections e mantém a recuperação para uma única pesquisa de chave por entidade.

Operações Online Centrais

A loja online do MongoDB implementa a API de loja online padrão da Feast:

• online_write_batch - Durante a materialização, o Feast grava os valores de funcionalidade mais recentes para cada entidade nos documentos do MongoDB . Cada upsert de lote atualiza somente o subdocumento features.<feature_view> aninhado relevante e sua entrada correspondente em event_timestamps, mantendo os documentos da entidade atômicos e consistentes.

• online_read e get_online_features - A entrega online resolve as chaves de entidade em valores _id usando a mesma lógica de serialização que offline e, em seguida, realiza pesquisas de chaves. Cada pesquisa retorna todas as feições solicitadas para a entidade em uma única viagem de ida e volta, aproveitando a estrutura features aninhada.

• TTL e atualização — O recurso TTL é configurado no FeatureView e usado principalmente em junções de PIT offline; O TTL online pode ser implementado com um índice em updated_at ou registro de data e hora semelhante, consistente com a noção da Feast de que as lojas offline são somente para anexação, enquanto as lojas online mantêm o estado mais recente.

Configuração da loja offline

O armazenamento offline é configurado usando MongoDBOfflineStoreConfig:

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

Exemplo 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 é o DataSource correspondente. Seu name campo se torna o feature_view discriminador armazenado em todos os documento. Para obter opções de configuração completas, consulte a referência da Fonte de Dados do MongoDB nos documentos do Feast.

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