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 casi 1,9M 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 casi 600 fraudes no detectados al mes. Son 1.9M 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("country_code", StringType(), True),
23    StructField("status", StringType(), False),
24])
25
26# Leer del topic Kafka (reutilizable desde el día 1)
27raw_stream = spark.readStream \
28    .format("kafka") \
29    .option("kafka.bootstrap.servers",
30            "b-1.neobank-msk.kafka.eu-west-1.amazonaws.com:9092") \
31    .option("subscribe", "transactions.completed") \
32    .option("startingOffsets", "latest") \
33    .load()
34
35# Parsear y preparar
36from pyspark.sql.functions import from_json
37parsed = raw_stream \
38    .selectExpr("CAST(value AS STRING) as json_str") \
39    .select(from_json(col("json_str"), txn_schema).alias("txn")) \
40    .select("txn.*") \
41    .withColumn("event_time", to_timestamp(col("timestamp"))) \
42    .filter(col("status") == "completed")
43
44# === FEATURES HOT: ventana deslizante 1h, slide 1min, watermark 5min ===
45hot_features = parsed \
46    .withWatermark("event_time", "5 minutes") \
47    .groupBy(
48        col("account_id"),
49        window(col("event_time"), "1 hour", "1 minute")
50    ) \
51    .agg(
52        count("*").alias("txn_count_1h"),
53        avg("amount").alias("avg_amount_1h"),
54        approx_count_distinct("merchant_id").alias("distinct_merchants_1h"),
55    )
56
57# Escribir a Redis via forEach (simplificado aquí como console para demo)
58# En producción: hot_features.writeStream.foreach(RedisSink()).start()
59# (RedisSink lo implementas esta tarde, ejercicio 3 de la lección de streaming)
60query = hot_features.writeStream \
61    .format("console") \
62    .outputMode("update") \
63    .trigger(processingTime="30 seconds") \
64    .start()
65
66print("Feature Store HOT iniciado:")
67print("  - txn_count_1h: transacciones en última hora")
68print("  - avg_amount_1h: importe promedio en última hora")
69print("  - distinct_merchants_1h: merchants únicos en última hora")
70print("  - Ventana: 1h sliding, 1min step")
71print("  - Watermark: 5 min (tolera late data)")
72print("  - 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": 300,
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": 300,
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": 300,
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 cuestan memoria en dos sitios, y el segundo es el que sorprende. En Redis, la cuenta es sencilla: 2,1M clientes × 5 features = 10,5M claves, unos 100 bytes por clave = 1 GB; con TTLs y picos puedes plantarte en 4-5 GB. Monitoriza redis_used_memory desde el día 1 y alerta al 70%. En Spark la cuenta es peor: tu ventana de 1 hora con slide de 1 minuto significa que cada transacción cae en 60 ventanas solapadas, y el state store guarda una entrada por cada pareja (cliente, ventana). Con 100K clientes activos por hora, no son 100K entradas de estado: son 6 millones. Cada una guarda count, media y el sketch de approx_count_distinct. Varios gigas de estado en los ejecutores. Qué hacer: sube el slide si el modelo lo tolera (de 1 min a 5 divide el estado por cinco), vigila stateOperators en las métricas del stream (no solo el consumer lag), y recuerda que al reiniciar de cero el job tarda una hora en tener la ventana llena. La pregunta de Alberto: «¿y si escala a 10x?». Con 10x, Redis pasa de 1 GB a 10, y el estado de Spark de 6M de entradas a 60M.

Regístrate para guardar tu progreso.