alpinashop_analitica ja existeix i té a dins l'històric de comandes. Però aquest històric hi va arribar d'una manera que no es pot repetir cada nit: la Lucía va executar a mà una consulta federada contra la rèplica de PostgreSQL. Va funcionar una vegada. La pregunta és què passa demà, i demà passat, i el dia que algú demani veure les vendes de la campanya avui i no demà.

La resposta ingènua és escriure un script. Un fitxer sincronizar.py que es connecti a Cloud SQL, llegeixi les comandes del dia, les transformi i les insereixi a BigQuery, llançat per un cron en una màquina virtual. Aquest script funcionarà durant setmanes, i després fallarà. Fallarà perquè les dades creixen i l'script triga cinc hores; perquè la VM es reinicia a mig procés i ningú no sap si va escriure la meitat de les files; perquè una comanda amb un caràcter estrany llança una excepció i es perd la resta del bolcat; perquè ningú no sap si es va executar ahir; i perquè el dia que algú demani dades en temps real, tot el disseny s'ha de llençar.

Cloud Dataflow és el servei gestionat de Google per executar pipelines de dades escrits amb Apache Beam. La seva promesa concreta és doble: escrius la lògica de transformació una sola vegada i Google s'encarrega del paral·lelisme, del reintent, de l'escalat i de la coherència; i aquest mateix codi serveix tant per processar dos anys d'històric com per processar les comandes a mesura que entren.

En aquesta lliçó entendràs el model de Beam, escriuràs un pipeline per lots que carrega l'històric d'AlpinaShop, el provaràs en local, el llançaràs a Dataflow, i després entraràs a la part que de debò separa el processament de dades seriós de l'amateur: el temps. Què vol dir "les vendes de les 10:00" quan un esdeveniment generat a les 10:00 arriba a les 10:07, i com es respon a això sense mentir.

1. Què és un pipeline de dades i per què un script no n'hi ha prou
2. Apache Beam: el model unificat
3. Els quatre conceptes: Pipeline, PCollection, PTransform, runner
4. Les transformacions que faràs servir el 90 % del temps
5. Primer pipeline per lots, explicat línia a línia
6. Execució local amb DirectRunner
7. Execució gestionada amb DataflowRunner
8. El temps en streaming: esdeveniment davant de procés
9. Marques d'aigua, finestres, activadors i dades tardanes
10. El pipeline en temps real de pedidos-nuevos
11. Plantilles de Dataflow: la via pràctica
12. Escalat automàtic i Dataflow Prime
13. Monitoratge, paral·lelisme i biaix de dades
14. Cost: què es paga exactament
15. Quan Dataflow no és la resposta

Un pipeline de dades és una seqüència declarada d'operacions que porten dades d'un origen a un destí transformant-les pel camí. L'adjectiu important és declarada: descrius què vols que passi, no com es reparteix la feina entre màquines.

Comparem amb honestedat l'script en una VM i el pipeline gestionat:

La fila que més fa mal a la pràctica és la de la fallada d'un registre. Un bolcat de 800.000 comandes en què la comanda 412.337 té un import amb coma en comptes de punt no ha de perdre les 387.663 comandes restants. Un script mal fet les perd; un de ben fet requereix un esforç considerable de tractament d'errors que a Beam ve de sèrie amb les sortides etiquetades.

I la fila del cost en repòs té el seu matís: Dataflow en mode lot costa zero quan no s'executa, però un pipeline en temps real està permanentment encès i factura sense parar. Hi tornarem a l'apartat 14, perquè és la sorpresa més habitual.

Apache Beam és un model de programació open source —donat per Google a l'Apache Software Foundation— per definir pipelines de dades que després s'executen en motors diferents.

La paraula clau és unificat. Abans de Beam, processar per lots i processar en temps real eren mons separats amb eines separades, i les empreses mantenien dues implementacions de la mateixa lògica de negoci: una per a l'històric i una altra per al temps real, que inevitablement divergien i donaven números diferents. Beam parteix de la idea que un lot és simplement un flux acotat i un stream és un flux no acotat, i que la lògica de transformació és la mateixa en tots dos casos.

El runner és el motor que executa el pipeline. El mateix fitxer Python pot córrer al teu portàtil amb DirectRunner, a Dataflow amb DataflowRunner, o en un clúster de Flink o Spark. Això redueix el bloqueig amb el proveïdor: si AlpinaShop hagués de sortir de Google Cloud, la lògica dels seus pipelines viatjaria.

