Todo lo calculado hasta ahora en el módulo partía de un conjunto cerrado: el fichero del 14 de septiembre, los clics de siete días. Pero los datos de Kilómetro Cero no nacen cerrados. pedidos.eventos recibe cientos de eventos stock.actualizado por segundo durante una campaña, los 140 repartidores (como furgoneta-3) envían 2,4 millones de posiciones al día, y hay preguntas que no pueden esperar al lote nocturno: qué productos se están quedando sin stock en Girona ahora, dónde está cada repartidor ahora, qué repartidor lleva tres minutos sin dar señal. El procesamiento de flujos responde a esas preguntas ejecutando el cálculo de forma continua, evento a evento, sobre un conjunto que nunca termina. Eso rompe suposiciones que en los lotes eran gratuitas: no se puede "esperar a tener todo" antes de agrupar, los eventos llegan desordenados y con retraso respecto al momento en que ocurrieron, el estado del cálculo debe sobrevivir a los fallos sin poder reejecutar "desde el principio", y la entrada puede llegar más rápido de lo que se procesa. Esta lección da nombre a cada uno de esos problemas (tiempo de evento, marcas de agua, ventanas, estado y checkpoints, exactly-once, backpressure), compara las arquitecturas Lambda y Kappa y las tres herramientas habituales (Kafka Streams, Flink, Spark Structured Streaming), y resuelve los dos casos de analitica en tiempo real: la alerta de stock bajo y el panel de reparto. La entrega de esos resultados al navegador del operador queda para 08-02; aquí terminamos en un tópico de Kafka.

1. De los lotes a los flujos: qué cambia
2. Tiempo de evento, tiempo de procesamiento y marcas de agua
3. Ventanas: tumbling, sliding y session
4. Estado y checkpoints
5. Exactly-once en streaming
6. Backpressure
7. Arquitecturas Lambda y Kappa
8. Herramientas: Kafka Streams, Flink y Spark Structured Streaming
9. Los casos de Kilómetro Cero: stock bajo y panel de reparto
10. Práctica: simulación en Python, PyFlink y Structured Streaming
11. Errores Comunes y Consejos
12. Ejercicios
13. Conclusión

Recuperamos la tabla de 05-01 y la afinamos con lo aprendido desde entonces:

Un motor de flujos ejecuta el mismo tipo de DAG de operadores que Spark (05-03), pero desplegado de forma permanente: cada operador es un proceso (o varios, en paralelo por clave) que espera eventos, los procesa y emite hacia el siguiente, sin que el DAG termine nunca. La fuente típica es Kafka (02-04), que aporta lo que un flujo necesita: particiones (paralelismo), offsets (posición reanudable) y retención (releer desde atrás). El sink es otro tópico, una base de datos o un panel.

Hay dos formas de mover eventos por el DAG: evento a evento (Flink, Kafka Streams: latencia de milisegundos) y por microlotes (Spark Structured Streaming: cada pocos cientos de milisegundos o segundos se toma lo que ha llegado y se ejecuta como un pequeño lote). La segunda reutiliza toda la maquinaria de lotes y simplifica exactly-once; la primera da latencias menores. Para el panel de reparto y la alerta de stock, segundos son suficientes, así que ambas sirven.

En 01-05 distinguimos el momento en que algo ocurrió del momento en que otro nodo se entera. En flujos esa distinción es la que más consecuencias tiene:

• Tiempo de evento (event time): cuándo ocurrió el hecho, según el reloj de quien lo generó. Es el fecha_ms de la envoltura de 02-05. La posición de furgoneta-3 tomada a las 10:04:58 tiene ese tiempo aunque el móvil la envíe a las 10:07 por falta de cobertura.
• Tiempo de procesamiento (processing time): cuándo el operador ve el evento, según su reloj. Depende de la cola en Kafka, de la red, de la carga.

La pregunta "¿cuántos stock.actualizado de queso-curado hubo entre las 10:00 y las 10:05?" solo tiene una respuesta correcta en tiempo de evento. Con tiempo de procesamiento, un reinicio del consumidor a las 10:03 metería en la ventana 10:05–10:10 eventos ocurridos a las 10:02; el resultado dependería del comportamiento del sistema, no de los hechos. Pero calcular en tiempo de evento crea un problema nuevo: cuando llega un evento con fecha_ms = 10:04:58 a las 10:07, ¿la ventana 10:00–10:05 ya se había emitido? ¿Hay que reabrirla? ¿Cuánto esperar antes de darla por cerrada?

