Saltar al contenido
~/bigdatastack

lección 6

Día 4 (3AM) — El incidente: un solo cliente tumba el pipeline

PagerDuty te despierta: OOM en el streaming job. Diagnóstico bajo presión, data skew por merchant, hotfix con cuarentena y salting como solución permanente.

⏱ 45 min

### Viernes, 3:12 AM — PagerDuty

Tu teléfono vibra con la intensidad de una alarma nuclear. La pantalla muestra: "CRITICAL — neobank-txn-streaming OOM killed. Consumer lag > 50,000 events. Feature store stale > 15 min." Te despiertas de golpe con el corazón a 180. El job de streaming que desplegaste ayer — tu pipeline, tu código, tu responsabilidad — se ha caído en producción.

1# Slack #incidents -- 3:14 AM
2Paula García:
3"Vi la alerta. Estoy mirando los logs de EMR. Executor 3 OOM
4killed a las 03:08. El job no se ha recuperado. Consumer lag
5subiendo: 50K... 80K... ya va por 120K eventos sin procesar.
6

7Te necesito online YA. Tú conoces el job mejor que nadie --
8acabas de escribirlo. Yo abro el Spark UI y CloudWatch.
9Tú miras los datos del último micro-batch exitoso vs el fallido."
10

11Tú (3:16 AM):
12"Estoy. Dame 2 min que abro el portátil. ¿El batch nocturno
13también cayó o es solo streaming?"
14

15Paula García:
16"Solo streaming. Batch está programado para las 4:00, todavía
17no ha arrancado. Pero si no restauramos antes de las 4:00,
18el batch va a intentar leer de un topic con 200K de lag y
19puede cascadear. Reloj corriendo."

Slack del incidente — Paula y tú coordinando a las 3AM

### Diagnóstico — el merchant maldito

Abres Spark UI y CloudWatch en paralelo. La timeline del incidente se reconstruye rápido:

  • 02:47 — Micro-batch inusualmente grande entra en procesamiento
  • 02:53 — Executor 3 sube a 14.2GB de memoria (límite configurado: 15GB)
  • 03:01 — GC pauses de >30 segundos. Executor no responde al heartbeat del driver
  • 03:08 — Spark mata executor 3 por OOM. Tasks se redistribuyen a executor 1 y 2
  • 03:09 — Executor 1 también OOM al recibir las tasks redistribuidas (cascada)
  • 03:12 — Job completo falla. PagerDuty dispara alerta a Paula y a ti
  • 03:12 — Consumer lag empieza a crecer: 50K, 100K, 150K, 200K eventos sin procesar

Comparas los micro-batches de las últimas horas. Desde las 02:47 todos llegan al tope: 50.000 eventos, uno detrás de otro, sin bajar. El maxOffsetsPerTrigger está haciendo su trabajo — ningún batch es monstruoso — pero la cola de Kafka no deja de crecer, así que algo está entrando mucho más rápido de lo que el pipeline puede sacar. Miras qué hay dentro: 2,8 millones de transacciones en seis minutos, todas del mismo merchant MERCH-48291, y todas — esto es lo que te hiela — de la misma cuenta, acc_00007741. Todas de 0,01 EUR. Un merchant normal genera entre 100 y 10.000 transacciones al día. Este ha generado 2,8 millones en seis minutos contra una única tarjeta. Eso no es tráfico: es un TPV reintentando el mismo cobro en bucle. Y aquí es donde tu pipeline se rompe. El freno de entrada aguanta, pero el groupBy de tu agregación va por account_id, y TODAS esas filas tienen el mismo account_id. Batch tras batch, la misma partición, el mismo executor. El tope de 50.000 limita lo que entra por la puerta; no limita lo que se apila en la habitación del fondo.

1# Diagnóstico ejecutado a las 3:30 AM (query sobre S3 del último batch exitoso)
2import pandas as pd
3

4# Distribución de los micro-batches problemáticos (02:47-03:08)
5batch_stats = pd.DataFrame({
6    "merchant_id": ["MERCH-48291", "MERCH-10001", "MERCH-22340", 
7                    "MERCH-15678", "MERCH-09923", "MERCH-31445"],
8    "txn_count": [2_800_000, 10_200, 8_750, 7_100, 12_400, 9_300],
9    "avg_amount_eur": [0.01, 45.30, 128.50, 22.10, 67.80, 34.50],
10    "window_minutes": [6, 60, 60, 60, 60, 60],
11})
12

13total = batch_stats["txn_count"].sum()
14normal_avg = batch_stats[batch_stats["merchant_id"] != "MERCH-48291"]["txn_count"].mean()
15skew_ratio = 2_800_000 / normal_avg
16