Beam té SDK per a Java, Python i Go. Farem servir Python, coherent amb la resta de l'aplicació d'AlpinaShop, que ja és Flask.

Fixar la versió és deliberat: els treballadors de Dataflow faran servir exactament la versió del SDK amb què llances el pipeline, i les diferències entre versions són una font clàssica de fallades que només apareixen al núvol.

Pipeline és l'objecte que conté el graf complet. Es construeix, es declara i s'executa. No passa res mentre l'escrius: estàs dibuixant un plànol.

PCollection és un conjunt de dades distribuït i immutable. No és una llista de Python: pot tenir zero elements o bilions, pot estar repartida per cent màquines, i no es pot modificar. Cada transformació produeix una PCollection nova. La immutabilitat és el que permet reintentar un fragment fallit sense corrompre res.

Una PCollection pot ser:

• Acotada (bounded): té un final conegut. Un fitxer, una taula. És el cas de lot.
• No acotada (unbounded): no acaba mai. Un topic de Pub/Sub. És el cas de temps real.

PTransform és una operació que pren una o més PCollection i produeix una o més PCollection. S'aplica amb l'operador |, que a Beam està sobrecarregat per significar "aplica aquesta transformació".

Runner és el motor d'execució, ja vist.

La sintaxi, que resulta estranya la primera vegada:

L'operador >> associa un nom al pas. No és decoratiu: aquest nom és el que apareix al graf de la consola de Dataflow i a les mètriques. Un pipeline amb passos anomenats Map(<lambda at main.py:34>) és impossible de depurar en producció. Anomena tots els passos, sempre.

Aquest graf és exactament el que escriuràs a l'apartat 5. Fixa't que la branca d'errors és part del disseny, no un afegit.

Dos aclariments que eviten errors conceptuals importants:

GroupByKey davant de CombinePerKey. GroupByKey porta tots els valors d'una clau a una sola màquina i els materialitza en memòria. Si un sku té tres milions de línies, aquella màquina pot rebentar. CombinePerKey amb una funció associativa i commutativa (sum, max, min) fa agregació parcial a cada treballador abans de moure res per la xarxa: cada worker suma el seu i només viatgen els resultats parcials. Sempre que puguis fer servir CombinePerKey, fes-lo servir; és ordres de magnitud més eficient i no es trenca amb claus calentes.