La respuesta es la marca de agua (watermark): una afirmación que el sistema hace fluir por el DAG que dice "no espero más eventos con tiempo de evento anterior a T". Se genera en la fuente con una heurística, normalmente máximo tiempo de evento visto − retraso tolerado: con un retraso tolerado de 2 minutos, tras ver un evento de las 10:07:00 la marca de agua es 10:05:00, y en ese momento la ventana 10:00–10:05 se considera completa y se emite. El retraso tolerado es una decisión de negocio y de medición: cuánto tardan de verdad los eventos en llegar (el percentil 99 del desfase llegada − fecha_ms), contra cuánta latencia se acepta en el resultado. Poco retraso tolerado: resultados rápidos que descartan más eventos; mucho: resultados más completos pero tardíos.

Los eventos que llegan después de que la marca de agua haya pasado su ventana son eventos tardíos (late events). Cada motor ofrece tres destinos: descartarlos (el comportamiento por defecto), incorporarlos reemitiendo la ventana actualizada durante un margen adicional (allowed lateness en Flink, que obliga a mantener la ventana en el estado más tiempo), o desviarlos a una salida lateral (side output) para contarlos, registrarlos o reprocesarlos por lotes. Contar los tardíos es obligatorio: es la métrica que dice si el retraso tolerado está bien elegido.

Con varias particiones de Kafka, cada partición tiene su propia marca de agua y la del operador es el mínimo de todas: una partición sin tráfico (un mercado cerrado por la noche) retiene la marca de agua global y bloquea la emisión de ventanas de todos. Los motores lo tratan con idle timeouts que excluyen las particiones inactivas del mínimo.

Sobre un flujo infinito, cualquier agregación ("cuántos", "el mínimo") necesita un límite: la ventana. Las tres formas básicas:

Las ventanas se combinan con una clave: "tumbling de 5 minutos por (producto, mercado)" mantiene una ventana por cada pareja, y el paralelismo del operador es por clave (todas las queso-curado/girona van a la misma instancia, como en el shuffle de 05-02). Y con el tiempo de evento: es el fecha_ms el que decide en qué ventana cae un evento, y la marca de agua la que decide cuándo se emite.

Una ventana abierta, un contador por clave, la última posición conocida de cada repartidor: todo eso es estado, y en un flujo vive indefinidamente en el operador. Los motores lo guardan en un state backend local al operador (memoria, o RocksDB en disco local para estados de gigabytes) particionado por clave, exactamente como un actor por clave (05-01). El problema es la durabilidad: si el nodo muere, su estado local se pierde, y no se puede "reejecutar desde el principio" porque el principio fue hace meses.

La solución es el checkpoint: periódicamente (cada 10 s, cada minuto) el motor escribe una copia consistente del estado de todos los operadores, junto con los offsets de Kafka que ese estado refleja, en almacenamiento duradero (HDFS, MinIO). Ante un fallo, restaura el último checkpoint en nodos sanos y reanuda el consumo desde esos offsets: los eventos posteriores al checkpoint se vuelven a procesar, y el estado acaba siendo el mismo que si no hubiera habido fallo.

La palabra difícil es consistente: el estado de todos los operadores debe corresponder al mismo punto del flujo, aunque cada uno vaya por un evento distinto. Flink lo consigue con el algoritmo de Chandy-Lamport adaptado (las barreras de checkpoint): la fuente inyecta en el flujo un marcador con el número de checkpoint; cada operador, al recibirlo por todas sus entradas, guarda su estado y reenvía el marcador; el estado guardado refleja exactamente los eventos anteriores al marcador. Es una instantánea distribuida sin detener el flujo. Spark Structured Streaming lo tiene más fácil: cada microlote es una unidad, y el checkpoint registra qué microlotes se han completado con sus rangos de offsets. Kafka Streams guarda el estado en tópicos de Kafka (changelog topics) y lo reconstruye releyéndolos.

Un savepoint es un checkpoint disparado a mano, con formato estable, que se usa para parar el trabajo, cambiar el código o el paralelismo, y reanudar desde el mismo estado: el equivalente de un despliegue sin perder "los últimos cinco minutos". Y el tamaño del estado importa: una session window por repartidor es pequeña, pero "todos los pedidos de las últimas 24 horas por cliente" son gigabytes que hay que checkpointear cada minuto; los checkpoints incrementales (solo lo cambiado desde el anterior) y un TTL para el estado que no se toca son las herramientas.

