Integra MongoDB con Feast

Tiendas en linea y offline

• La tienda en linea es un almacén clave-valor respaldado por una única colección de MongoDB, optimizada para la recuperación de baja latencia de las últimas funcionalidades por entidad durante la inferencia en linea.

• La tienda offline es una capa de computación y traducción que **query** filas de datos históricos de **funcionalidad** almacenados en una **colección** de **MongoDB** (normalmente llamada feature_history) para conjuntos de **datasets** de **entrenamiento**, puntuación y materialización (promocionando datos a la tienda **en linea**).

Patrones comunes de flujo de trabajo

Un flujo de trabajo típico de extremo a extremo se ve así:

1. Define entidades, vistas de características y fuentes de datos que apunten a colecciones respaldadas por MongoDB.

2. Incorpora datos de características en el almacén offline mediante offline_write_batch, que acepta una tabla PyArrow como input e inserta los datos en la colección de MongoDB feature_history siguiendo el esquema del almacén offline.

3. Genera datos de entrenamiento utilizando get_historical_features, que ejecuta una unión eficiente en el tiempo sobre filas de características históricas almacenadas en MongoDB.

4. Materialice los últimos valores de las funcionalidades desde la tienda offline en la tienda en linea mediante pull_latest_from_table_or_query y online_write_batch.

5. Sirve funcionalidades en linea a través de las APIs en linea de Feast, que leen de una única colección de MongoDB indexada por una clave de entidad serializada.

Mapeo conceptual

modelo de datos

La tienda offline de MongoDB utiliza una sola colección compartida (por defecto feature_history) que almacena filas históricas de características sólo de adición para todas las vistas de características.

Cada document representa una observación de una entidad para un FeatureView en una marca de tiempo de evento específica:

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

Propiedades clave:

• Solo añadir: los datos históricos se tratan como inmutables; las correcciones se escriben como nuevas filas con marcas temporales más recientes de created_at en lugar de actualizaciones en el lugar.

• Compatible con series temporales: event_timestamp representa cuándo se observó el valor de la funcionalidad; created_at se utiliza como criterio de desempate cuando varias observaciones comparten la misma marca de tiempo del evento.

• Agrupación por FeatureView: feature_view identifica a qué FeatureView pertenece la fila, por lo que una única colección puede alojar varios FV.

Un solo índice compuesto admite todos los patrones de query principales:

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

Este índice permite escaneos eficientes de rango sobre entidades y vistas de características, mientras garantiza que la observación más reciente por (entity_id, feature_view) se vea primero durante la agregación.

Operaciones principales sin conexión

La tienda offline de MongoDB implementa la interfaz estándar de tienda offline de Feast:

• offline_write_batch - Escribe un pyarrow.Table de datos de funcionalidades en la colección subyacente de MongoDB, utilizando la MongoDBSource configurada de metadatos para determinar connection_string, database y collection.

• get_historical_features - Dada una entity_df de entidades y marcas de tiempo de eventos más un conjunto de FeatureViews, devuelve una tabla ampliada donde cada fila incluye valores de funcionalidad correctos en el punto en el tiempo: para cada par (entity_id, event_timestamp), se selecciona el valor de funcionalidad más reciente cuyo event_timestamp <= entity_event_timestamp y dentro del TTL.

• pull_latest_from_table_or_query - Devuelve una fila por entidad que contiene los valores de características más recientes en una ventana temporal, utilizada por el motor de materialización de Feast para sembrar la tienda en línea.

• pull_all_from_table_or_query - Recupera todas las filas de una fuente de datos en un intervalo de fechas especificado para exportar o inspeccionar, respaldada por el mismo esquema e índice de feature_history.

• persist (a través de RetrievalJob.persist) - Guarda el resultado de una consulta de funcionalidades históricas en una colección separada o en un sumidero externo a través de SavedDatasetStorage, diferente 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.* }

Implementación de agregación

La implementación offline recomendada es la tienda offline basada en agregación de MongoDB, llamada MongoDBOfflineStore.

Características clave:

• Utiliza una única colección feature_history compartida por todas las Visualizaciones de Características, diferenciada por feature_view.

• Se apoya en el índice compuesto (entity_id, feature_view, event_timestamp, created_at) para todas las consultas, evitando el escaneo completo de la colección.

• Utiliza $group $first del lado del servidor para cargas de trabajo de "puntuación" (una fila por entidad), y pd.merge_asof para cargas de trabajo de "entrenamiento" con identificadores de entidad repetidos, equilibrando la corrección y el rendimiento.

• Uso de memoria limitado mediante el procesamiento por partes, para que se puedan procesar grandes valores de entity_df sin agotar la memoria RAM.

Los benchmarks muestran que esta implementación ofrece la mejor combinación de rendimiento y eficiencia de memoria en comparación con los enfoques alternativos de MongoDB offline.

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"])

Capacidades de la tienda offline

Las comodidades adicionales, como la exportación directa a lagos de datos o almacenes, dependen de la implementación específica de RetrievalJob y se espera que sigan los patrones estándar de Feast para almacenes offline.

modelo de datos

La tienda en línea de MongoDB utiliza una única colección para todas las FeatureViews, indexada por la clave de entidad serializada.

• _id: serialized_entity_key(entity_key), producido por la función de codificación estable de Feast que ordena los nombres y valores de las entidades y los codifica en bytes.

• featuressubdocumento anidado donde cada FeatureView mantiene su propio namespace de funcionalidades.

• event_timestamps● timestamps por FeatureView que indican cuándo se escribió el último valor para ese FeatureView.

• created_timestamp o updated_at: campos de contabilidad útiles para la indexación TTL y la diagnostica.

Ejemplo (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)"
}

Justificación del diseño:

• Una única colección mantiene el estado de cada entidad en un solo document, lo que coincide con la expectativa de Feast de búsquedas basadas en claves y evita el fraccionamiento del estado entre colecciones por FeatureView.

• El uso de la clave de entidad serializada como _id reutiliza la codificación determinista de Feast, evita llaves primarias duplicadas en diferentes colecciones y mantiene la recuperación en solo una búsqueda de clave por entidad.

Operaciones en linea principales

La tienda en línea de MongoDB implementa la API estándar de la tienda en línea de Feast:

• online_write_batch - Durante la materialización, Feast guarda los valores actuales de las funcionalidades de cada entidad en los documents de MongoDB. Cada actualización de upsert por lote actualiza solo el subdocumento anidado correspondiente features.<feature_view> y su entrada correspondiente en event_timestamps, manteniendo los document de la entidad atómicos y consistentes.

• online_read y get_online_features: la prestación en línea resuelve las claves de entidad en valores _id utilizando la misma lógica de serialización que fuera de línea y, a continuación, realiza búsquedas de claves. Cada búsqueda devuelve todas las funcionalidades solicitadas para la entidad en un solo viaje de ida y vuelta, aprovechando la estructura anidada features.

• TTL y frescura: el TTL de la funcionalidad se configura en el FeatureView y se utiliza principalmente en uniones PIT fuera de línea; el TTL en linea puede implementarse con un índice en updated_at o en una marca de tiempo similar, en consonancia con la noción de Feast de que los almacenes fuera de línea solo agregan, mientras que los almacenes en linea mantienen el estado más reciente.

Configuración de la tienda offline

La tienda offline se configura usando MongoDBOfflineStoreConfig:

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

Ejemplo 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 es el DataSource correspondiente. Su name campo se convierte en el discriminador feature_view almacenado en cada document. Para conocer todas las opciones de configuración, consulta la referencia MongoDB fuente de datos en los Feast Docs.

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