Tot el que hem calculat fins ara en el mòdul partia d'un conjunt tancat: el fitxer del 14 de setembre, els clics de set dies. Però les dades de Quilòmetre Zero no neixen tancades. comandes.esdeveniments rep centenars d'esdeveniments estoc.actualitzat per segon durant una campanya, els 140 repartidors (com furgoneta-3) envien 2,4 milions de posicions al dia, i hi ha preguntes que no poden esperar el lot nocturn: quins productes s'estan quedant sense estoc a Girona ara, on és cada repartidor ara, quin repartidor porta tres minuts sense donar senyal. El processament de fluxos respon a aquestes preguntes executant el càlcul de manera contínua, esdeveniment a esdeveniment, sobre un conjunt que no acaba mai. Això trenca suposicions que en els lots eren gratuïtes: no es pot "esperar a tenir-ho tot" abans d'agrupar, els esdeveniments arriben desordenats i amb retard respecte del moment en què van passar, l'estat del càlcul ha de sobreviure a les fallades sense poder reexecutar "des del principi", i l'entrada pot arribar més ràpid del que es processa. Aquesta lliçó posa nom a cadascun d'aquests problemes (temps d'esdeveniment, marques d'aigua, finestres, estat i checkpoints, exactly-once, backpressure), compara les arquitectures Lambda i Kappa i les tres eines habituals (Kafka Streams, Flink, Spark Structured Streaming), i resol els dos casos d'analitica en temps real: l'alerta d'estoc baix i el panell de repartiment. El lliurament d'aquests resultats al navegador de l'operador queda per a 08-02; aquí acabem en un tòpic de Kafka.

1. Dels lots als fluxos: què canvia
2. Temps d'esdeveniment, temps de processament i marques d'aigua
3. Finestres: tumbling, sliding i session
4. Estat i checkpoints
5. Exactly-once en streaming
6. Backpressure
7. Arquitectures Lambda i Kappa
8. Eines: Kafka Streams, Flink i Spark Structured Streaming
9. Els casos de Quilòmetre Zero: estoc baix i panell de repartiment
10. Pràctica: simulació en Python, PyFlink i Structured Streaming
11. Errors Comuns i Consells
12. Exercicis
13. Conclusió

Recuperem la taula de 05-01 i l'afinem amb el que hem après des de llavors:

Un motor de fluxos executa el mateix tipus de DAG d'operadors que Spark (05-03), però desplegat de manera permanent: cada operador és un procés (o diversos, en paral·lel per clau) que espera esdeveniments, els processa i els emet cap al següent, sense que el DAG acabi mai. La font típica és Kafka (02-04), que aporta el que un flux necessita: particions (paral·lelisme), offsets (posició reprenible) i retenció (rellegir des d'enrere). El sink és un altre tòpic, una base de dades o un panell.