En 02-05 concluimos que "exactly-once" en la entrega no existe, y que lo alcanzable es at-least-once más consumidores idempotentes. En streaming el término se usa con un significado preciso y alcanzable: el estado del motor refleja cada evento exactamente una vez, aunque haya fallos y reprocesamientos. Se consigue con el checkpoint del apartado anterior: tras un fallo, el estado vuelve al del checkpoint y los eventos posteriores se reaplican; como el estado que reflejaba esos eventos se ha descartado, no hay doble conteo.

Lo que el checkpoint no cubre es lo que ya salió del motor: la alerta escrita en el tópico inventario.alertas, la fila insertada en PostgreSQL, antes del fallo y después del último checkpoint. Al reprocesar, esos efectos se producen otra vez. Para que el resultado externo sea también exactly-once (end-to-end), el sink tiene que ser una de dos cosas:

• Idempotente. Escribir de nuevo el mismo resultado no cambia nada: UPSERT por clave (ventana, producto, mercado) en PostgreSQL, un PUT en Redis con la misma clave, un producer de Kafka con una clave y un consumidor idempotente aguas abajo (02-05). Es la opción más simple y la que se recomienda siempre que la salida tenga una clave natural.
• Transaccional. El sink escribe en una transacción que solo se confirma cuando el checkpoint se completa (two-phase commit sink): Flink con el producer transaccional de Kafka (DeliveryGuarantee.EXACTLY_ONCE), o con una tabla y una transacción por checkpoint. Entre el pre-commit y el commit los datos existen pero no son visibles para consumidores con isolation.level=read_committed. Añade latencia (el intervalo de checkpoint) y complejidad; se reserva para sinks sin clave natural (un log de eventos en Kafka).

Un DAG de flujos es una cadena de productores y consumidores, y en cualquier momento uno puede ir más despacio que el anterior: el operador de ventanas escribiendo un checkpoint grande, el sink de PostgreSQL saturado, un pico de la Semana del Queso Artesano que triplica los stock.actualizado. Si el operador rápido siguiera enviando, las colas intermedias crecerían hasta agotar la memoria. Backpressure (contrapresión) es el mecanismo por el que la lentitud se propaga hacia atrás: el operador lento deja de aceptar, el anterior llena su búfer de salida y deja de leer de su entrada, y así hasta la fuente, que deja de consumir de Kafka. Los eventos se acumulan en Kafka (que está diseñado para ello, con retención de días), no en la memoria del motor, y el lag del grupo de consumidores (02-04) se convierte en la métrica que indica que el flujo no da abasto.

Flink lo implementa con créditos entre tareas (el receptor anuncia cuánto puede recibir); Kafka Streams lo tiene gratis porque cada instancia hace poll solo cuando ha terminado con el lote anterior; Spark Structured Streaming lo aproxima limitando cuántos offsets lee por microlote (maxOffsetsPerTrigger). Lo que ninguno hace es resolver la causa: si el lag crece de forma sostenida, hay que aumentar el paralelismo (más particiones y más instancias), aligerar el operador o aceptar resultados aproximados. Y ojo con la marca de agua: bajo backpressure, el tiempo de procesamiento se aleja del tiempo de evento, pero las ventanas siguen siendo correctas porque se definen en tiempo de evento; eso es precisamente lo que el apartado 2 compraba.

Cuando el streaming era nuevo y poco fiable, la respuesta a "quiero resultados en tiempo real pero también exactos" fue la arquitectura Lambda (Marz, 2011): mantener dos caminos. La capa batch recalcula cada noche, desde el lago, las vistas completas y exactas; la capa de velocidad calcula con streaming las últimas horas, de forma aproximada; una capa de servicio combina ambas al consultar. Funciona, pero obliga a escribir y mantener la misma lógica dos veces, en dos motores, con dos semánticas, y a reconciliar sus diferencias. La arquitectura Kappa (Kreps, 2014) propone un solo camino: todo es un flujo, con Kafka reteniendo el historial (o un lago de eventos releíble), y el "lote" es simplemente reprocesar el flujo desde un offset antiguo con la misma aplicación de streaming, escribiendo en una tabla nueva y cambiando el puntero cuando alcanza el presente.