17print("═══ DIAGNÓSTICO DEL INCIDENTE ═══")
18print(f"Total transacciones en periodo: {total:,}")
19print(f"MERCH-48291: {2_800_000:,} txns en 6 minutos")
20print(f"  --> {2_800_000/6:.0f} txns/minuto ({2_800_000/360:.0f} txns/segundo)")
21print(f"  --> Importe promedio: 0.01 EUR (micro-transacciones)")
22print(f"  --> TODAS de la misma cuenta: acc_00007741")
23print(f"Promedio otros merchants: {normal_avg:,.0f} txns/hora")
24print(f"Skew ratio: {skew_ratio:.0f}x")
25print()
26print("CAUSA RAÍZ IDENTIFICADA:")
27print("  Bug en TPV de MERCH-48291 genera bucle de retry infinito.")
28print("  Cada retry crea una transacción de 0.01 EUR contra acc_00007741.")
29print("  El job agrupa por account_id en el groupBy.")
30print("  -> 2.8M eventos de UNA cuenta van al MISMO executor -> OOM.")
31print("  maxOffsetsPerTrigger (50K) frena la entrada, pero el estado")
32print("  del groupBy se acumula batch tras batch en el mismo executor.")
33print()
34print(f"IMPACTO:")
35print(f"  - Features de fraude stale: {47} minutos (SLA: 5 min)")
36print(f"  - Transacciones en cola: ~{200_000:,} (Kafka retiene 7 días)")
37print(f"  - Transacciones perdidas: 0 (Kafka retention = safety net)")
38print(f"  - Impacto en clientes: NINGUNO (modelo usó features cached)")

Diagnóstico: acc_00007741 (MERCH-48291) con skew de 293x causó OOM en cascada

### Hotfix — restaurar servicio en minutos

A las 3:45 decides el hotfix con Paula. La prioridad absoluta es restaurar el servicio — la solución elegante vendrá después. El hotfix: filtrar temporalmente MERCH-48291 del stream principal y enviar sus transacciones a un path de cuarentena en S3 (no se pierden, solo se aparcan):

1# HOTFIX -- PR de emergencia aprobado por Paula a las 3:52 AM
2# Restaura el servicio filtrando el merchant problemático
3

4# Lista externalizada en SSM Parameter Store (no requiere redeploy para añadir)
5BLOCKED_MERCHANTS = ["MERCH-48291"]  # Config dinámica via AWS SSM
6

7# Flujo principal: filtrar merchants bloqueados
8# El filtro va ANTES de deduplicar, no después. Filtrar al final quita las filas
9# del resultado pero no del estado: el merchant bloqueado seguiría ocupando
10# memoria en el state store, que es justo lo que se ha caído.
11main_stream = parsed.filter(
12    ~col("merchant_id").isin(BLOCKED_MERCHANTS)
13)
14

15# Cuarentena: capturar merchants bloqueados (NO perder datos)
16quarantine_stream = parsed.filter(
17    col("merchant_id").isin(BLOCKED_MERCHANTS)
18)
19

20# Escribir cuarentena a ruta separada para análisis posterior
21quarantine_query = quarantine_stream.writeStream \
22    .format("parquet") \
23    .option("path", "s3://neobank-lake/quarantine/blocked_merchants/") \
24    .option("checkpointLocation", "s3://neobank-checkpoints/quarantine/") \
25    .trigger(processingTime="30 seconds") \
26    .outputMode("append") \
27    .start()
28

29# Timeline del hotfix:
30# 03:45 - Decisión tomada
31# 03:47 - Código escrito (5 líneas de cambio)
32# 03:49 - PR abierto con descripción del incidente
33# 03:52 - Paula aprueba (code review de emergencia)
34# 03:55 - Deploy via GitHub Actions -> EMR
35# 03:56 - Job reinicia desde último checkpoint
36# 03:57 - Consumer lag empieza a bajar: 200K -> 100K -> 50K
37# 03:59 - Lag en 0. Servicio completamente restaurado.
38# Total downtime: 47 minutos.
39

40print("HOTFIX DESPLEGADO - Servicio restaurado a las 03:59")
41print(f"  Merchants bloqueados: {BLOCKED_MERCHANTS}")
42print(f"  Datos en cuarentena: s3://neobank-lake/quarantine/")
43print(f"  Consumer lag: 0 (restaurado)")

Hotfix: filtrar + cuarentena. Servicio restaurado en 47 minutos.

### Los tres sitios donde un job de streaming se queda sin memoria

Cuando un executor muere por OOM, la pregunta no es «cuántos datos hay»: es dónde están. En un pipeline de Structured Streaming hay tres depósitos distintos, y cada uno se controla con una palanca distinta. Confundirlos es la forma más rápida de aplicar el arreglo equivocado.

  1. 01.El micro-batch. Lo que Spark lee de Kafka en cada disparo. Crece con el volumen de entrada y se vacía en cuanto el batch termina. Palanca: maxOffsetsPerTrigger. Nosotros lo tenemos en 50.000, así que este depósito tiene techo: 50.000 × ~500 B ≈ 24 MB. Este NO puede ser la causa.
  2. 02.La partición del shuffle. Cuando haces groupBy, todas las filas de la misma clave viajan al mismo executor. Si una clave concentra la mitad de los datos, un executor hace la mitad del trabajo. No crece con el volumen total: crece con lo desigual que sea el reparto. Palanca: salting.
  3. 03.El state store. Lo que Spark recuerda ENTRE batches: los transaction_id que ya viste (deduplicación) y los acumuladores de cada ventana abierta (agregación). No se vacía al terminar el batch: se vacía cuando el watermark pasa de largo. Palanca: el watermark, y el número de claves distintas que metes en él.