Hi ha dues maneres de moure esdeveniments pel DAG: esdeveniment a esdeveniment (Flink, Kafka Streams: latència de mil·lisegons) i per microlots (Spark Structured Streaming: cada pocs centenars de mil·lisegons o segons es pren el que ha arribat i s'executa com un lot petit). La segona reutilitza tota la maquinària de lots i simplifica l'exactly-once; la primera dona latències menors. Per al panell de repartiment i l'alerta d'estoc, segons són suficients, així que totes dues serveixen.

A 01-05 vam distingir el moment en què una cosa va passar del moment en què un altre node se n'assabenta. En fluxos aquesta distinció és la que té més conseqüències:

• Temps d'esdeveniment (event time): quan va passar el fet, segons el rellotge de qui el va generar. És el data_ms de l'embolcall de 02-05. La posició de furgoneta-3 presa a les 10:04:58 té aquest temps encara que el mòbil l'enviï a les 10:07 per falta de cobertura.
• Temps de processament (processing time): quan l'operador veu l'esdeveniment, segons el seu rellotge. Depèn de la cua a Kafka, de la xarxa, de la càrrega.

La pregunta "quants estoc.actualitzat de formatge-curat hi va haver entre les 10:00 i les 10:05?" només té una resposta correcta en temps d'esdeveniment. Amb temps de processament, un reinici del consumidor a les 10:03 posaria a la finestra 10:05–10:10 esdeveniments que van passar a les 10:02; el resultat dependria del comportament del sistema, no dels fets. Però calcular en temps d'esdeveniment crea un problema nou: quan arriba un esdeveniment amb data_ms = 10:04:58 a les 10:07, la finestra 10:00–10:05 ja s'havia emès? Cal reobrir-la? Quant esperar abans de donar-la per tancada?

La resposta és la marca d'aigua (watermark): una afirmació que el sistema fa fluir pel DAG que diu "no espero més esdeveniments amb temps d'esdeveniment anterior a T". Es genera a la font amb una heurística, normalment màxim temps d'esdeveniment vist − retard tolerat: amb un retard tolerat de 2 minuts, després de veure un esdeveniment de les 10:07:00 la marca d'aigua és 10:05:00, i en aquell moment la finestra 10:00–10:05 es considera completa i s'emet. El retard tolerat és una decisió de negoci i de mesura: quant triguen de debò els esdeveniments a arribar (el percentil 99 del desfasament arribada − data_ms), contra quanta latència s'accepta en el resultat. Poc retard tolerat: resultats ràpids que descarten més esdeveniments; molt: resultats més complets però tardans.

Els esdeveniments que arriben després que la marca d'aigua hagi passat la seva finestra són esdeveniments tardans (late events). Cada motor ofereix tres destins: descartar-los (el comportament per defecte), incorporar-los reemetent la finestra actualitzada durant un marge addicional (allowed lateness a Flink, que obliga a mantenir la finestra a l'estat més temps), o desviar-los a una sortida lateral (side output) per comptar-los, registrar-los o reprocessar-los per lots. Comptar els tardans és obligatori: és la mètrica que diu si el retard tolerat està ben triat.

Amb diverses particions de Kafka, cada partició té la seva pròpia marca d'aigua i la de l'operador és el mínim de totes: una partició sense trànsit (un mercat tancat de nit) reté la marca d'aigua global i bloqueja l'emissió de finestres de tothom. Els motors ho tracten amb idle timeouts que exclouen les particions inactives del mínim.

Sobre un flux infinit, qualsevol agregació ("quants", "el mínim") necessita un límit: la finestra. Les tres formes bàsiques:

Les finestres es combinen amb una clau: "tumbling de 5 minuts per (producte, mercat)" manté una finestra per cada parella, i el paral·lelisme de l'operador és per clau (totes les formatge-curat/girona van a la mateixa instància, com al shuffle de 05-02). I amb el temps d'esdeveniment: és el data_ms el que decideix a quina finestra cau un esdeveniment, i la marca d'aigua la que decideix quan s'emet.

Una finestra oberta, un comptador per clau, l'última posició coneguda de cada repartidor: tot això és estat, i en un flux viu indefinidament a l'operador. Els motors el desen en un state backend local a l'operador (memòria, o RocksDB a disc local per a estats de gigabytes) particionat per clau, exactament com un actor per clau (05-01). El problema és la durabilitat: si el node mor, el seu estat local es perd, i no es pot "reexecutar des del principi" perquè el principi va ser fa mesos.

La solució és el checkpoint: periòdicament (cada 10 s, cada minut) el motor escriu una còpia consistent de l'estat de tots els operadors, juntament amb els offsets de Kafka que aquell estat reflecteix, en emmagatzematge durador (HDFS, MinIO). Davant d'una fallada, restaura l'últim checkpoint en nodes sans i reprèn el consum des d'aquells offsets: els esdeveniments posteriors al checkpoint es tornen a processar, i l'estat acaba sent el mateix que si no hi hagués hagut fallada.

