Día 4 (3AM) -- El incidente: transacciones duplicadas

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

1# Diagnóstico ejecutado a las 3:30 AM (query sobre S3 del último batch exitoso)
2import pandas as pd
3
4# Distribución del batch problemático
5batch_stats = pd.DataFrame({
6    "merchant_id": ["MERCH-48291", "MERCH-10001", "MERCH-22340", 
7                    "MERCH-15678", "MERCH-09923", "OTROS (4200+)"],
8    "txn_count": [2_800_000, 10_200, 8_750, 7_100, 12_400, 45_000],
9    "avg_amount_eur": [0.01, 45.30, 128.50, 22.10, 67.80, 52.40],
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 batch: {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"Promedio otros merchants: {normal_avg:,.0f} txns/hora")
23print(f"Skew ratio: {skew_ratio:.0f}x")
24print()
25print("CAUSA RAÍZ IDENTIFICADA:")
26print("  Bug en TPV de MERCH-48291 genera bucle de retry infinito.")
27print("  Cada retry crea una transacción de 0.01 EUR.")
28print("  El job particiona por merchant_id en el groupBy.")
29print("  -> 2.8M eventos van al MISMO executor -> OOM.")
30print()
31print(f"IMPACTO:")
32print(f"  - Features de fraude stale: {47} minutos (SLA: 5 min)")
33print(f"  - Transacciones en cola: ~{200_000:,} (Kafka retiene 7 días)")
34print(f"  - Transacciones perdidas: 0 (Kafka retention = safety net)")
35print(f"  - Impacto en clientes: NINGUNO (modelo usó features cached)")

Diagnóstico: MERCH-48291 con skew de 280x causó OOM en cascada

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
8main_stream = deduplicated.filter(
9    ~col("merchant_id").isin(BLOCKED_MERCHANTS)
10)
11
12# Cuarentena: capturar merchants bloqueados (NO perder datos)
13quarantine_stream = deduplicated.filter(
14    col("merchant_id").isin(BLOCKED_MERCHANTS)
15)
16
17# Escribir cuarentena a ruta separada para análisis posterior
18quarantine_query = quarantine_stream.writeStream \
19    .format("parquet") \
20    .option("path", "s3://neobank-lake/quarantine/blocked_merchants/") \
21    .option("checkpointLocation", "s3://neobank-checkpoints/quarantine/") \
22    .trigger(processingTime="30 seconds") \
23    .outputMode("append") \
24    .start()
25
26# Timeline del hotfix:
27# 03:45 - Decisión tomada
28# 03:47 - Código escrito (5 líneas de cambio)
29# 03:49 - PR abierto con descripción del incidente
30# 03:52 - Paula aprueba (code review de emergencia)
31# 03:54 - Deploy via GitHub Actions -> EMR
32# 03:55 - Job reinicia desde último checkpoint
33# 03:57 - Consumer lag empieza a bajar: 200K -> 150K -> 100K
34# 03:59 - Lag en 0. Servicio completamente restaurado.
35# Total downtime: 47 minutos.
36
37print("HOTFIX DESPLEGADO - Servicio restaurado a las 03:59")
38print(f"  Merchants bloqueados: {BLOCKED_MERCHANTS}")
39print(f"  Datos en cuarentena: s3://neobank-lake/quarantine/")
40print(f"  Consumer lag: 0 (restaurado)")

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

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            approx_count_distinct("merchant_id").alias("partial_merchants"),
39        )
40    
41    # Paso 3: Segunda agregación sin salt (combina los N parciales)
42    final_agg = partial_agg \
43        .groupBy(col("account_id"), col("window")) \
44        .agg(
45            spark_sum("partial_txn_count").alias("txn_count_1h"),
46            (spark_sum("partial_amount_sum") / spark_sum("partial_amount_count")).alias("avg_amount_1h"),
47            spark_sum("partial_merchants").alias("distinct_merchants_1h_approx"),
48        )
49    
50    return final_agg
51
52# Resultado:
53# Sin salting: 1 executor recibe 2.8M txns -> OOM garantizado
54# Con salting: 16 executors reciben ~175K cada uno -> 1.1GB por executor
55print(f"Sin salting: 1 executor con 2.8M txns (14.2GB -> OOM)")
56print(f"Con salting ({SALT_BUCKETS} buckets):")
57print(f"  Cada executor: ~{2_800_000 // SALT_BUCKETS:,} txns")
58print(f"  Memoria estimada: ~{2_800_000 // SALT_BUCKETS * 500 / 1024 / 1024:.1f} MB por executor")
59print(f"  Overhead: 2x shuffles (acceptable trade-off vs OOM)")

Salting: distribuir merchants con skew en 16 sub-particiones