Map davant de ParDo. Map és sucre sintàctic sobre ParDo. Fes servir Map per a transformacions simples i sense estat; fes servir ParDo amb una classe DoFn quan necessitis inicialització costosa (obrir un client d'API una vegada per treballador, no una vegada per element), mètriques pròpies, o diverses sortides.

Objectiu concret: llegir els fitxers CSV d'exportació de comandes que hi ha a gs://alpinashop-catalogo/exportaciones/2026/03/14/, validar-los, netejar-los, calcular un agregat de vendes per SKU i escriure dues coses a alpinashop_analitica: les línies netes a lineas_pedido i l'agregat en una taula nova ventas_diarias_sku. Els registres defectuosos van a un fitxer de quarantena al bucket.

Tot el de dalt és Python normal. Les constants en majúscules van al principi perquè canviar de projecte no obligui a buscar cadenes pel fitxer.

Punts importants d'aquesta classe:

• yield en comptes de return. Un DoFn és un generador: pot emetre zero, un o molts elements per entrada. Un return amb valor no funciona com esperes.
• TaggedOutput marca l'element perquè surti per una branca diferent del graf. És el mecanisme de Beam per a "això ha anat malament però el pipeline continua". Sense ell, una excepció reintenta el paquet quatre vegades i després mata el treball sencer.
• replace(",", ".") normalitza els decimals de l'ERP espanyol, que exporta 89,90. És exactament el tipus de brutícia real que hi ha en qualsevol exportació.
• Els comptadors (Metrics.counter) pugen a la consola de Dataflow i permeten respondre "quantes files es van rebutjar ahir a la nit?" sense obrir un registre.

Detall que costa car descobrir en producció: els valors NUMERIC es passen a WriteToBigQuery com a cadena, no com a float. Si converteixes a float per serialitzar, has reintroduït l'error de coma flotant que vam evitar amb tanta cura a 04-01.

Ara el pipeline complet:

Cinc coses que mereixen comentari:

1. with beam.Pipeline(...) as p: en sortir del bloc with, Beam crida run() i espera. Sense el with, cal cridar p.run().wait_until_finish() explícitament.
2. WRITE_APPEND davant de WRITE_TRUNCATE: el detall s'afegeix (volem històric acumulat); l'agregat es reescriu sencer (volem la foto vigent). Triar malament aquí duplica dades silenciosament.
3. additional_bq_parameters crea la taula ja particionada i agrupada en clústers si no existia. És la manera de no perdre el que hem après a 04-01 quan la taula la crea el pipeline.
4. beam.Flatten() uneix diverses PCollection del mateix tipus en una. És la unió de branques del graf, l'equivalent a un UNION ALL.
5. save_main_session=True serialitza l'àmbit global del mòdul per als treballadors. Sense això, un pipeline que funciona en local falla a Dataflow amb NameError sobre les constants o els imports. És l'error de novell número u.

Abans de gastar un cèntim, es prova en local. El DirectRunner executa el pipeline a la teva màquina, amb un subconjunt de dades.

Aquest fitxer de prova està brut expressament, i així és com ha de ser qualsevol joc de proves d'un pipeline:

• La primera línia té nou camps perquè 89,90 hi posa una coma de més. L'analitzador la rebutjarà amb "esperades 8 columnes". És exactament la fallada real d'un ERP mal configurat.
• La tercera té quantitat negativa: la rebutja el validador.
• La quarta té una data invàlida: la rebutja el validador.
• Només la segona passa.

El DirectRunner és deliberadament estricte: comprova la immutabilitat dels elements, serialitza i deserialitza entre passos, i desordena les dades expressament. Si el teu pipeline depèn de l'ordre d'arribada o modifica un objecte in situ, el DirectRunner ho detecta i falla, mentre que al núvol produiria resultats incorrectes de manera intermitent. Que sigui lent i primmirat és la funcionalitat, no un defecte.

Bona pràctica: a més de provar el pipeline sencer, prova les transformacions amb unittest i les utilitats de Beam:

beam.Create(...) fabrica una PCollection a partir d'una llista de Python: és la manera d'injectar dades de prova. assert_that amb equal_to compara el contingut sense importar l'ordre, que és el correcte en un sistema distribuït.

Amb les proves en verd, al servei real.

I el llançament:

Opció per opció, perquè cadascuna té conseqüències:

Comprovació de l'estat:

Aquesta última comanda retorna els comptadors files_ok i files_rebutjades que hem instrumentat. Aquest és el retorn d'haver fet servir mètriques de Beam en comptes de print.

Aquí comença la part difícil, i també la que fa que valgui la pena aprendre Beam en comptes d'improvisar.

En lot, el temps és simple: tens el fitxer, té un final, processes i acabes. En temps real no hi ha final, i apareix una distinció que ho canvia tot:

• Temps de l'esdeveniment (event time): quan va passar el fet al món real. El client va prémer "Comprar" a les 10:00:00.
• Temps de procés (processing time): quan la dada arriba al teu pipeline. 10:00:03, o 10:07:12 si el mòbil era en un túnel, o 11:30 si l'aplicació va desar l'esdeveniment en local fins a recuperar cobertura.

Amb AlpinaShop és molt concret. Un client al metro de Barcelona navega pel catàleg, afegeix una motxilla a la cistella a les 18:42, perd cobertura, i l'aplicació envia els esdeveniments acumulats a les 18:51.

Ara la pregunta de negoci: quants productes es van veure entre les 18:40 i les 18:45? Si agrupes per temps de procés, la resposta és zero, i és falsa. Si agrupes per temps de l'esdeveniment, la resposta inclou aquest client, i és la correcta.

Beam agrupa per temps de l'esdeveniment per defecte. Aquesta és la seva decisió de disseny més important i la raó per la qual els seus números quadren amb els de l'informe per lots de l'endemà. Un sistema que agrupa per temps de procés dona resultats que depenen de la xarxa i que mai no són reproduïbles: reprocessar el mateix dia dona un resultat diferent.

Si esperes per temps de l'esdeveniment, sorgeix la pregunta inevitable: quan deixes d'esperar? Un esdeveniment de les 18:42 podria arribar demà. Tanques la finestra o esperes eternament?

La resposta de Beam són quatre mecanismes que es combinen.

Trossegen la PCollection no acotada en trossos finits sobre els quals sí que es pot agregar.

La finestra de sessió és especialment elegant i no té equivalent senzill en SQL: la seva durada no es fixa per endavant, la determinen les mateixes dades. És exactament la definició de "sessió de navegació" que necessita la taula visitas de 04-01.

La marca d'aigua és l'estimació que fa el sistema de "ja no espero esdeveniments anteriors a aquest instant". Dataflow la calcula automàticament observant les marques de temps de les dades que van arribant: si de Pub/Sub fa deu minuts que arriben esdeveniments de les 18:50 endavant, la marca d'aigua avança més enllà de les 18:45 i les finestres anteriors es poden tancar.

No és una garantia, és una heurística. Sempre pot arribar alguna cosa després. Per això existeixen els altres dos mecanismes.

L'activador decideix quan emetre el resultat d'una finestra. Per defecte, en passar la marca d'aigua: un resultat per finestra, quan es considera completa.

Però el tauler de la campanya de tardor no pot esperar una hora a veure el primer número. Amb un activador s'emeten resultats parcials:

Interpretació pràctica d'aquesta configuració per a AlpinaShop:

• Cada minut s'emet un resultat provisional: el tauler es mou i la gent veu que el sistema és viu.
• Quan passa la marca d'aigua, s'emet el resultat considerat definitiu.
• Durant dues hores més (allowed_lateness), si arriben esdeveniments d'aquella hora —el client del metro—, s'emet una correcció cada deu minuts.
• Passades les dues hores, el que arribi es descarta. Aquest descart és una decisió de negoci explícita, no un accident, i cal instrumentar-lo amb un comptador per saber quant es perd.

AccumulationMode.ACCUMULATING significa que cada emissió conté el total acumulat de la finestra, així que el destí ha de sobreescriure. L'alternativa, DISCARDING, emet només el que és nou des de l'última emissió, i el destí ha de sumar. Confondre-les produeix xifres doblades o dividides, i és un error molt difícil de veure en un tauler.

Tot el que arriba després de la marca d'aigua és una dada tardana. Amb allowed_lateness decideixes quant de temps l'acceptes. La regla és simple: com més esperes, més correctes són els números i més recursos (estat en memòria) consumeix el pipeline. Dues hores per a un comerç electrònic europeu és un valor raonable; dos dies seria caríssim i no canviaria les decisions.

A la propera lliçó, 04-04, crearem el topic de Pub/Sub pedidos-nuevos, on l'aplicació Flask publicarà un missatge JSON cada vegada que es confirmi una comanda. Anticipem el consumidor, perquè és el cas d'ús canònic de Dataflow.

Tres decisions que cal entendre:

• timestamp_attribute="momento_evento": li diu a Beam que faci servir l'atribut del missatge com a temps de l'esdeveniment, en comptes del moment de publicació a Pub/Sub. Sense això, un missatge retingut nou minuts al mòbil es comptabilitzaria a l'hora equivocada. És la línia que fa que tot l'apartat 9 funcioni de debò, i s'oblida constantment.
• method="STORAGE_WRITE_API": fa servir l'API moderna d'escriptura, més barata i amb semàntica d'exactament una vegada, en comptes de les insercions en streaming clàssiques.
• Es llegeix d'una subscripció, no del topic. Amb una subscripció, si el pipeline s'atura, els missatges s'acumulen i es recuperen en arrencar. Llegir del topic directament fa que Dataflow creï una subscripció efímera i es perdi tot el que s'ha publicat mentre el pipeline està aturat.

Tot l'anterior és potent i també és feina. Per a tasques comunes existeix una cosa molt més simple: les plantilles.

Una plantilla és un pipeline ja compilat i parametritzat que es llança sense escriure ni compilar codi. Google en publica desenes.

Llançar la de Pub/Sub a BigQuery per a AlpinaShop és una línia:

Fixa't en outputDeadletterTable: els missatges que no encaixin amb l'esquema van a una taula d'errors en comptes de bloquejar el pipeline. És el mateix patró de quarantena que vam programar a mà, ja resolt.

Hi ha dos sabors:

• Plantilles clàssiques: el graf es compila en crear-la; els paràmetres només poden ser valors en temps d'execució (ValueProvider).
• Plantilles flexibles (Flex Templates): el pipeline s'empaqueta com a imatge de contenidor a Artifact Registry i el graf es construeix en llançar. Són més flexibles i són l'opció recomanada avui per a plantilles pròpies.

Crear una plantilla flexible del pipeline per lots d'AlpinaShop, perquè Composer o Cloud Scheduler la invoquin a 04-06:

La imatge va al mateix Artifact Registry que ja fa servir el catàleg (europe-west1-docker.pkg.dev/alpinashop-prod/alpinashop/), cosa que manté un únic inventari d'artefactes, coherent amb el que veurem al mòdul 6.

El criteri d'AlpinaShop: per a "Pub/Sub a BigQuery" sense transformació, la plantilla de Google, sense escriure ni una línia. Per al pipeline per lots amb validació de negoci pròpia, codi Beam empaquetat com a plantilla flexible. Escriure codi només quan aporta lògica que la plantilla no té.

Dataflow ajusta el nombre de treballadors durant l'execució. En lot, mira la feina pendent; en temps real, mira el retard de la cua (backlog) i l'ús de CPU.

Posar un max_num_workers baix no sempre estalvia: un pipeline que triga deu hores amb dos treballadors pot costar el mateix que un que triga una hora amb vint, perquè es factura per treballador i hora. El que sí que evita el sostre és la sorpresa d'un treball amb un bucle patològic consumint cent màquines tota la nit.

Dataflow Prime és l'evolució del servei amb tres diferències pràctiques:

L'avantatge real de Prime apareix quan un pas del pipeline necessita molta memòria i els altres no: al model clàssic dimensiones totes les màquines per al pas més exigent i malbarates recursos la resta del temps. S'activa amb --dataflow_service_options=enable_prime.

Per a AlpinaShop, amb pipelines modestos i predictibles, el model clàssic amb n2-standard-2 és suficient i més fàcil de raonar a la factura. Prime queda anotat per a quan el volum ho justifiqui.

La consola de Dataflow mostra el graf d'execució amb cada pas i, a cadascun, elements processats, temps de CPU i estat. És la millor interfície de depuració de dades de la plataforma, i per això insistim a anomenar els passos.

Mètriques que cal mirar:

El problema més freqüent i més difícil de diagnosticar és el biaix de dades (data skew): una clau concentra una proporció enorme dels elements. A AlpinaShop és molt fàcil que passi: si agrupes visites per sku i la motxilla estrella acumula el 40 % del trànsit, un sol treballador processarà el 40 % de la feina mentre els altres dinou esperen. El símptoma és inconfusible: l'escalat no millora res i hi ha un pas amb un treballador al 100 % i la resta ociosos.

Tres remeis, per ordre de preferència:

La tècnica 2, la sal, mereix explicació: en afegir un sufix aleatori a la clau, la motxilla estrella es converteix en vint claus diferents que es reparteixen entre vint treballadors; després es treu la sal i es fa una segona agregació sobre vint valors, que és trivial. Només funciona amb operacions associatives, però cobreix gairebé tots els casos d'agregació.

La tècnica 3, l'entrada lateral (side input), és la que més faràs servir a AlpinaShop: el catàleg de productes són uns quants milers de files i cap a la memòria de cada treballador. Difondre'l evita completament el shuffle del JOIN. Compte amb el límit: si l'entrada lateral no cap a la memòria, el pipeline es degrada moltíssim.

Dataflow no factura per pipeline ni per dada processada: factura els recursos consumits pels treballadors.

Un exemple realista per a AlpinaShop: el pipeline per lots nocturn, amb 4 treballadors n2-standard-2 (2 vCPU, 8 GB) durant 20 minuts, surt per cèntims. El pipeline en temps real, en canvi, funciona 24×7: 2 treballadors permanents són unes 1.440 hores de vCPU al mes, de l'ordre de 80-100 € mensuals. Aquest és el número que sorprèn tothom, i és la raó de l'advertiment del principi de la lliçó.

Consells concrets per no pagar de més:

• Activa Streaming Engine i Shuffle Service (--enable_streaming_engine, --experiments=shuffle_mode=service). Mouen el shuffle i l'estat fora dels treballadors, permetent màquines més petites i discos molt menors. Gairebé sempre surt a compte.
• Redueix el disc. Amb Streaming Engine, --disk_size_gb=30 n'hi ha prou; el valor per defecte és molt més gran i es paga per hora.
• Posa --max_num_workers sempre. És el fre de mà.
• Fes servir VM Spot en lot tolerant a interrupcions: --flexrs_goal=COST_OPTIMIZED endarrereix l'arrencada fins a 6 hores a canvi d'un descompte notable. Perfecte per al bolcat nocturn; inacceptable per a temps real.
• De debò necessites streaming? Un microlot cada 15 minuts amb la plantilla de Cloud Storage a BigQuery costa una fracció d'un pipeline permanent. Si el negoci tolera 15 minuts de retard, la resposta és no.
• Vigila els pipelines de streaming oblidats. Un job de proves que ningú no va aturar és la partida fantasma més comuna a la factura de dades. El detectaràs amb gcloud dataflow jobs list --status=active.

L'honestedat sobre els límits és part de saber fer servir una eina.

La confusió més habitual és la de la penúltima fila. Dataflow no és un orquestrador. Pot executar un pipeline complexíssim, però no sap "primer exporta Cloud SQL, després carrega a BigQuery, després llança aquest pipeline, i si alguna cosa falla avisa la Marta". Això és exactament el que resol 04-06.

Oblidar save_main_session=True. El pipeline funciona en local i falla a Dataflow amb NameError: name 'PROJECTE' is not defined. Els treballadors no reben l'àmbit global del mòdul si no els ho dius.

No fixar la versió del SDK. Els treballadors fan servir la versió amb què vas llançar el pipeline. Un pip install apache-beam[gcp] sense versió fa que el mateix codi funcioni avui i falli demà. Fixa la versió a requirements.txt.

No posar noms als passos. Sense "Nom" >>, el graf de la consola és il·legible i les mètriques no diuen res. A més, canviar el nom d'un pas impedeix actualitzar un pipeline en temps real en marxa (--update), perquè Beam no pot mapar l'estat del pas antic al nou.

Fer servir GroupByKey on hi cap CombinePerKey. És la diferència entre un pipeline que escala i un que cau amb una clau calenta.

Llegir d'un topic en comptes d'una subscripció. Amb topic, Dataflow crea una subscripció temporal i es perd tot el que s'ha publicat mentre el pipeline està aturat. Amb subscripció, els missatges s'acumulen i es recuperen.

Oblidar timestamp_attribute. Les finestres es calculen sobre el moment de publicació en comptes del moment de l'esdeveniment, i els números no quadren amb els del procés per lots. És subtil i greu.

Deixar un pipeline de streaming de proves encès. Factura 24×7. Posa'ls etiquetes i revisa gcloud dataflow jobs list --status=active amb periodicitat.

No fer servir el compte de servei propi. Sense --service_account_email, els treballadors fan servir el compte per defecte de Compute Engine, que en molts projectes és Editor. Un pipeline amb permisos d'Editor sobre producció és un risc innecessari.

Consell: escriu sempre la branca d'errors. Un pipeline sense sortida de rebuigs no és un pipeline de producció. Si no pots veure què s'ha descartat i per què, no pots confiar en els números.

Consell: --update per modificar un pipeline en temps real. Permet substituir el codi conservant l'estat en curs, en comptes de drenar-lo i arrencar de zero. Requereix compatibilitat del graf, cosa que és una altra raó per no canviar el nom dels passos alegrement.

Consell: drena, no cancel·lis. gcloud dataflow jobs drain deixa de llegir entrades noves i acaba el que té en vol. cancel mata el treball i pot perdre dades en procés.

Escriu un pipeline per lots en Beam que llegeixi gs://alpinashop-catalogo/exportaciones/2026/03/opiniones-*.csv amb les columnes opinion_id,sku,fecha,puntuacion,texto,pais, i que:

• rebutgi les files la puntuacion de les quals no sigui un enter entre 1 i 5, o el pais de les quals no sigui un codi de dues lletres;
• normalitzi el sku a majúscules i retalli el texto a 500 caràcters;
• escrigui les vàlides a alpinashop-datos:alpinashop_analitica.opiniones;
• escrigui les rebutjades, amb el seu motiu, a gs://alpinashop-catalogo/cuarentena/opiniones/;
• porti un comptador de vàlides i rebutjades visible a Dataflow.

Prova'l amb DirectRunner i un fitxer local amb almenys dues files defectuoses.

Direcció vol un tauler de la campanya de tardor amb les vendes per hora i per país. Requisits: agrupar per temps de l'esdeveniment, mostrar un avanç parcial cada 30 segons perquè el tauler es vegi viu, acceptar esdeveniments amb fins a 90 minuts de retard emetent correccions, i que cada emissió contingui el total acumulat de l'hora. Escriu únicament el fragment de WindowInto i l'agregació, i explica què escriu exactament el destí i per què el mode d'acumulació escollit obliga a un write_disposition concret.

El pipeline en temps real de visites porta tres dies funcionant. Des d'ahir, el system lag ha passat de 4 segons a 22 minuts i continua creixent. L'escalat automàtic ha arribat als 20 treballadors (el seu màxim), però l'ús mitjà de vCPU del conjunt és del 18 %. Al graf, el pas AgruparPerSKU mostra un treballador amb 6 hores de temps de CPU i els altres amb menys de 10 minuts. Ahir, màrqueting va llançar una campanya de la motxilla MOCH-40L-AZ que ha multiplicat per dotze les seves visites.

Diagnostica la causa, explica per què pujar --max_num_workers a 50 no arreglaria res, i proposa dues solucions de codi amb la seva diferència pràctica.

Solució 1

Fitxer de prova amb defectes deliberats:

La primera passa (el sku es normalitza a majúscules i es a ES); les tres següents van a quarantena amb motius diferents. Mètriques esperades: valides=1, rebutjades=3.

Solució 2

Què escriu exactament el destí. Amb ACCUMULATING, cada emissió d'una finestra conté el total acumulat des de l'inici de la finestra, no l'increment. Per a la finestra 10:00-11:00 i el país ES, el destí rebrà una seqüència com: 120 € (a les 10:00:30), 345 € (10:01:00), … 4.210 € (en passar la marca d'aigua), 4.235 € (a les 11:20, correcció per un esdeveniment tardà).