La plataforma de datos de Kilómetro Cero acaba siendo pragmáticamente mixta: ventas_diarias.py y ALS son lotes porque necesitan el conjunto completo y no tienen prisa; el stock y el reparto son flujos porque la latencia es el requisito. Lo que Kappa aporta es el criterio: si la misma lógica se escribe dos veces, algo está mal; que los motores modernos ejecuten el mismo código en lote y en flujo (Spark, Flink) hace que la elección sea de despliegue, no de reescritura.

Elegimos Flink como motor de flujos de analitica por la semántica de tiempo de evento y por PyFlink, y mantenemos Structured Streaming como alternativa porque reutiliza la API de 05-03. Kafka Streams queda nombrado: es la opción correcta si el streaming vive dentro de un servicio Java y no en una plataforma de datos.

Caso (a): alerta de stock bajo. inventario publica stock.actualizado en pedidos.eventos con cada cambio (04-05 los usaba para invalidar Redis). Datos del evento: producto, mercado, stock_actual, delta, replica (inv-bcn o inv-vlc). Requisito: cada 5 minutos, por producto y mercado, si el stock mínimo observado ha bajado de 10 unidades, emitir una alerta en el tópico inventario.alertas con la ventana, el mínimo y cuántas actualizaciones hubo. Ventana tumbling de 5 minutos por (producto, mercado), en tiempo de evento (el fecha_ms de inventario, porque una réplica con retraso no debe desplazar la alerta), con marca de agua de 2 minutos (medido: el percentil 99 del desfase es 40 s). Sink idempotente: la clave del mensaje de alerta es ventana|producto|mercado, así que un reprocesamiento produce el mismo mensaje y el consumidor de 08-02 lo tratará como el mismo.

Caso (b): panel de reparto. Las posiciones de furgoneta-3 llegan por MQTT (02-01) y un puente las publica en el tópico reparto.posiciones con clave = id de repartidor: {"repartidor":"furgoneta-3","lat":41.9794,"lon":2.8214,"fecha_ms":...}. Dos cálculos: la distancia recorrida en los últimos 5 minutos, refrescada cada minuto (sliding 5/1 por repartidor: detecta repartidores parados o desviados), y repartidores sin señal (session window con gap de 3 minutos: cuando la sesión se cierra, el repartidor lleva 3 minutos sin enviar; cuando se abre una nueva, ha vuelto). Ambos escriben en reparto.panel, que 08-02 empujará a los navegadores de los operadores.

Antes de usar un motor, conviene ver el mecanismo desnudo. Este script procesa una lista de eventos stock.actualizado con dos tiempos cada uno, el de evento (fecha_ms) y el de llegada (llegada_ms), en orden de llegada, manteniendo ventanas tumbling de 5 minutos por producto y una marca de agua con 2 minutos de retraso tolerado. Los tiempos están en minutos desde las 10:00 para leerlos fácilmente.

Lo que muestra la traza:

• El evento de t=4.9 que llega a las 10:07 sí entra en la ventana 0–5, porque la marca de agua en ese momento era 4,5 (< 5): la ventana seguía abierta gracias al retraso tolerado. Con tiempo de procesamiento habría caído en 5–10, y el mínimo de la ventana 0–5 habría sido 12, sin alerta.
• La ventana 0–5 se emite cuando llega el evento de t=7.3: la marca de agua pasa a 5,3 ≥ 5. Ni antes (no se sabía si faltaban eventos) ni después (no hace falta esperar más).
• El evento de t=4.2 que llega a las 10:09:30 es tardío: su ventana ya se emitió. Aquí se descarta y se cuenta; con allowed lateness se reemitiría la ventana 0–5 con min=8, n=5 (el mínimo no cambia, pero el conteo sí).
• Al acabar la entrada quedan dos ventanas abiertas (10–15): en un flujo real no hay "fin", y se emitirán cuando la marca de agua llegue a 15. En un lote, el fin de la entrada dispara la emisión de todo.

Observa también que la marca de agua no avanza con el evento tardío ni retrocede nunca: es monótona. Y que cada línea de EMITIR es una salida que, si el proceso muriera y reprocesara desde el evento 1, se produciría otra vez con los mismos valores: la clave (ventana, producto) la hace idempotente.

El clúster Flink se añade al docker-compose.yml con dos servicios (jobmanager y taskmanager de la imagen flink:1.19-python, o una imagen propia con pip install apache-flink), y el trabajo se envía con flink run -py. Con la Table API, el DAG del caso (a) es tres sentencias SQL:

