Día 4 -- BLACK FRIDAY: el pipeline no escala

1# Script de diagnóstico: consumer lag por particion
2# Ejecutas esto desde tu portatil a las 7:30 AM
3
4def diagnose_kafka_lag():
5    """Simula la salida de kafka-consumer-groups.sh --describe"""
6    partitions = [
7        {"partition": 0,  "current": 1_847_293, "end": 1_852_104, "lag": 4811},
8        {"partition": 1,  "current": 1_845_011, "end": 1_851_892, "lag": 6881},
9        {"partition": 2,  "current": 1_849_500, "end": 1_852_003, "lag": 2503},
10        {"partition": 3,  "current": 1_844_200, "end": 1_851_750, "lag": 7550},
11        {"partition": 4,  "current": 1_847_800, "end": 1_852_200, "lag": 4400},
12        {"partition": 5,  "current": 1_843_100, "end": 1_851_600, "lag": 8500},
13        {"partition": 6,  "current": 1_848_900, "end": 1_852_100, "lag": 3200},
14        {"partition": 7,  "current": 1_846_500, "end": 1_851_900, "lag": 5400},
15        {"partition": 8,  "current": 1_842_800, "end": 1_851_500, "lag": 8700},
16        {"partition": 9,  "current": 1_847_100, "end": 1_852_000, "lag": 4900},
17        {"partition": 10, "current": 1_844_900, "end": 1_851_800, "lag": 6900},
18        {"partition": 11, "current": 1_848_200, "end": 1_852_300, "lag": 4100},
19    ]
20    
21    total_lag = sum(p["lag"] for p in partitions)
22    max_lag = max(p["lag"] for p in partitions)
23    min_lag = min(p["lag"] for p in partitions)
24    
25    print(f"=== DIAGNOSTICO CONSUMER LAG ===")
26    print(f"Total lag: {total_lag:,} mensajes")
27    print(f"Lag promedio/particion: {total_lag // 12:,}")
28    print(f"Peor particion: p{max(partitions, key=lambda p: p['lag'])['partition']} ({max_lag:,})")
29    print(f"Mejor particion: p{min(partitions, key=lambda p: p['lag'])['partition']} ({min_lag:,})")
30    print(f"Skew ratio: {max_lag / min_lag:.1f}x")
31    print(f"")
32    print(f"A 600 ev/s, retraso estimado: {total_lag / 600:.0f}s ({total_lag / 600 / 60:.1f} min)")
33    print(f"")
34    print(f"DIAGNOSTICO: throughput insuficiente + skew moderado")
35    print(f"ACCION: escalar executors de Spark (rápido) + evaluar más particiones (medio plazo)")
36
37diagnose_kafka_lag()

Diagnóstico a las 7:30 AM: identificar total lag, skew y retraso estimado

1# CONFIGURACION NORMAL (200 ev/s)
2spark_config_normal = {
3    "spark.executor.instances": "4",
4    "spark.executor.cores": "2",
5    "spark.executor.memory": "4g",
6    "spark.sql.shuffle.partitions": "12",
7    "maxOffsetsPerTrigger": "10000",
8}
9
10# CONFIGURACION BLACK FRIDAY (600+ ev/s)
11spark_config_peak = {
12    "spark.executor.instances": "8",         # 4 -> 8 (doble paralelismo)
13    "spark.executor.cores": "3",             # 2 -> 3 (más tasks por executor)
14    "spark.executor.memory": "6g",           # 4g -> 6g (más estado en memoria)
15    "spark.sql.shuffle.partitions": "24",    # 12 -> 24 (más paralelismo shuffle)
16    "spark.streaming.backpressure.enabled": "true",  # Auto-ajuste velocidad
17    "maxOffsetsPerTrigger": "50000",         # 10K -> 50K (drenar lag rápido)
18}
19
20# Para aplicar en EMR sin downtime:
21# 1. Escalar cluster: aws emr modify-instance-groups --cluster-id j-XXX
22# 2. Reiniciar streaming query (Spark reanuda desde checkpoint)
23# El checkpoint garantiza exactly-once: no pierde ni duplica eventos

Configuración normal vs Black Friday: duplicar recursos para 3x volumen

1from concurrent.futures import ThreadPoolExecutor
2import boto3
3from datetime import datetime, timedelta
4
5def write_etas_dynamodb_parallel(batch_df, batch_id):
6    """
7    Version optimizada para Black Friday:
8    - Divide el batch en chunks de 25 (máximo DynamoDB batch_writer)
9    - Escribe en paralelo con ThreadPoolExecutor (8 threads)
10    - Reduce tiempo de escritura de 14s a ~3s
11    """
12    if batch_df.isEmpty():
13        return
14    
15    rows = batch_df.collect()
16    chunks = [rows[i:i+25] for i in range(0, len(rows), 25)]
17    
18    def write_chunk(chunk):
19        dynamo = boto3.resource('dynamodb', region_name='eu-west-1')
20        table = dynamo.Table('delivery_etas')
21        with table.batch_writer() as writer:
22            for row in chunk:
23                writer.put_item(Item={
24                    'delivery_id': row['delivery_id'],
25                    'driver_id': row['driver_id'],
26                    'eta_minutes': int(row['eta_minutes']),
27                    'distance_km': str(round(float(row['distance_km']), 2)),
28                    'is_at_risk': bool(row['is_at_risk']),
29                    'zone': row['zone'],
30                    'calculated_at': datetime.now().isoformat(),
31                    'ttl': int((datetime.now() + timedelta(minutes=30)).timestamp())
32                })
33    
34    # 8 threads escriben en paralelo -> 8x speedup teorico
35    with ThreadPoolExecutor(max_workers=8) as pool:
36        pool.map(write_chunk, chunks)
37
38# Resultado: foreachBatch pasa de 14s a 3s
39# Total micro-batch: ~8s (lectura 2s + proceso 3s + escritura 3s)
40# Con trigger=30s, hay 22s de margen -> estable

Sink DynamoDB paralelo: 8 threads concurrentes reducen 14s a 3s