Bonus avanzado - Monitoring de Kafka y health de eventos

1Marcos Delgado  (call, lunes semana 3)
2
3"Las alertas que montaste la semana pasada son buenas para
4detectar problemas en los datos ya cargados. Pero hay un
5gap: si el Kafka Connect se cae y los eventos dejan de
6llegar a S3, tu alerta de frescura no se dispara hasta
72 horas después. Son 2 horas de datos perdidos.
8
9Quiero que montes un health dashboard con 3 capas:
10
111. Kafka consumer lag: cuantos mensajes hay en el topic
12   que el consumer todavía no ha procesado. Si sube, el
13   consumer está atrasado.
14
152. S3 landing check: verificar que cada hora se crea un
16   archivo nuevo en S3. Si no hay archivo en la última
17   hora, el Connect está roto.
18
193. Pipeline throughput: eventos por minuto que tu pipeline
20   procesa. Si baja a 0, algo está mal en TU código.
21
22No necesitas montarlo en Grafana todavía. Hazlo con Python
23y que escriba métricas a un archivo o las imprima. La
24infraestructura de dashboarding la montamos después."

1# kafka_health.py - Monitor de consumer lag (simulado)
2# En producción usarias kafka-python o confluent_kafka
3# Aqui simulamos leyendo las métricas de un archivo de estado
4
5import json
6from datetime import datetime
7from typing import Dict, List
8
9def get_consumer_lag(state_file: str = 'kafka_state.json') -> Dict:
10    """
11    Lee el estado del consumer group y calcula el lag.
12    En producción: kafka.admin.AdminClient.list_consumer_group_offsets()
13    """
14    with open(state_file) as f:
15        state = json.load(f)
16    
17    total_lag = 0
18    partition_lags = []
19    
20    for partition in state['partitions']:
21        lag = partition['latest_offset'] - partition['consumer_offset']
22        total_lag += lag
23        partition_lags.append({
24            'partition': partition['id'],
25            'lag': lag,
26            'latest_offset': partition['latest_offset'],
27            'consumer_offset': partition['consumer_offset'],
28        })
29    
30    # Clasificar severidad
31    if total_lag > 100000:
32        status = 'CRITICAL'
33    elif total_lag > 10000:
34        status = 'WARNING'
35    else:
36        status = 'OK'
37    
38    return {
39        'check': 'kafka_consumer_lag',
40        'status': status,
41        'total_lag': total_lag,
42        'partitions': partition_lags,
43        'message': f'Lag total: {total_lag:,} mensajes ({status})',
44        'timestamp': datetime.utcnow().isoformat(),
45    }
46
47def check_s3_landing(s3_path: str, max_age_minutes: int = 60) -> Dict:
48    """
49    Verifica que hay archivos recientes en S3.
50    En producción: boto3 list_objects con filtro por fecha.
51    Aqui simulamos con el filesystem local.
52    """
53    from pathlib import Path
54    import os
55    
56    path = Path(s3_path)
57    if not path.exists():
58        return {
59            'check': 's3_landing',
60            'status': 'CRITICAL',
61            'message': f'Path no existe: {s3_path}',
62        }
63    
64    # Buscar el archivo más reciente
65    files = sorted(path.glob('**/*.jsonl'), key=os.path.getmtime, reverse=True)
66    
67    if not files:
68        return {
69            'check': 's3_landing',
70            'status': 'CRITICAL',
71            'message': 'No hay archivos JSONL en el path',
72        }
73    
74    latest_file = files[0]
75    age_minutes = (datetime.now() - datetime.fromtimestamp(
76        os.path.getmtime(latest_file)
77    )).total_seconds() / 60
78    
79    if age_minutes > max_age_minutes * 2:
80        status = 'CRITICAL'
81    elif age_minutes > max_age_minutes:
82        status = 'WARNING'
83    else:
84        status = 'OK'
85    
86    return {
87        'check': 's3_landing',
88        'status': status,
89        'latest_file': str(latest_file.name),
90        'age_minutes': round(age_minutes),
91        'message': f'Ultimo archivo: {latest_file.name} ({age_minutes:.0f} min ago)',
92    }

Monitoring de Kafka y S3: detectar problemas antes de que lleguen a Redshift