Cada pieza corresponde a un apartado: ts AS TO_TIMESTAMP_LTZ(fecha_ms, 3) declara el tiempo de evento; WATERMARK FOR ts AS ts - INTERVAL '2' MINUTE la marca de agua (Flink la genera por partición de Kafka y toma el mínimo, con el idle-timeout para particiones paradas); TUMBLE(..., INTERVAL '5' MINUTE) la ventana con window_start/window_end como columnas; el GROUP BY es el keyBy que reparte el estado por clave entre TaskManagers; execution.checkpointing.interval el checkpoint que guarda estado y offsets en el state.checkpoints.dir configurado (HDFS /km0/checkpoints/); y upsert-kafka con PRIMARY KEY el sink idempotente: un reprocesamiento tras un fallo reescribe la misma clave con el mismo valor. Se lanza y se observa así:

La interfaz web del JobManager (localhost:8081) muestra el DAG desplegado, el paralelismo de cada operador, la marca de agua actual de cada uno, los checkpoints (duración, tamaño) y el backpressure por operador, coloreado.

Para el caso (b), la misma estructura con SESSION para los repartidores sin señal:

Cada fila emitida significa "el repartidor furgoneta-3 envió posiciones de forma continua entre desde y hasta, y después estuvo al menos 3 minutos sin señal": es el aviso del panel. La ventana deslizante de distancia se escribe con HOP(TABLE posiciones, DESCRIPTOR(ts), INTERVAL '1' MINUTE, INTERVAL '5' MINUTE) y una función de agregación propia (la distancia entre posiciones consecutivas necesita el orden, que en SQL se resuelve con LAG sobre una ventana OVER antes de agregar).

El mismo cálculo con la API de DataFrames de 05-03, en modo streaming:

La correspondencia con Flink es directa: withWatermark ↔ WATERMARK FOR, F.window("ts", "5 minutes") ↔ TUMBLE, checkpointLocation ↔ execution.checkpointing, maxOffsetsPerTrigger ↔ backpressure. Dos diferencias importantes. El outputMode("append") con marca de agua emite cada ventana una sola vez, al cerrarla, y descarta los tardíos sin posibilidad de reemitir (no hay allowed lateness); update emitiría resultados provisionales en cada microlote, que exigen un sink que sepa sobrescribir. Y el sink de Kafka de Spark es at-least-once: el mensaje puede duplicarse tras un fallo, y es la clave key (idéntica en el duplicado) la que hace que el consumidor de 08-02 lo trate como una repetición inocua, exactamente el consumidor idempotente de 02-05. Se lanza con spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1 alertas_stock_spark.py, y en localhost:4040 aparece una pestaña Structured Streaming con la tasa de entrada, la duración de cada microlote y la marca de agua.

• Usar tiempo de procesamiento porque es más fácil. Los resultados dependen de la carga, los reinicios y el backpressure, y no se pueden reproducir. Si el evento tiene marca de tiempo (y con la envoltura de 02-05 siempre la tiene), usa tiempo de evento.
• Marca de agua sin medir. Un retraso tolerado inventado descarta eventos válidos o retrasa los resultados. Mide el desfase llegada − fecha_ms en producción (percentil 99) y cuenta los tardíos como métrica permanente.
• Olvidar las particiones inactivas. Una partición de Kafka sin tráfico retiene la marca de agua global y ninguna ventana se emite. Configura el idle timeout (Flink) o revisa que todas las particiones reciban eventos.
• Estado sin límite. Una agregación por clave sin ventana ni TTL crece para siempre (una clave por pedido_id, por ejemplo). Ventanas, TTL de estado, o claves acotadas.
• Sink no idempotente con reprocesamiento. Tras un fallo, las salidas posteriores al checkpoint se repiten. Clave natural + upsert, o sink transaccional. Nunca un INSERT sin clave ni un append a un fichero.
• Checkpoint en disco local. Si el nodo muere, el checkpoint muere con él. HDFS, MinIO o un almacenamiento replicado, siempre.
• Cambiar el DAG y reanudar del checkpoint. Un checkpoint guarda estado por operador; si el DAG cambia (nuevo operador, otra clave) puede no ser compatible. Savepoints con identificadores de operador (uid) en Flink; en Spark, cambios de esquema de estado suelen exigir empezar de cero.
• Confundir lag con latencia. El lag del grupo de consumidores (eventos pendientes en Kafka) crece bajo backpressure y es la señal de que falta paralelismo; la latencia de un resultado en tiempo de evento es siempre al menos el retraso tolerado más el tamaño de la ventana.
• Reescribir la lógica dos veces (Lambda por inercia). Si el mismo agregado se calcula en lote y en flujo, saldrá distinto y nadie sabrá cuál es el bueno. Un código, dos despliegues.

