lección 4

Día 3 -- Feature Store: arquitectura hot/cold para ML

Diseñas el Feature Store con capas hot/warm/cold para el modelo de fraude de Miguel. Implementas las 5 features y el registry de metadata.

⏱ 45 min

### Jueves, 11:00 AM -- La reunión con Miguel

Después del Architecture Review (donde Clara aprobó tu ADR), Miguel te intercepta en el pasillo. "Ey, ¿tienes 30 minutos? Necesito explicarte exactamente qué features necesito y por qué. Si me das las features equivocadas, el modelo no mejora -- y seguimos perdiendo 2.1M al mes en fraude no detectado."

1Miguel Ortiz (sala de café, 11:05 AM):
2"Mi modelo detecta fraude con 92% precisión si le doy las
3features en 5 minutos. Con batch diario, la precisión baja a
467%. Eso son 340 fraudes no detectados al mes. Son 2.1M EUR
5de pérdidas.
6
7Necesito 5 features específicas:
8
9HOT (ventana 1 hora, actualización cada 30s):
101. txn_count_1h - Número de transacciones del cliente en 1h
112. avg_amount_1h - Importe promedio en 1h
123. distinct_merchants_1h - Merchants únicos en 1h
13
14WARM (ventana 24h, actualización cada 15 min):
154. txn_velocity_24h - Ratio transacciones/hora en 24h
16
17COLD (ventana 30 días, actualización batch nocturno):
185. avg_monthly_spend - Gasto promedio mensual
19
20¿Por qué estas 5? Porque los patrones de fraude son:
21- Muchas txns pequeñas rápidas (test de tarjeta robada)
22- Txn grande en merchant nuevo
23- Velocidad anómala respecto al histórico
24- Gasto que no cuadra con el perfil mensual
25
26Mi modelo hace scoring en <50ms. Necesito las features
27pre-calculadas en Redis con latencia <10ms por feature."

Miguel explica las 5 features y por qué cada una importa

### Arquitectura del Feature Store

Te sientas con Alberto y Miguel a diseñar. El concepto central es separar las features por "temperatura" -- la frecuencia con la que se actualizan y la latencia máxima aceptable para servirlas al modelo:

Feature Store NeoBank: hot/warm/cold con serving unificado para el modelo de fraude

### Implementación de features hot con Spark Streaming

La primera prioridad son las 3 features hot -- las que más impacto tienen en la precisión del modelo. Se calculan con ventana deslizante de 1 hora y watermark de 5 minutos (tolerancia a eventos que llegan tarde):

1from pyspark.sql import SparkSession
2from pyspark.sql.functions import (
3    col, window, count, avg, approx_count_distinct,
4    to_timestamp, current_timestamp, struct, to_json
5)
6from pyspark.sql.types import StructType, StructField, StringType, DoubleType
7
8spark = SparkSession.builder \
9    .appName("neobank-feature-store-hot") \
10    .config("spark.sql.streaming.checkpointLocation",
11            "s3://neobank-checkpoints/feature-store-hot/") \
12    .getOrCreate()
13
14# Schema de transacción (idéntico al job de streaming del día 1)
15txn_schema = StructType([
16    StructField("transaction_id", StringType(), False),
17    StructField("account_id", StringType(), False),
18    StructField("merchant_id", StringType(), False),
19    StructField("amount", DoubleType(), False),
20    StructField("currency", StringType(), False),
21    StructField("timestamp", StringType(), False),
22    StructField("status", StringType(), False),
23])
24
25# Leer del topic Kafka (reutilizable desde el día 1)
26raw_stream = spark.readStream \
27    .format("kafka") \
28    .option("kafka.bootstrap.servers",
29            "b-1.neobank-msk.kafka.eu-west-1.amazonaws.com:9092") \
30    .option("subscribe", "transactions.completed") \
31    .option("startingOffsets", "latest") \
32    .load()
33
34# Parsear y preparar
35from pyspark.sql.functions import from_json
36parsed = raw_stream \
37    .selectExpr("CAST(value AS STRING) as json_str") \
38    .select(from_json(col("json_str"), txn_schema).alias("txn")) \
39    .select("txn.*") \
40    .withColumn("event_time", to_timestamp(col("timestamp"))) \
41    .filter(col("status") == "completed")
42
43# === FEATURES HOT: ventana deslizante 1h, slide 1min, watermark 5min ===
44hot_features = parsed \
45    .withWatermark("event_time", "5 minutes") \
46    .groupBy(
47        col("account_id"),
48        window(col("event_time"), "1 hour", "1 minute")
49    ) \
50    .agg(
51        count("*").alias("txn_count_1h"),
52        avg("amount").alias("avg_amount_1h"),
53        approx_count_distinct("merchant_id").alias("distinct_merchants_1h"),
54    )
55
56# Escribir a Redis via forEach (simplificado aquí como console para demo)
57# En producción: hot_features.writeStream.foreach(RedisSink()).start()
58query = hot_features.writeStream \
59    .format("console") \
60    .outputMode("update") \
61    .trigger(processingTime="30 seconds") \
62    .start()
63
64print("Feature Store HOT iniciado:")
65print("  - txn_count_1h: transacciones en última hora")
66print("  - avg_amount_1h: importe promedio en última hora")
67print("  - distinct_merchants_1h: merchants únicos en última hora")
68print("  - Ventana: 1h sliding, 1min step")
69print("  - Watermark: 5 min (tolera late data)")
70print("  - Destino: Redis (key: feat:{account_id}:{feature_name})")