Per què obliga a un write_disposition concret. Si es fes WRITE_APPEND sobre la taula final, aquestes sis emissions se sumarien i el tauler mostraria més de 9.000 € on n'hi ha 4.235: la dada quedaria multiplicada. Les opcions correctes són:

• Escriure en una taula d'estats intermedis amb WRITE_APPEND i que el tauler consulti només l'última emissió per finestra i clau, per exemple amb QUALIFY ROW_NUMBER() OVER (PARTITION BY hora_inicio, pais ORDER BY momento_emision DESC) = 1. És l'opció recomanada en temps real, perquè conserva l'historial de correccions i permet auditar.
• O bé fer servir un destí que admeti sobreescriptura per clau (MERGE posterior sobre hora_inicio + pais).

Si en canvi es triés AccumulationMode.DISCARDING, cada emissió portaria només l'increment i llavors WRITE_APPEND amb una suma posterior seria el correcte. La combinació mode d'acumulació + disposició d'escriptura s'ha de decidir conjuntament; equivocar-se produeix taulers amb xifres inflades que ningú no detecta fins que algú creua el número amb comptabilitat.

Solució 3

Diagnòstic: biaix de dades (data skew) per clau calenta.

Les tres evidències apunten al mateix i es reforcen entre si:

1. 20 treballadors amb 18 % de CPU mitjana. Si el pipeline estigués realment saturat, la CPU estaria alta. Un ús baix amb retard creixent significa que els treballadors esperen, no que treballin.
2. Un treballador amb 6 hores de CPU i la resta amb 10 minuts. Aquesta és la signatura exacta del biaix: la feina no es reparteix.
3. La campanya de MOCH-40L-AZ. En un GroupByKey per sku, tots els esdeveniments de la motxilla estrella s'envien pel shuffle a un únic treballador, perquè la clau determina el destí. Els altres no el poden ajudar.