Un análisis del tópico reparto.posiciones durante una semana muestra que el desfase entre fecha_ms y la llegada a Kafka tiene esta distribución: mediana 1,2 s; percentil 95, 8 s; percentil 99, 45 s; percentil 99,9, 4 min (túneles, zonas sin cobertura); máximo 22 min (un móvil apagado que reenvió al encender). El panel debe mostrar la posición y la distancia de los últimos 5 minutos con no más de 1 minuto de retraso. Elige el retraso tolerado de la marca de agua, el tipo y tamaño de ventana, y qué hacer con los eventos tardíos. Justifica los porcentajes de eventos que se descartarán y qué pasa con el móvil que reenvía tras 22 minutos.

Ejecuta mentalmente (o modificando el script) ventana_tumbling.py con RETRASO_TOLERADO = 0 y con RETRASO_TOLERADO = 5. Para cada caso, indica cuándo se emite la ventana 0–5 de queso-curado, con qué mínimo y conteo, y cuántos eventos tardíos hay. ¿Cuál de las tres configuraciones (0, 2, 5) daría la alerta correcta más pronto?

El job de PyFlink de alertas lleva checkpoints cada 30 s. A las 10:07:50 el TaskManager que ejecuta la ventana de queso-curado/girona muere; el último checkpoint completado es de las 10:07:30, y a las 10:07:40 el job había emitido la alerta de la ventana 10:00–10:05. Describe qué hace Flink al recuperarse: desde qué offsets lee, qué pasa con el estado de la ventana 10:05–10:10, si la alerta 10:00–10:05 se vuelve a emitir y qué ve el consumidor de inventario.alertas. Después, explica qué cambiaría si el sink fuera un INSERT en PostgreSQL sin clave primaria, y cómo lo arreglarías.

Ejercicio 1.

El presupuesto de latencia es 1 minuto, y la latencia mínima de un resultado es el retraso tolerado (más el intervalo de emisión). Un retraso tolerado de 45 s (el percentil 99) cumple el presupuesto y descarta como tardíos el 1 % de las posiciones; con 8 s (p95) se descartaría el 5 %, demasiado para una traza de posiciones; con 4 min (p99,9) se violaría el requisito de 1 minuto. Ventana deslizante de 5 minutos con avance de 1 minuto por repartidor (o de 30 s si el panel debe refrescar más a menudo, a costa de más ventanas abiertas por clave: 10 en vez de 5). Los tardíos (1 %) van a una salida lateral que se cuenta y se escribe en el lago: para el panel en tiempo real da igual perder una posición de cada cien, pero la distancia recorrida del día que calcula el lote nocturno debe incluirlas, y para eso el lote lee todas las posiciones del lago, tardías incluidas (una Lambda de facto justificada). El móvil que reenvía tras 22 minutos entrega posiciones cuya ventana cerró hace 20 minutos: todas tardías, todas a la salida lateral; el panel las ignora y el lote las incorpora. Alternativa si el negocio lo pidiera: allowed lateness de 5 minutos para reemitir ventanas recientes, no de 22 (mantendría 27 minutos de ventanas en estado por repartidor).

Ejercicio 2.

Con RETRASO_TOLERADO = 0 la marca de agua es el máximo tiempo de evento visto. Al llegar t=6.1 (llegada 6,2) la marca de agua es 6,1 ≥ 5 y la ventana 0–5 se emite con los eventos vistos hasta entonces: 0,5, 1,2, 3,8 y... el de t=4.9 llegó a las 7,0, después, así que la ventana se emite con min=12, n=3: sin alerta, incorrecta. Los eventos de t=4.9 (llegada 7,0) y t=4.2 (llegada 9,5) son ambos tardíos: 2 tardíos. Emisión más temprana, resultado equivocado.