La regla práctica: si el OOM llega de golpe con un pico de entrada, mira el micro-batch. Si llega con carga normal pero repartida mal, mira el shuffle. Si llega poco a poco sin que el tráfico haya subido, mira el state store. Lo de esta noche es un caso de shuffle en estado puro: el freno de entrada aguantó; lo que mató al executor fue una sola clave del groupBy recibiéndolo todo, batch tras batch, durante veintiún minutos.

### Solución permanente — salting para data skew

El hotfix restaura el servicio pero no resuelve el problema de fondo. Si otro merchant tiene un bug similar mañana, volverá a pasar. La solución permanente es "salting": distribuir las transacciones de un merchant con alto volumen en múltiples sub-particiones artificiales para que ningún executor reciba toda la carga solo.

1from pyspark.sql.functions import (
2    col, concat, lit, floor, rand, count, avg,
3    approx_count_distinct, window, to_timestamp, sum as spark_sum
4)
5

6# SOLUCIÓN PERMANENTE: Salting para data skew
7SALT_BUCKETS = 16  # Número de sub-particiones
8

9def apply_salting_to_stream(stream_df):
10    """
11    Aplica salting al groupBy para distribuir merchants con skew.
12    
13    Idea: en lugar de groupBy(account_id), hacemos:
14    1. Añadir salt aleatorio (0 a N-1) como columna
15    2. groupBy(account_id, salt) -> agregación parcial distribuida
16    3. groupBy(account_id) -> combinar parciales (segunda agregación)
17    
18    Sin salting: 2.8M txns de MERCH-48291 -> 1 executor -> OOM
19    Con salting (16 buckets): ~175K txns por executor -> manejable
20    """
21    # Paso 1: Añadir salt aleatorio
22    salted = stream_df.withColumn(
23        "salt", floor(rand() * SALT_BUCKETS).cast("int")
24    )
25    
26    # Paso 2: Primera agregación con salt (distribuida en N particiones)
27    partial_agg = salted \
28        .withWatermark("event_time", "5 minutes") \
29        .groupBy(
30            col("account_id"),
31            col("salt"),
32            window(col("event_time"), "1 hour", "1 minute")
33        ) \
34        .agg(
35            count("*").alias("partial_txn_count"),
36            spark_sum("amount").alias("partial_amount_sum"),
37            count("amount").alias("partial_amount_count"),
38            # OJO: count y sum SI se pueden partir y sumar. Los distintos NO.
39            # approx_count_distinct por bucket multiplica el resultado por N buckets.
40            # distinct_merchants_1h se calcula aparte, sin salt, en su propia query.
41            collect_set("merchant_id").alias("partial_merchants"),
42        )
43    
44    # Paso 3: Segunda agregación sin salt (combina los N parciales)
45    final_agg = partial_agg \
46        .groupBy(col("account_id"), col("window")) \
47        .agg(
48            spark_sum("partial_txn_count").alias("txn_count_1h"),
49            (spark_sum("partial_amount_sum") / spark_sum("partial_amount_count")).alias("avg_amount_1h"),
50            size(array_distinct(flatten(collect_list("partial_merchants")))).alias("distinct_merchants_1h"),
51        )
52    
53    return final_agg
54

55# Resultado:
56# Sin salting: 1 executor recibe 2.8M txns -> OOM garantizado
57# Con salting: 16 executors reciben ~175K cada uno -> ~83 MB por executor
58print(f"Sin salting: 1 executor con 2.8M txns (14.2GB -> OOM)")
59print(f"Con salting ({SALT_BUCKETS} buckets):")
60print(f"  Cada executor: ~{2_800_000 // SALT_BUCKETS:,} txns")
61print(f"  Memoria estimada: ~{2_800_000 // SALT_BUCKETS * 500 / 1024 / 1024:.1f} MB por executor")
62print(f"  Overhead: 2x shuffles (acceptable trade-off vs OOM)")

Salting: distribuir merchants con skew en 16 sub-particiones

Data skew es el asesino silencioso de jobs de Spark. Puede estar latente meses (todo funciona con datos uniformes) y explotar cuando un solo merchant/cliente genera volumen anómalo. SIEMPRE diseña para skew desde el día 1. El coste de salting (2x shuffles) es trivial comparado con el coste de un incidente a las 3AM.

En un incidente real, la prioridad SIEMPRE es restaurar el servicio primero (hotfix) y después investigar la causa raíz (solución permanente). Un hotfix feo que funciona en 5 minutos es infinitamente mejor que una solución elegante que tarda 2 horas. Paula te lo dijo al oído: "Deploy first, refactor later."

Regístrate para guardar tu progreso.