Per què pujar a 50 treballadors no arregla res. La unitat de paral·lelisme d'una agrupació per clau és la clau, no l'element. Els esdeveniments de MOCH-40L-AZ continuaran anant tots al mateix lloc, amb 20 treballadors o amb 500. Els 30 nous estarien ociosos i facturant: el retard continuaria creixent i el cost es multiplicaria per 2,5. Escalar horitzontalment no resol un problema de distribució.

Solució A — substituir GroupByKey per CombinePerKey (la bona, si l'operació ho permet):

Per la xarxa viatgen sumes parcials, una per treballador i clau, en comptes de milions d'elements individuals. Amb 20 treballadors, el treballador de MOCH-40L-AZ rep 20 números en comptes de dotze milions d'esdeveniments. És una línia de codi i sol resoldre el problema completament.

Solució B — afegir sal a la clau (quan cal GroupByKey de debò, per exemple per conservar els elements):

Diferència pràctica entre A i B. L'A és preferible sempre que es pugui: és més simple, més barata i no té paràmetres per ajustar. La B afegeix un shuffle extra i un paràmetre (N_SAL) que cal dimensionar —massa baix no reparteix, massa alt crea sobrecàrrega—, però és l'única via quan l'operació no és associativa (per exemple, si necessites la llista completa d'esdeveniments de la sessió per reconstruir un recorregut) o quan el biaix persisteix fins i tot amb agregació parcial.