Features hot calculadas con Spark Structured Streaming -- ventana deslizante + watermark

### Feature cold: avg_monthly_spend (SQL en batch nocturno)

1-- Feature COLD: avg_monthly_spend
2-- Ejecutada por DAG nocturno "feature_store_cold" a las 4:30 AM
3-- Calcula gasto promedio mensual últimos 30 días por cliente activo
4
5INSERT INTO feature_store.cold_features 
6    (account_id, feature_name, feature_value, computed_at, valid_until)
7SELECT 
8    account_id,
9    'avg_monthly_spend' as feature_name,
10    ROUND(AVG(daily_spend) * 30, 2) as feature_value,
11    CURRENT_TIMESTAMP as computed_at,
12    CURRENT_TIMESTAMP + INTERVAL '24 hours' as valid_until
13FROM (
14    SELECT 
15        account_id,
16        DATE(timestamp) as txn_date,
17        SUM(amount) as daily_spend
18    FROM gold.transactions
19    WHERE timestamp >= CURRENT_DATE - INTERVAL '30 days'
20      AND status = 'completed'
21      AND currency = 'EUR'
22    GROUP BY account_id, DATE(timestamp)
23) daily_totals
24GROUP BY account_id
25HAVING COUNT(DISTINCT txn_date) >= 7  -- Solo clientes con actividad mínima
26ON CONFLICT (account_id, feature_name) 
27DO UPDATE SET 
28    feature_value = EXCLUDED.feature_value,
29    computed_at = EXCLUDED.computed_at,
30    valid_until = EXCLUDED.valid_until;
31
32-- Verificar: ¿cuántos clientes tienen feature calculada?
33SELECT count(*) as clientes_con_feature,
34       avg(feature_value) as avg_spend_global,
35       percentile_cont(0.5) WITHIN GROUP (ORDER BY feature_value) as median_spend
36FROM feature_store.cold_features
37WHERE feature_name = 'avg_monthly_spend'
38  AND valid_until > CURRENT_TIMESTAMP;

Feature cold calculada en batch nocturno -- SQL sobre Redshift

### Feature Registry -- el catálogo de features

Alberto insiste: "Cada feature debe estar documentada en un registry. No solo para Miguel -- para que dentro de 6 meses alguien nuevo sepa qué features existen, quién las mantiene y cómo se calculan. Un feature store sin catálogo es una bomba de relojería."

