Integrar MongoDB con Feast

Tiendas online y físicas

• La tienda en línea es un almacén de clave-valor respaldado por una única colección de MongoDB, optimizado para la recuperación de baja latencia de las últimas características por entidad durante la inferencia en línea.

• El almacén sin conexión es una capa de computación y traducción que consulta filas de datos históricos de características almacenadas en una colección de MongoDB (normalmente llamada feature_history) para conjuntos de datos de entrenamiento, puntuación y materialización (promoción de datos a la tienda en línea).

Patrones comunes de flujo de trabajo

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

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

2. Ingiere datos de características en el almacén fuera de línea a través de offline_write_batch, que acepta una tabla PyArrow como entrada e inserta los datos en la colección feature_history de MongoDB siguiendo el esquema del almacén fuera de línea.

3. Genera datos de entrenamiento usando get_historical_features, que realiza una unión puntual eficiente sobre filas de características históricas almacenadas en MongoDB.

4. Materializa los últimos valores de características de la tienda física en la tienda en línea usando pull_latest_from_table_or_query y online_write_batch.

5. Las funcionalidades se ofrecen en línea a través de las API en línea de Feast, que leen datos de una única colección de MongoDB indexada por una clave de entidad serializada.

Mapeo conceptual

modelo de datos

El almacenamiento sin conexión de MongoDB utiliza una única colección compartida (por defecto feature_history) que almacena filas de características históricas de solo adición para todas las vistas de características.

Cada documento representa una observación de una entidad para una 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 anexar: los datos históricos se tratan como inmutables; las correcciones se escriben como nuevas filas con marcas de tiempo created_at más recientes en lugar de actualizaciones en el mismo lugar.

• Compatible con series temporales: event_timestamp representa el momento en que se observó el valor de la característica; created_at se utiliza como criterio de desempate cuando varias observaciones comparten la misma marca de tiempo del evento.

• Agrupación de características por FeatureView: feature_view identifica a qué FeatureView pertenece la fila, por lo que una sola colección puede albergar múltiples FV.

Un único índice compuesto admite todos los patrones de consulta principales:

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

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

Operaciones principales fuera de línea

El almacén sin conexión de MongoDB implementa la interfaz estándar de almacén sin conexión de Feast:

• offline_write_batch - Escribe un pyarrow.Table de datos de características en la colección subyacente de MongoDB, utilizando los metadatos MongoDBSource configurados para determinar connection_string, database y collection.

• get_historical_features - Dado un 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 características correctos en un punto en el tiempo: para cada par (entity_id, event_timestamp), se selecciona el valor de característica más reciente cuyo event_timestamp <= entity_event_timestamp y dentro de TTL.

• pull_latest_from_table_or_query - Devuelve una fila por entidad que contiene los valores de características más recientes en un intervalo de tiempo, utilizados por el motor de materialización de Feast para alimentar la tienda en línea.

• pull_all_from_table_or_query - Recupera todas las filas de una fuente de datos en un rango de fechas especificado para su exportación o inspección, respaldadas por el mismo esquema feature_history e índice.

• persist (a través de RetrievalJob.persist) - Escribe el resultado de una consulta de características históricas en una colección separada o un destino externo a través de 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.* }

Implementación de agregación

La implementación sin conexión recomendada es el almacén sin conexión de MongoDB basado en agregación, llamado MongoDBOfflineStore.

Características clave:

• Utiliza una única colección feature_history compartida por todas las FeatureViews, distinguida por feature_view.

• Se basa en el índice compuesto (entity_id, feature_view, event_timestamp, created_at) para todas las consultas, evitando escaneos completos 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 ID de entidad repetidos, equilibrando la corrección y el rendimiento.

• El uso limitado de memoria mediante la división en bloques permite procesar valores entity_df grandes sin agotar la RAM.

Las pruebas de rendimiento demuestran que esta implementación ofrece la mejor combinación de rendimiento y eficiencia de memoria en comparación con otros enfoques sin conexión para 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"])

Capacidades de la tienda física

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

modelo de datos

La tienda online 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.

• features: subdocumento anidado donde cada FeatureView mantiene su propio espacio de nombres de características.

• event_timestamps: marcas de tiempo por FeatureView que indican cuándo se escribió el último valor para esa FeatureView.

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

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

Fundamentación del diseño:

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

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

Operaciones principales en línea

La tienda online de MongoDB implementa la API estándar de tienda online de Feast:

• online_write_batch - Durante la materialización, Feast escribe los valores de características más recientes para cada entidad en documentos de MongoDB. Cada operación de inserción/actualización por lotes actualiza solo el subdocumento anidado features.<feature_view> relevante y su entrada correspondiente en event_timestamps, manteniendo los documentos de entidad atómicos y consistentes.

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

• TTL y actualidad: el TTL de la característica se configura en FeatureView y se utiliza principalmente en uniones PIT fuera de línea; el TTL en línea se puede implementar con un índice en updated_at o una marca de tiempo similar, de acuerdo con la noción de Feast de que los almacenes fuera de línea son solo de adición, mientras que los almacenes en línea mantienen el estado más reciente.

Configuración de la tienda sin conexión

La tienda sin conexión está configurada 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 feature_view discriminador almacenado en cada documento. Para ver todas las opciones de configuración, consulte la referencia de la fuente de datos de MongoDB en la documentación de Feast.

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