Mesura complementària immediata: abans de tocar codi, baixar --max_num_workers a 5 per deixar de pagar 15 màquines ocioses mentre es prepara el desplegament, i fer servir --update per substituir el pipeline conservant l'estat en curs, sense perdre les dades en vol.

AlpinaShop ja té canonades. En aquesta lliçó has vist per què un script en una VM és una solució que funciona fins que deixa de funcionar, i què aporta exactament un servei gestionat: paral·lelisme automàtic, reintents, escalat durant l'execució, semàntica d'escriptura fiable i observabilitat de sèrie.

Has après el model d'Apache Beam —Pipeline, PCollection immutable, PTransform, runner— i la seva idea central: un lot és un flux acotat i un stream és un flux no acotat, així que la mateixa lògica serveix per a tots dos. Coneixes les transformacions que cobreixen gairebé tots els casos i, sobretot, saps per què CombinePerKey és gairebé sempre millor que GroupByKey, i quan un ParDo amb classe DoFn guanya un Map.

Has escrit el pipeline per lots d'AlpinaShop línia a línia: llegeix les exportacions CSV d'alpinashop-catalogo, analitza, valida amb regles de negoci reals —imports amb coma, quantitats negatives, dates trencades—, escriu el detall net a lineas_pedido, calcula l'agregat de vendes per SKU amb agregació parcial, i envia tot el que és defectuós a una quarantena al bucket amb el seu motiu, sense tombar el procés. L'has provat en local amb el DirectRunner, que és lent i estricte expressament, amb un fitxer brut a consciència, i després l'has llançat a Dataflow amb el compte de servei sa-dataflow-pedidos, dins de sn-datos-euw1, sense IP pública, amb sostre de treballadors i etiquetes de facturació.