1# Feature Registry -- metadata centralizada de cada feature
2# En producción esto sería una tabla en Redshift o un servicio tipo Feast
3
4FEATURE_REGISTRY = {
5    "txn_count_1h": {
6        "description": "Número de transacciones del cliente en la última hora",
7        "type": "hot",
8        "source": "Spark Structured Streaming (job: feature-store-hot)",
9        "window": "1 hour sliding, 1 minute step",
10        "watermark": "5 minutes",
11        "storage": "Redis (key: feat:{account_id}:txn_count_1h)",
12        "ttl_seconds": 3600,
13        "owner": "data-engineering",
14        "consumers": ["fraud-model-v2", "alerting-service"],
15        "sla_freshness": "< 60 seconds",
16        "data_type": "integer",
17        "typical_range": [0, 50],
18        "anomaly_threshold": 30,  # Más de 30 txns/hora es sospechoso
19    },
20    "avg_amount_1h": {
21        "description": "Importe promedio de transacciones en la última hora",
22        "type": "hot",
23        "source": "Spark Structured Streaming (job: feature-store-hot)",
24        "window": "1 hour sliding, 1 minute step",
25        "watermark": "5 minutes",
26        "storage": "Redis (key: feat:{account_id}:avg_amount_1h)",
27        "ttl_seconds": 3600,
28        "owner": "data-engineering",
29        "consumers": ["fraud-model-v2"],
30        "sla_freshness": "< 60 seconds",
31        "data_type": "float",
32        "typical_range": [10.0, 500.0],
33        "anomaly_threshold": 5000.0,
34    },
35    "distinct_merchants_1h": {
36        "description": "Merchants únicos donde el cliente transacciónó en 1h",
37        "type": "hot",
38        "source": "Spark Structured Streaming (job: feature-store-hot)",
39        "window": "1 hour sliding, 1 minute step",
40        "watermark": "5 minutes",
41        "storage": "Redis (key: feat:{account_id}:distinct_merchants_1h)",
42        "ttl_seconds": 3600,
43        "owner": "data-engineering",
44        "consumers": ["fraud-model-v2"],
45        "sla_freshness": "< 60 seconds",
46        "data_type": "integer",
47        "typical_range": [1, 5],
48        "anomaly_threshold": 10,
49    },
50    "txn_velocity_24h": {
51        "description": "Ratio transacciones/hora en las últimas 24h",
52        "type": "warm",
53        "source": "Micro-batch Spark (cada 15 minutos)",
54        "window": "24 hours",
55        "storage": "Redis (key: feat:{account_id}:txn_velocity_24h, TTL 900s)",
56        "ttl_seconds": 900,
57        "owner": "data-engineering",
58        "consumers": ["fraud-model-v2", "risk-scoring-service"],
59        "sla_freshness": "< 15 minutes",
60        "data_type": "float",
61        "typical_range": [0.1, 3.0],
62        "anomaly_threshold": 10.0,
63    },
64    "avg_monthly_spend": {
65        "description": "Gasto promedio mensual (últimos 30 días)",
66        "type": "cold",
67        "source": "Batch nocturno SQL (DAG: feature_store_cold, 4:30 AM)",
68        "window": "30 days",
69        "storage": "Redshift (feature_store.cold_features) + Redis cache TTL 24h",
70        "ttl_seconds": 86400,
71        "owner": "data-engineering",
72        "consumers": ["fraud-model-v2", "credit-scoring", "marketing-segmentation"],
73        "sla_freshness": "< 24 hours",
74        "data_type": "float",
75        "typical_range": [200.0, 5000.0],
76        "anomaly_threshold": 20000.0,
77    },
78}
79
80# Imprimir registry
81print("╔══════════════════════════════════════════════════════════════╗")
82print("║  FEATURE REGISTRY -- NeoBank Fraud Detection               ║")
83print("╠══════════════════════════════════════════════════════════════╣")
84for name, meta in FEATURE_REGISTRY.items():
85    tipo = meta["type"].upper()
86    print(f"║  [{tipo:4s}] {name:25s} SLA: {meta['sla_freshness']:15s} ║")
87print("╚══════════════════════════════════════════════════════════════╝")

Feature Registry con metadata completa para cada feature

Un feature store sin registry es como una biblioteca sin catálogo. El registry evita que distintos equipos calculen la misma feature de formas diferentes (feature drift silencioso). En empresas maduras se usan herramientas como Feast, Tecton o Hopsworks. Un diccionario Python es un buen comienzo -- pero documéntalo en Confluence para que sea descubrible.

Las features hot tienen un coste de Redis que escala con clientes activos. Con 2.1M clientes x 5 features = ~10.5M keys. A 100 bytes por key = ~1GB. Parece poco, pero con TTLs, picos de tráfico y las estructuras internas de Redis, puedes llegar a 4-5GB fácilmente. Monitoriza redis_used_memory desde el día 1 y configura alertas al 70% de capacidad.

## ejercicios

[01]

Implementar features hot con ventana deslizante

Completa el job de Spark que calcula las 3 features hot usando ventanas deslizantes con watermark. Usa un DataFrame de test para verificar los resultados.

Cargando editor...

[02]

Feature Registry con validación

Implementa un FeatureRegistry que almacena metadata, válida campos obligatorios, permite buscar por tipo y detecta features con SLA vencido.

Cargando editor...

[03]

Feature cold: avg_monthly_spend

Escribe la query SQL que calcula avg_monthly_spend para todos los clientes activos e inserta en la tabla feature_store con upsert.

Cargando editor...

Regístrate para guardar tu progreso.

## comentarios

Reporta erratas, ayuda a otros o comparte tu opinión. Sé constructivo.

Inicia sesión para comentar y responder.

cargando comentarios...