La paraula difícil és consistent: l'estat de tots els operadors ha de correspondre al mateix punt del flux, encara que cadascun vagi per un esdeveniment diferent. Flink ho aconsegueix amb l'algorisme de Chandy-Lamport adaptat (les barreres de checkpoint): la font injecta al flux un marcador amb el número de checkpoint; cada operador, en rebre'l per totes les seves entrades, desa el seu estat i reenvia el marcador; l'estat desat reflecteix exactament els esdeveniments anteriors al marcador. És una instantània distribuïda sense aturar el flux. Spark Structured Streaming ho té més fàcil: cada microlot és una unitat, i el checkpoint registra quins microlots s'han completat amb els seus rangs d'offsets. Kafka Streams desa l'estat en tòpics de Kafka (changelog topics) i el reconstrueix rellegint-los.

Un savepoint és un checkpoint disparat a mà, amb format estable, que es fa servir per aturar el treball, canviar el codi o el paral·lelisme, i reprendre des del mateix estat: l'equivalent d'un desplegament sense perdre "els últims cinc minuts". I la mida de l'estat importa: una session window per repartidor és petita, però "totes les comandes de les últimes 24 hores per client" són gigabytes que cal checkpointejar cada minut; els checkpoints incrementals (només el que ha canviat des de l'anterior) i un TTL per a l'estat que no es toca són les eines.

A 02-05 vam concloure que l'"exactly-once" en el lliurament no existeix, i que l'assolible és at-least-once més consumidors idempotents. En streaming el terme es fa servir amb un significat precís i assolible: l'estat del motor reflecteix cada esdeveniment exactament una vegada, encara que hi hagi fallades i reprocessaments. S'aconsegueix amb el checkpoint de l'apartat anterior: després d'una fallada, l'estat torna al del checkpoint i els esdeveniments posteriors es reapliquen; com que l'estat que reflectia aquests esdeveniments s'ha descartat, no hi ha doble compte.

El que el checkpoint no cobreix és el que ja va sortir del motor: l'alerta escrita al tòpic inventari.alertes, la fila inserida a PostgreSQL, abans de la fallada i després de l'últim checkpoint. En reprocessar, aquests efectes es produeixen una altra vegada. Perquè el resultat extern sigui també exactly-once (end-to-end), el sink ha de ser una de dues coses:

• Idempotent. Escriure de nou el mateix resultat no canvia res: UPSERT per clau (finestra, producte, mercat) a PostgreSQL, un PUT a Redis amb la mateixa clau, un producer de Kafka amb una clau i un consumidor idempotent aigües avall (02-05). És l'opció més simple i la que es recomana sempre que la sortida tingui una clau natural.
• Transaccional. El sink escriu en una transacció que només es confirma quan el checkpoint es completa (two-phase commit sink): Flink amb el producer transaccional de Kafka (DeliveryGuarantee.EXACTLY_ONCE), o amb una taula i una transacció per checkpoint. Entre el pre-commit i el commit les dades existeixen però no són visibles per a consumidors amb isolation.level=read_committed. Afegeix latència (l'interval de checkpoint) i complexitat; es reserva per a sinks sense clau natural (un log d'esdeveniments a Kafka).

Un DAG de fluxos és una cadena de productors i consumidors, i en qualsevol moment un pot anar més a poc a poc que l'anterior: l'operador de finestres escrivint un checkpoint gran, el sink de PostgreSQL saturat, un pic de la Setmana del Formatge Artesà que triplica els estoc.actualitzat. Si l'operador ràpid continués enviant, les cues intermèdies creixerien fins a esgotar la memòria. Backpressure (contrapressió) és el mecanisme pel qual la lentitud es propaga cap enrere: l'operador lent deixa d'acceptar, l'anterior omple el seu búfer de sortida i deixa de llegir de la seva entrada, i així fins a la font, que deixa de consumir de Kafka. Els esdeveniments s'acumulen a Kafka (que està dissenyat per a això, amb retenció de dies), no a la memòria del motor, i el lag del grup de consumidors (02-04) es converteix en la mètrica que indica que el flux no dona l'abast.