Has entrat al territori del temps, que és el que de debò distingeix el processament de dades seriós: temps de l'esdeveniment davant de temps de procés, amb el client del metro com a exemple de per què agrupar pel segon produeix números falsos; finestres fixes, lliscants i de sessió; marques d'aigua com a estimació de "ja no espero més"; activadors per emetre resultats parcials sense renunciar a la correcció posterior; i allowed_lateness com a decisió de negoci explícita sobre quant s'espera els endarrerits. I has deixat escrit el pipeline en temps real que consumirà pedidos-nuevos així que existeixi, amb timestamp_attribute —la línia que fa que tot l'anterior sigui cert— i llegint d'una subscripció, no d'un topic.

Has vist que per al que és comú hi ha drecera: les plantilles de Google resolen "Pub/Sub a BigQuery" sense escriure codi, i les plantilles flexibles empaqueten el teu propi pipeline com a imatge a Artifact Registry perquè un altre sistema l'invoqui. Coneixes l'escalat automàtic i els seus límits, Dataflow Prime i quan compensa, i saps llegir el graf, el system lag i la frescor de la dada per diagnosticar. Saps reconèixer el biaix de dades —l'escalat que no millora res— i corregir-lo amb agregació parcial, sal a la clau o entrades laterals. I coneixes la factura: per vCPU, memòria, disc i shuffle, amb l'advertiment gros que un pipeline en temps real està encès sempre i costa desenes d'euros al mes encara que no passi res.