Con RETRASO_TOLERADO = 5, la ventana 0–5 se emite cuando la marca de agua alcanza 5, es decir, al ver un evento con t ≥ 10: el de t=10.2 (llegada 10,3). Para entonces han entrado 0,5, 1,2, 3,8, 4,9 y 4,2 (que llegó a las 9,5 con la ventana aún abierta): min=8, n=5, correcto y completo, 0 tardíos. Pero la alerta sale a las 10:10:18, cinco minutos más tarde que con retraso 2 (10:07:24, min=8, n=4).

La configuración con 2 minutos da la alerta correcta (el mínimo 8 estaba en ambas) más pronto; la de 5 minutos da además el conteo exacto; la de 0 falla. Es el compromiso completitud/latencia del apartado 2, y la razón de que el retraso tolerado se mida y no se adivine.

Ejercicio 3.

Flink detecta la muerte del TaskManager (heartbeat), reinicia el job entero (o la región afectada, con fine-grained recovery) desde el checkpoint de las 10:07:30: restaura el estado de todos los operadores tal como estaba entonces (la ventana 10:05–10:10 con los eventos anteriores a las 10:07:30 ya aplicados; la 10:00–10:05, que aún no se había emitido en ese instante, también restaurada con su estado) y reposiciona el consumidor de Kafka en los offsets guardados en ese checkpoint. Los eventos entre las 10:07:30 y las 10:07:50 se vuelven a leer y a aplicar sobre ese estado: ninguno se cuenta dos veces, porque el estado que los contenía se descartó. La marca de agua vuelve a avanzar y la ventana 10:00–10:05 se emite de nuevo, con el mismo contenido. Como el sink es upsert-kafka con clave (ventana_inicio, producto, mercado), el tópico recibe un segundo mensaje con la misma clave y el mismo valor; el consumidor de 08-02 (o la compactación de Kafka) lo trata como una actualización sin cambios. Exactly-once en el estado, y efectivamente una sola alerta visible.

Con un INSERT en PostgreSQL sin clave, la fila de la alerta 10:00–10:05 existiría dos veces. Arreglos, de menor a mayor esfuerzo: clave primaria (ventana_inicio, producto, mercado) con INSERT ... ON CONFLICT DO UPDATE (sink idempotente, el JDBC sink de Flink lo hace con upsert); o el sink JDBC en modo exactly-once con XA (dos fases, confirma con el checkpoint), que añade la latencia del checkpoint a cada alerta. Para alertas, la primera opción es la correcta.

Procesar un flujo es ejecutar de forma permanente el mismo DAG de operadores que un lote, sobre una entrada que no termina, y eso obliga a responder preguntas que el lote no tenía: cuándo está completa una ventana (la marca de agua, derivada del tiempo de evento y de un retraso tolerado que se mide), qué hacer con lo que llega después (descartar, reemitir o desviar los tardíos), cómo agrupar (ventanas tumbling, sliding y session por clave), cómo hacer durable un estado que vive para siempre (checkpoints consistentes con los offsets, savepoints para desplegar) y cómo no contar dos veces tras un fallo (exactly-once en el estado por checkpoint, y en el sink por idempotencia o transacción). Backpressure protege al motor dejando que Kafka absorba los picos, y el lag es la señal. Lambda y Kappa son dos formas de convivir con los lotes: la primera duplica la lógica, la segunda la unifica, y los motores modernos hacen que la elección sea de despliegue. En Kilómetro Cero, simulaciones/ventana_tumbling.py hizo visible el mecanismo, PyFlink resolvió la alerta de stock bajo con tres sentencias SQL y un sink upsert-kafka, la session window detecta repartidores sin señal, y Spark Structured Streaming demostró que el código de 05-03 sirve casi sin cambios con withWatermark y window.

Con esta lección, analitica tiene sus dos mitades: los lotes de ventas_diarias.py y ALS, y los flujos de alertas y reparto. Pero los lotes no se lanzan solos. Alguien tiene que esperar a que el fichero del día esté completo en HDFS, validarlo, lanzar spark-submit, cargar el resultado en la base de datos de analitica, avisar si algo falla y reintentar, y hacerlo cada día, y para los siete días de la Semana de la Vendimia cuando se corrigió el precio del vino. Hasta ahora eso era un cron y una cadena de scripts. La última lección del módulo trata la planificación de trabajos y los pipelines de datos: cómo expresar esas dependencias como un DAG de tareas en Airflow, con sensores, reintentos, backfill y alertas, para que la plataforma de datos de Kilómetro Cero funcione sin que nadie la lance a mano.