Flink ho implementa amb crèdits entre tasques (el receptor anuncia quant pot rebre); Kafka Streams ho té gratis perquè cada instància fa poll només quan ha acabat amb el lot anterior; Spark Structured Streaming ho aproxima limitant quants offsets llegeix per microlot (maxOffsetsPerTrigger). El que cap d'ells no fa és resoldre la causa: si el lag creix de manera sostinguda, cal augmentar el paral·lelisme (més particions i més instàncies), alleugerir l'operador o acceptar resultats aproximats. I compte amb la marca d'aigua: sota backpressure, el temps de processament s'allunya del temps d'esdeveniment, però les finestres continuen sent correctes perquè es defineixen en temps d'esdeveniment; això és precisament el que l'apartat 2 comprava.

Quan l'streaming era nou i poc fiable, la resposta a "vull resultats en temps real però també exactes" va ser l'arquitectura Lambda (Marz, 2011): mantenir dos camins. La capa batch recalcula cada nit, des del llac, les vistes completes i exactes; la capa de velocitat calcula amb streaming les últimes hores, de manera aproximada; una capa de servei combina totes dues en consultar. Funciona, però obliga a escriure i mantenir la mateixa lògica dues vegades, en dos motors, amb dues semàntiques, i a reconciliar-ne les diferències. L'arquitectura Kappa (Kreps, 2014) proposa un sol camí: tot és un flux, amb Kafka retenint l'historial (o un llac d'esdeveniments rellegible), i el "lot" és simplement reprocessar el flux des d'un offset antic amb la mateixa aplicació d'streaming, escrivint en una taula nova i canviant el punter quan arriba al present.

La plataforma de dades de Quilòmetre Zero acaba sent pragmàticament mixta: vendes_diaries.py i ALS són lots perquè necessiten el conjunt complet i no tenen pressa; l'estoc i el repartiment són fluxos perquè la latència és el requisit. El que Kappa aporta és el criteri: si la mateixa lògica s'escriu dues vegades, alguna cosa està malament; que els motors moderns executin el mateix codi en lot i en flux (Spark, Flink) fa que la tria sigui de desplegament, no de reescriptura.

Triem Flink com a motor de fluxos d'analitica per la semàntica de temps d'esdeveniment i per PyFlink, i mantenim Structured Streaming com a alternativa perquè reutilitza l'API de 05-03. Kafka Streams queda anomenat: és l'opció correcta si l'streaming viu dins d'un servei Java i no en una plataforma de dades.

Cas (a): alerta d'estoc baix. inventari publica estoc.actualitzat a comandes.esdeveniments amb cada canvi (04-05 els feia servir per invalidar Redis). Dades de l'esdeveniment: producte, mercat, estoc_actual, delta, replica (inv-bcn o inv-vlc). Requisit: cada 5 minuts, per producte i mercat, si l'estoc mínim observat ha baixat de 10 unitats, emetre una alerta al tòpic inventari.alertes amb la finestra, el mínim i quantes actualitzacions hi va haver. Finestra tumbling de 5 minuts per (producte, mercat), en temps d'esdeveniment (el data_ms d'inventari, perquè una rèplica amb retard no ha de desplaçar l'alerta), amb marca d'aigua de 2 minuts (mesurat: el percentil 99 del desfasament és 40 s). Sink idempotent: la clau del missatge d'alerta és finestra|producte|mercat, així que un reprocessament produeix el mateix missatge i el consumidor de 08-02 el tractarà com el mateix.

Cas (b): panell de repartiment. Les posicions de furgoneta-3 arriben per MQTT (02-01) i un pont les publica al tòpic repartiment.posicions amb clau = id de repartidor: {"repartidor":"furgoneta-3","lat":41.9794,"lon":2.8214,"data_ms":...}. Dos càlculs: la distància recorreguda en els últims 5 minuts, refrescada cada minut (sliding 5/1 per repartidor: detecta repartidors aturats o desviats), i repartidors sense senyal (session window amb gap de 3 minuts: quan la sessió es tanca, el repartidor porta 3 minuts sense enviar; quan se n'obre una de nova, ha tornat). Tots dos escriuen a repartiment.panell, que 08-02 empenyerà als navegadors dels operadors.

Abans de fer servir un motor, convé veure el mecanisme nu. Aquest script processa una llista d'esdeveniments estoc.actualitzat amb dos temps cadascun, el d'esdeveniment (data_ms) i el d'arribada (arribada_ms), en ordre d'arribada, mantenint finestres tumbling de 5 minuts per producte i una marca d'aigua amb 2 minuts de retard tolerat. Els temps són en minuts des de les 10:00 per llegir-los fàcilment.

El que mostra la traça:

• L'esdeveniment de t=4.9 que arriba a les 10:07 sí que entra a la finestra 0–5, perquè la marca d'aigua en aquell moment era 4,5 (< 5): la finestra continuava oberta gràcies al retard tolerat. Amb temps de processament hauria caigut a 5–10, i el mínim de la finestra 0–5 hauria estat 12, sense alerta.
• La finestra 0–5 s'emet quan arriba l'esdeveniment de t=7.3: la marca d'aigua passa a 5,3 ≥ 5. Ni abans (no se sabia si faltaven esdeveniments) ni després (no cal esperar més).
• L'esdeveniment de t=4.2 que arriba a les 10:09:30 és tardà: la seva finestra ja s'ha emès. Aquí es descarta i es compta; amb allowed lateness es reemetria la finestra 0–5 amb min=8, n=5 (el mínim no canvia, però el compte sí).
• En acabar l'entrada queden dues finestres obertes (10–15): en un flux real no hi ha "fi", i s'emetran quan la marca d'aigua arribi a 15. En un lot, el final de l'entrada dispara l'emissió de tot.

Observa també que la marca d'aigua no avança amb l'esdeveniment tardà ni retrocedeix mai: és monòtona. I que cada línia d'EMETRE és una sortida que, si el procés morís i reprocessés des de l'esdeveniment 1, es produiria una altra vegada amb els mateixos valors: la clau (finestra, producte) la fa idempotent.

El clúster Flink s'afegeix al docker-compose.yml amb dos serveis (jobmanager i taskmanager de la imatge flink:1.19-python, o una imatge pròpia amb pip install apache-flink), i el treball s'envia amb flink run -py. Amb la Table API, el DAG del cas (a) són tres sentències SQL:

Cada peça correspon a un apartat: ts AS TO_TIMESTAMP_LTZ(data_ms, 3) declara el temps d'esdeveniment; WATERMARK FOR ts AS ts - INTERVAL '2' MINUTE la marca d'aigua (Flink la genera per partició de Kafka i en pren el mínim, amb l'idle-timeout per a particions aturades); TUMBLE(..., INTERVAL '5' MINUTE) la finestra amb window_start/window_end com a columnes; el GROUP BY és el keyBy que reparteix l'estat per clau entre TaskManagers; execution.checkpointing.interval el checkpoint que desa estat i offsets a l'state.checkpoints.dir configurat (HDFS /km0/checkpoints/); i upsert-kafka amb PRIMARY KEY el sink idempotent: un reprocessament després d'una fallada reescriu la mateixa clau amb el mateix valor. Es llança i s'observa així:

La interfície web del JobManager (localhost:8081) mostra el DAG desplegat, el paral·lelisme de cada operador, la marca d'aigua actual de cadascun, els checkpoints (durada, mida) i el backpressure per operador, acolorit.

Per al cas (b), la mateixa estructura amb SESSION per als repartidors sense senyal:

Cada fila emesa significa "el repartidor furgoneta-3 va enviar posicions de manera contínua entre inici i fi, i després va estar com a mínim 3 minuts sense senyal": és l'avís del panell. La finestra lliscant de distància s'escriu amb HOP(TABLE posicions, DESCRIPTOR(ts), INTERVAL '1' MINUTE, INTERVAL '5' MINUTE) i una funció d'agregació pròpia (la distància entre posicions consecutives necessita l'ordre, que en SQL es resol amb LAG sobre una finestra OVER abans d'agregar).

El mateix càlcul amb l'API de DataFrames de 05-03, en mode streaming:

La correspondència amb Flink és directa: withWatermark ↔ WATERMARK FOR, F.window("ts", "5 minutes") ↔ TUMBLE, checkpointLocation ↔ execution.checkpointing, maxOffsetsPerTrigger ↔ backpressure. Dues diferències importants. L'outputMode("append") amb marca d'aigua emet cada finestra una sola vegada, en tancar-la, i descarta els tardans sense possibilitat de reemetre (no hi ha allowed lateness); update emetria resultats provisionals a cada microlot, que exigeixen un sink que sàpiga sobreescriure. I el sink de Kafka d'Spark és at-least-once: el missatge es pot duplicar després d'una fallada, i és la clau key (idèntica al duplicat) la que fa que el consumidor de 08-02 el tracti com una repetició innòcua, exactament el consumidor idempotent de 02-05. Es llança amb spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1 alertes_estoc_spark.py, i a localhost:4040 apareix una pestanya Structured Streaming amb la taxa d'entrada, la durada de cada microlot i la marca d'aigua.

• Fer servir temps de processament perquè és més fàcil. Els resultats depenen de la càrrega, els reinicis i el backpressure, i no es poden reproduir. Si l'esdeveniment té marca de temps (i amb l'embolcall de 02-05 sempre la té), fes servir temps d'esdeveniment.
• Marca d'aigua sense mesurar. Un retard tolerat inventat descarta esdeveniments vàlids o retarda els resultats. Mesura el desfasament arribada − data_ms en producció (percentil 99) i compta els tardans com a mètrica permanent.
• Oblidar les particions inactives. Una partició de Kafka sense trànsit reté la marca d'aigua global i cap finestra no s'emet. Configura l'idle timeout (Flink) o revisa que totes les particions rebin esdeveniments.
• Estat sense límit. Una agregació per clau sense finestra ni TTL creix per sempre (una clau per comanda_id, per exemple). Finestres, TTL d'estat, o claus acotades.
• Sink no idempotent amb reprocessament. Després d'una fallada, les sortides posteriors al checkpoint es repeteixen. Clau natural + upsert, o sink transaccional. Mai un INSERT sense clau ni un append a un fitxer.
• Checkpoint a disc local. Si el node mor, el checkpoint mor amb ell. HDFS, MinIO o un emmagatzematge replicat, sempre.
• Canviar el DAG i reprendre del checkpoint. Un checkpoint desa estat per operador; si el DAG canvia (operador nou, una altra clau) pot no ser compatible. Savepoints amb identificadors d'operador (uid) a Flink; a Spark, els canvis d'esquema d'estat solen exigir començar de zero.
• Confondre lag amb latència. El lag del grup de consumidors (esdeveniments pendents a Kafka) creix sota backpressure i és el senyal que falta paral·lelisme; la latència d'un resultat en temps d'esdeveniment és sempre com a mínim el retard tolerat més la mida de la finestra.
• Reescriure la lògica dues vegades (Lambda per inèrcia). Si el mateix agregat es calcula en lot i en flux, sortirà diferent i ningú no sabrà quin és el bo. Un codi, dos desplegaments.

Una anàlisi del tòpic repartiment.posicions durant una setmana mostra que el desfasament entre data_ms i l'arribada a Kafka té aquesta distribució: mediana 1,2 s; percentil 95, 8 s; percentil 99, 45 s; percentil 99,9, 4 min (túnels, zones sense cobertura); màxim 22 min (un mòbil apagat que va reenviar en encendre's). El panell ha de mostrar la posició i la distància dels últims 5 minuts amb no més d'1 minut de retard. Tria el retard tolerat de la marca d'aigua, el tipus i mida de finestra, i què fer amb els esdeveniments tardans. Justifica els percentatges d'esdeveniments que es descartaran i què passa amb el mòbil que reenvia després de 22 minuts.

Executa mentalment (o modificant l'script) finestra_tumbling.py amb RETARD_TOLERAT = 0 i amb RETARD_TOLERAT = 5. Per a cada cas, indica quan s'emet la finestra 0–5 de formatge-curat, amb quin mínim i compte, i quants esdeveniments tardans hi ha. Quina de les tres configuracions (0, 2, 5) donaria l'alerta correcta més aviat?

El job de PyFlink d'alertes porta checkpoints cada 30 s. A les 10:07:50 el TaskManager que executa la finestra de formatge-curat/girona mor; l'últim checkpoint completat és de les 10:07:30, i a les 10:07:40 el job havia emès l'alerta de la finestra 10:00–10:05. Descriu què fa Flink en recuperar-se: des de quins offsets llegeix, què passa amb l'estat de la finestra 10:05–10:10, si l'alerta 10:00–10:05 es torna a emetre i què veu el consumidor d'inventari.alertes. Després, explica què canviaria si el sink fos un INSERT a PostgreSQL sense clau primària, i com ho arreglaries.

Exercici 1.

El pressupost de latència és 1 minut, i la latència mínima d'un resultat és el retard tolerat (més l'interval d'emissió). Un retard tolerat de 45 s (el percentil 99) compleix el pressupost i descarta com a tardanes l'1 % de les posicions; amb 8 s (p95) se'n descartaria el 5 %, massa per a una traça de posicions; amb 4 min (p99,9) es violaria el requisit d'1 minut. Finestra lliscant de 5 minuts amb avanç d'1 minut per repartidor (o de 30 s si el panell s'ha de refrescar més sovint, a costa de més finestres obertes per clau: 10 en comptes de 5). Els tardans (1 %) van a una sortida lateral que es compta i s'escriu al llac: per al panell en temps real tant li fa perdre una posició de cada cent, però la distància recorreguda del dia que calcula el lot nocturn les ha d'incloure, i per a això el lot llegeix totes les posicions del llac, tardanes incloses (una Lambda de facto justificada). El mòbil que reenvia després de 22 minuts lliura posicions la finestra de les quals es va tancar fa 20 minuts: totes tardanes, totes a la sortida lateral; el panell les ignora i el lot les incorpora. Alternativa si el negoci ho demanés: allowed lateness de 5 minuts per reemetre finestres recents, no de 22 (mantindria 27 minuts de finestres a l'estat per repartidor).

Exercici 2.

Amb RETARD_TOLERAT = 0 la marca d'aigua és el màxim temps d'esdeveniment vist. En arribar t=6.1 (arribada 6,2) la marca d'aigua és 6,1 ≥ 5 i la finestra 0–5 s'emet amb els esdeveniments vistos fins llavors: 0,5, 1,2, 3,8 i... el de t=4.9 va arribar a les 7,0, després, així que la finestra s'emet amb min=12, n=3: sense alerta, incorrecta. Els esdeveniments de t=4.9 (arribada 7,0) i t=4.2 (arribada 9,5) són tots dos tardans: 2 tardans. Emissió més primerenca, resultat equivocat.

Amb RETARD_TOLERAT = 5, la finestra 0–5 s'emet quan la marca d'aigua arriba a 5, és a dir, en veure un esdeveniment amb t ≥ 10: el de t=10.2 (arribada 10,3). Per llavors han entrat 0,5, 1,2, 3,8, 4,9 i 4,2 (que va arribar a les 9,5 amb la finestra encara oberta): min=8, n=5, correcte i complet, 0 tardans. Però l'alerta surt a les 10:10:18, cinc minuts més tard que amb retard 2 (10:07:24, min=8, n=4).

La configuració amb 2 minuts dona l'alerta correcta (el mínim 8 era a totes dues) més aviat; la de 5 minuts dona a més el compte exacte; la de 0 falla. És el compromís completesa/latència de l'apartat 2, i la raó que el retard tolerat es mesuri i no s'endevini.

Exercici 3.

Flink detecta la mort del TaskManager (heartbeat), reinicia el job sencer (o la regió afectada, amb fine-grained recovery) des del checkpoint de les 10:07:30: restaura l'estat de tots els operadors tal com era llavors (la finestra 10:05–10:10 amb els esdeveniments anteriors a les 10:07:30 ja aplicats; la 10:00–10:05, que encara no s'havia emès en aquell instant, també restaurada amb el seu estat) i reposiciona el consumidor de Kafka als offsets desats en aquell checkpoint. Els esdeveniments entre les 10:07:30 i les 10:07:50 es tornen a llegir i a aplicar sobre aquell estat: cap no es compta dues vegades, perquè l'estat que els contenia s'ha descartat. La marca d'aigua torna a avançar i la finestra 10:00–10:05 s'emet de nou, amb el mateix contingut. Com que el sink és upsert-kafka amb clau (finestra_inici, producte, mercat), el tòpic rep un segon missatge amb la mateixa clau i el mateix valor; el consumidor de 08-02 (o la compactació de Kafka) el tracta com una actualització sense canvis. Exactly-once a l'estat, i efectivament una sola alerta visible.

Amb un INSERT a PostgreSQL sense clau, la fila de l'alerta 10:00–10:05 existiria dues vegades. Arranjaments, de menor a major esforç: clau primària (finestra_inici, producte, mercat) amb INSERT ... ON CONFLICT DO UPDATE (sink idempotent, el JDBC sink de Flink ho fa amb upsert); o el sink JDBC en mode exactly-once amb XA (dues fases, confirma amb el checkpoint), que afegeix la latència del checkpoint a cada alerta. Per a alertes, la primera opció és la correcta.

Processar un flux és executar de manera permanent el mateix DAG d'operadors que un lot, sobre una entrada que no acaba, i això obliga a respondre preguntes que el lot no tenia: quan és completa una finestra (la marca d'aigua, derivada del temps d'esdeveniment i d'un retard tolerat que es mesura), què fer amb el que arriba després (descartar, reemetre o desviar els tardans), com agrupar (finestres tumbling, sliding i session per clau), com fer durador un estat que viu per sempre (checkpoints consistents amb els offsets, savepoints per desplegar) i com no comptar dues vegades després d'una fallada (exactly-once a l'estat per checkpoint, i al sink per idempotència o transacció). Backpressure protegeix el motor deixant que Kafka absorbeixi els pics, i el lag és el senyal. Lambda i Kappa són dues maneres de conviure amb els lots: la primera duplica la lògica, la segona la unifica, i els motors moderns fan que la tria sigui de desplegament. A Quilòmetre Zero, simulacions/finestra_tumbling.py va fer visible el mecanisme, PyFlink va resoldre l'alerta d'estoc baix amb tres sentències SQL i un sink upsert-kafka, la session window detecta repartidors sense senyal, i Spark Structured Streaming va demostrar que el codi de 05-03 serveix gairebé sense canvis amb withWatermark i window.

Amb aquesta lliçó, analitica té les seves dues meitats: els lots de vendes_diaries.py i ALS, i els fluxos d'alertes i repartiment. Però els lots no es llancen sols. Algú ha d'esperar que el fitxer del dia sigui complet a HDFS, validar-lo, llançar spark-submit, carregar el resultat a la base de dades d'analitica, avisar si alguna cosa falla i reintentar, i fer-ho cada dia, i per als set dies de la Setmana de la Verema quan es va corregir el preu del vi. Fins ara això era un cron i una cadena d'scripts. L'última lliçó del mòdul tracta la planificació de treballs i els pipelines de dades: com expressar aquestes dependències com un DAG de tasques a Airflow, amb sensors, reintents, backfill i alertes, perquè la plataforma de dades de Quilòmetre Zero funcioni sense que ningú no la llanci a mà.