Queda alguna cosa pendent que has notat al llarg de tota la lliçó. El pipeline per lots funciona perquè les dades ja són al bucket, i el de temps real funciona perquè algú publica a pedidos-nuevos. Però aquest topic encara no existeix, i no hem parlat de què passa amb la resta de la casa: quan entra una comanda, el magatzem l'ha de preparar, facturació ha d'emetre la factura, el client ha de rebre el seu correu de confirmació i l'analítica se n'ha d'assabentar. Avui l'aplicació Flask hauria de cridar els quatre, un darrere l'altre, i quedar-se penjada si el quart no respon.

Abans de resoldre-ho, però, hi ha una peça de l'ecosistema de dades que convé conèixer, perquè molta gent arriba a Google Cloud amb ella ja posada. A 04-03, Cloud Dataproc, veurem Spark i Hadoop gestionats: què són, per què continuen important després de vint anys, i quin és el patró que converteix un clúster car i permanent en un d'efímer que viu quatre minuts, fa la seva feina sobre les dades del bucket i s'autodestrueix. Crearem alpinashop-spark, executarem un treball PySpark que calcula quins productes es compren junts —la base del futur recomanador del mòdul 5— i compararem amb honestedat quan convé Dataproc, quan Dataflow i quan no cal cap dels dos.