Día 3 (tarde) -- Late data y event time vs processing time

### Miércoles, 14:30 -- El café con Carlos

Después de comer, Carlos te busca para el café que te prometió esta mañana. "Ok, el tema de los duplicados ya lo tienes cubierto. Pero hay otro problema más sutil que necesitas entender: cuando un repartidor entra en un túnel o parking subterráneo, pierde señal GPS. Puede estar 30 segundos, 1 minuto, incluso 3 minutos sin enviar nada. Cuando sale y recupera señal, la app envía TODOS los puntos GPS acumulados de golpe -- pero con sus timestamps ORIGINALES, no con el timestamp de cuando los recibe el servidor."

Esto significa que puedes recibir a las 14:25:00 un lote de 6 eventos con timestamps de 14:24:30, 14:24:35, 14:24:40, 14:24:45, 14:24:50, 14:24:55. Son eventos ÚNICOS (no duplicados) pero llegan 30 segundos después de su event time real. Si tu pipeline usa processing time (cuando llegan) en vez de event time (cuando ocurrieron), calcularías el ETA basándote en la posición de hace 30 segundos como si fuera la actual.

### Event time vs Processing time -- la distincion crítica

Esta es una de las distinciones más importantes en streaming:

• ▹Event time: el timestamp del EVENTO (cuando el GPS midió la posición). Es inmutable e inherente al dato.
• ▹Processing time: cuando el sistema de procesamiento (Spark) recibe el evento. Depende de latencia de red, túneles, retries.
• ▹Ingestion time: cuando Kafka recibió el evento. Intermedio entre event y processing.
• ▹En un mundo perfecto: event_time == processing_time. En la realidad: processing_time = event_time + latencia variable.

Para el cálculo de ETA, SIEMPRE quieres usar event time. El ETA se basa en "donde estaba el repartidor en el momento X" -- no en "cuando nos enteramos de donde estaba". Si un repartidor estaba en la posición A a las 14:24:55 y ahora (14:25:05) ya ha avanzado, usar la posición A como "actual" genera un ETA incorrecto (sobreestimado porque la distancia real ya es menor).

### Como Spark maneja late data con watermarks

El watermark de 30 segundos que configuraste esta mañana también resuelve este problema de late arrivals. El mecanismo es: Spark rastrea el event_time máximo visto (ej: 14:25:00). El watermark es max_event_time - 30s = 14:24:30. Cualquier evento con event_time >= 14:24:30 se acepta y procesa normalmente. Los eventos del túnel (14:24:00 a 14:24:30) tienen event_time entre 14:24:00 y 14:24:30 -- están JUSTO en el límite del watermark.

1# El watermark protege contra late data de forma automática
2# Ejemplo con los datos del tunel:
3
4# Evento más reciente visto: event_time = 14:25:00
5# Watermark actual: 14:25:00 - 30s = 14:24:30
6# 
7# Evento que llega tarde con event_time = 14:24:35 -> ACEPTADO (> watermark)
8# Evento que llega tarde con event_time = 14:24:25 -> DESCARTADO (< watermark)
9# Evento que llega tarde con event_time = 14:24:30 -> ACEPTADO (= watermark)
10
11# En nuestro escenario del tunel:
12# - Eventos con event_time 14:24:30+ -> se procesan correctamente
13# - Eventos con event_time 14:24:00-14:24:29 -> se descartan (demasiado tarde)
14# - Conclusion: con watermark=30s, perdemos ~30s de datos del inicio del tunel
15#   pero mantenemos la posición MAS RECIENTE post-tunel
16
17# Para el cálculo de ETA, perder los primeros puntos del tunel es ACEPTABLE
18# porque lo que importa es la posición MAS RECIENTE, no el histórico completo

### El tradeoff del watermark: latencia vs completitud

Elegir el valor del watermark es un tradeoff entre dos objetivos opuestos:

• ▹Watermark BAJO (10s): cierra ventanas rápido, baja latencia, pero pierde muchos datos de túneles/parking
• ▹Watermark ALTO (60s): acepta casi todos los late arrivals, pero añade 60s de latencia al cierre de ventanas
• ▹FlashRoute: 30s es un buen balance -- captura 99% de eventos y añade 30s de "espera" antes de cerrar
• ▹Si Ana pidiera latencia <10s en vez de <30s, tendrías que bajar el watermark y aceptar perder más datos

### Patrón: tomar la posición más reciente por repartidor

Para el ETA, solo te importa la posición MÁS RECIENTE de cada repartidor -- no todas las posiciones históricas. Si recibes un burst de 6 eventos de un túnel (t=00, 05, 10, 15, 20, 25), solo necesitas el de t=25 (el más reciente en event_time). El resto es histórico útil para S3 pero no para el cálculo en vivo.

1from pyspark.sql.functions import col, row_number, desc
2from pyspark.sql.window import Window
3
4def get_latest_positions(batch_df, batch_id):
5    """
6    En cada micro-batch, quedarse solo con la posición más reciente
7    de cada repartidor basandose en event_time (no processing_time).
8    
9    Esto es crucial para late data: si llegan 6 eventos del tunel,
10    solo usamos el más reciente para calcular el ETA actual.
11    """
12    if batch_df.isEmpty():
13        return batch_df
14    
15    # Window por driver, ordenado por event_time descendente
16    w = Window.partitionBy("driver_id").orderBy(desc("event_time"))
17    
18    latest = batch_df \
19        .withColumn("rn", row_number().over(w)) \
20        .filter(col("rn") == 1) \
21        .drop("rn")
22    
23    return latest
24
25
26# Uso en el pipeline completo:
27# 1. Leer de Kafka
28# 2. Parsear + watermark + dedup + filtros (del pipeline de la mañana)
29# 3. foreachBatch -> get_latest_positions -> calcular ETA -> escribir DynamoDB
30
31query = clean_gps_stream.writeStream \
32    .foreachBatch(lambda df, bid: process_etas(get_latest_positions(df, bid))) \
33    .trigger(processingTime="30 seconds") \
34    .start()

### El pipeline actualizado: mañana + tarde integrados

Al final del día 3, tu pipeline de GPS tiene 4 capas de proteccion contra datos problematicos. Cada capa resuelve un problema diferente y el orden importa:

1. 01.Watermark (30s): define la ventana temporal de aceptación. Eventos más antiguos se descartan.
2. 02.Deduplicación (dropDuplicates): elimina reintentos de la app. Misma clave = descarta.
3. 03.Filtros de sanidad: descarta datos físicamente imposibles (coord 0,0, velocidad 200km/h).
4. 04.Latest position (row_number): de múltiples posiciones por repartidor, toma la más reciente.

Le envías a Carlos un diagrama del pipeline actualizado por Slack. Carlos responde: "Perfecto. Esto cubre el 99% de los problemas que he visto en 2 años. El 1% restante son GPSs completamente defectuosos que envían basura continuamente -- eso lo cubres con un circuit breaker, pero es tarea para otro día." Lo anotas para el día 5.