1# pipeline_throughput.py - Medir cuantos eventos procesa tu pipeline
2import pandas as pd
3from datetime import datetime, timedelta
4from pathlib import Path
5
6def check_pipeline_throughput(
7    output_path: str = 'output/staging/',
8    window_minutes: int = 60,
9) -> Dict:
10    """
11    Verifica cuantos eventos ha procesado el pipeline recientemente.
12    Lee el campo ingested_at del output para calcularlo.
13    """
14    path = Path(output_path)
15    parquet_files = list(path.glob('*.parquet'))
16    
17    if not parquet_files:
18        return {
19            'check': 'pipeline_throughput',
20            'status': 'CRITICAL',
21            'message': 'No hay archivos de output',
22            'events_per_minute': 0,
23        }
24    
25    # Leer el más reciente
26    df = pd.read_parquet(parquet_files[-1])
27    df['ingested_at'] = pd.to_datetime(df['ingested_at'])
28    
29    now = datetime.utcnow()
30    recent = df[df['ingested_at'] >= now - timedelta(minutes=window_minutes)]
31    
32    events_per_minute = len(recent) / window_minutes if window_minutes > 0 else 0
33    
34    # Umbrales (basados en 2.5M eventos/día = ~1736/min)
35    if events_per_minute == 0:
36        status = 'CRITICAL'
37    elif events_per_minute < 500:
38        status = 'WARNING'
39    else:
40        status = 'OK'
41    
42    return {
43        'check': 'pipeline_throughput',
44        'status': status,
45        'events_per_minute': round(events_per_minute),
46        'events_last_window': len(recent),
47        'message': f'{events_per_minute:.0f} eventos/min (esperado: ~1700/min)',
48    }
49
50# --- Dashboard completo ---
51def run_health_dashboard():
52    """Ejecuta todos los checks de health del pipeline."""
53    print("=" * 60)
54    print(f"PIPELINE HEALTH DASHBOARD - {datetime.utcnow().isoformat()}")
55    print("=" * 60)
56    
57    checks = [
58        get_consumer_lag(),
59        check_s3_landing('datos/taskflow_events/'),
60        check_pipeline_throughput(),
61    ]
62    
63    for check in checks:
64        icon = {'OK': '?', 'WARNING': '??', 'CRITICAL': '??'}[check['status']]
65        print(f"\n{icon} {check['check']}")
66        print(f"   Status: {check['status']}")
67        print(f"   {check['message']}")
68    
69    critical = [c for c in checks if c['status'] == 'CRITICAL']
70    if critical:
71        print(f"\n{'='*60}")
72        print(f"?? {len(critical)} CHECKS CRITICOS - INVESTIGAR")
73    else:
74        print(f"\n{'='*60}")
75        print("? Todos los sistemas operando normalmente")
76
77run_health_dashboard()

Health dashboard completo: Kafka + S3 + Pipeline en un solo reporte

1# tracking_health.py - Detectar tipos de evento que dejan de emitirse
2import pandas as pd
3from datetime import datetime, timedelta
4
5def check_event_type_health(
6    df: pd.DataFrame,
7    max_silence_hours: int = 4,
8) -> list:
9    """
10    Para cada tipo de evento, verifica cuando fue el último que se recibió.
11    Si un tipo que normalmente emite cada hora lleva más de max_silence_hours
12    sin emitir, es CRITICAL.
13    """
14    now = datetime.utcnow()
15    
16    # Ultimo evento por tipo
17    last_by_type = df.groupby('event_type')['event_timestamp'].max().reset_index()
18    last_by_type.columns = ['event_type', 'last_seen']
19    last_by_type['hours_since_last'] = (
20        (now - last_by_type['last_seen']).dt.total_seconds() / 3600
21    )
22    
23    # Tipos esperados (que deberian emitir al menos cada hora)
24    expected_frequent = [
25        'board_viewed', 'task_created', 'task_completed',
26        'comment_added', 'user_login', 'task_assigned',
27    ]
28    
29    alerts = []
30    for _, row in last_by_type.iterrows():
31        if row['event_type'] in expected_frequent:
32            if row['hours_since_last'] > max_silence_hours:
33                alerts.append({
34                    'event_type': row['event_type'],
35                    'hours_silent': round(row['hours_since_last'], 1),
36                    'status': 'CRITICAL',
37                    'action': f'Verificar con David si el tracking de {row["event_type"]} sigue activo',
38                })
39            elif row['hours_since_last'] > max_silence_hours / 2:
40                alerts.append({
41                    'event_type': row['event_type'],
42                    'hours_silent': round(row['hours_since_last'], 1),
43                    'status': 'WARNING',
44                    'action': 'Monitorizar - podria ser horario de baja actividad',
45                })
46    
47    return alerts
48
49# Uso
50events = pd.read_parquet('datos/stg_product_events.parquet')
51tracking_alerts = check_event_type_health(events)
52
53if tracking_alerts:
54    print("?? Alertas de tracking:")
55    for a in tracking_alerts:
56        print(f"  [{a['status']}] {a['event_type']}: {a['hours_silent']}h sin eventos")
57        print(f"    Accion: {a['action']}")
58else:
59    print("? Todos los tipos de evento emitiendo normalmente")

Deteccion de tracking roto: nunca más 3 días sin darnos cuenta

1Marcos Delgado  (viernes semana 3, después de la demo)
2
3"Has montado en 3 semanas lo que yo esperaba que tardases
4un mes y medio. El pipeline, las cohortes, las alertas,
5el health dashboard. Y todo clean, testeado, documentado.
6
7Te voy a ser honesto: cuando Pablo dijo que contrataba a
8alguien de FreshMart con 6 meses de experiencia, pense
9'va a necesitar mucha mano'. Me equivoque.
10
11La diferencia es que no solo escribes código -- entiendes
12el PROBLEMA. Cuando encontraste Simpson no fuiste y dijiste
13'mira que anomalia técnica'. Dijiste 'esto cambia la
14narrativa de negocio'. Eso es lo que necesitamos.
15
16El proximo trimestre quiero que lideres el proyecto de
17consumer lag monitoring con Kafka. Es más infra, menos
18analítica. Te va a sacar de tu zona de comfort. Te interesa?"
19
20Tu: "Me interesa mucho. Es justo el gap que quiero cerrar:
21entender la infra debajo del pipeline, no solo los datos
22que salen."
23
24Marcos: "Bien. El lunes hablamos de scope."