La lliçó anterior va acabar amb un coordinador de quaranta línies que repartia trossos de comandes.jsonl entre processos, sumava parcials i tornava a encuar la tasca d'un treballador mort. Funcionava en un portàtil; en un clúster de mil nodes caldria resoldre, per a cada treball, la localitat de dades, el shuffle entre nodes, la detecció de fallades, l'execució especulativa i la sortida atòmica. MapReduce és el model de programació amb què Google (Dean i Ghemawat, 2004) va resoldre tot això una sola vegada, de manera que el programador escriu únicament dues funcions, map i reduce, i el sistema s'encarrega de la resta. Hadoop n'és la implementació de codi obert, juntament amb HDFS (04-02) i el gestor de recursos YARN, i durant una dècada va ser sinònim de "big data". Avui ja gairebé ningú no escriu treballs MapReduce a mà, però tot el que va venir després (Spark, Flink, els motors SQL distribuïts) fa servir el seu vocabulari i les seves fases, i les seves decisions de disseny (escriure sempre a disc, tasques independents, un mestre que reassigna) continuen sent la referència contra la qual s'expliquen les millores. Aquesta lliçó presenta el model amb el càlcul de vendes per productor i per mercat de Quilòmetre Zero, recorre l'execució d'un treball dins de YARN, i l'implementa tres vegades: en Python amb Hadoop Streaming, en Java com a exemple canònic, i com a simulació local del shuffle per veure amb les mans el que el framework amaga.

1. El model de programació: map, shuffle & sort, reduce
2. Combiner i partitioner
3. Flux d'execució i tolerància a fallades
4. Hadoop: HDFS, YARN i MapReduce v2
5. Anatomia d'un job: de submit a _SUCCESS
6. Formats d'entrada i sortida
7. Per què MapReduce és lent i quin és el seu lloc avui
8. Pràctica: vendes per productor i mercat en Streaming, en Java i en simulació
9. Errors Comuns i Consells
10. Exercicis
11. Conclusió

MapReduce manlleva dos noms de la programació funcional i els dona un significat precís per a dades distribuïdes. Tot treball processa parelles clau/valor i passa per tres fases:

1. Map. El sistema divideix l'entrada en trossos (splits) i executa, per a cada registre de cada tros, la funció map(k1, v1) → llista de (k2, v2). El programador decideix què és la clau intermèdia k2: és la clau per la qual vol agrupar. Per a "vendes per productor", map rep una línia de comandes.jsonl i emet, per cada línia de la comanda, (productor, quantitat × preu).
2. Shuffle & sort. El sistema recull totes les parelles (k2, v2) de tots els mappers, les envia al reducer responsable de cada k2, i les lliura agrupades per clau i ordenades: (k2, [v2, v2, v2, ...]). És la fase que el programador no escriu i la que més costa (05-01, apartat 5).
3. Reduce. Per a cada clau, reduce(k2, llista de v2) → llista de (k3, v3). Per a les vendes, suma la llista i emet (productor, total).

L'exemple amb què sempre es presenta MapReduce és el comptador de paraules: map emet (paraula, 1) per cada paraula d'una línia i reduce suma els uns. És el mateix esquelet que el nostre canviant "paraula" per "productor" i "1" per "import", i per això es diu que MapReduce és un comptador de paraules generalitzat: qualsevol càlcul que es pugui expressar com "extreure una clau de cada registre i agregar per clau" hi encaixa directament; els que necessiten diverses agrupacions encadenades (vendes per productor i, després, el productor amb més vendes per mercat) s'expressen com a diversos treballs en cadena, cadascun llegint la sortida de l'anterior des d'HDFS, que és l'origen de la lentitud de l'apartat 7.

Tres propietats del model n'expliquen l'èxit:

• Les funcions són locals. map veu un registre; reduce veu una clau i els seus valors. Cap de les dues no necessita saber quants nodes hi ha ni on són les dades. El programador escriu lògica de negoci, no distribució.
• Les tasques són independents. Cada map sobre un split i cada reduce sobre una partició de claus es poden executar en qualsevol node, en qualsevol ordre, i repetir: és la reexecució determinista de 05-01.
• El shuffle és genèric. Un únic mecanisme (particionar per clau, ordenar, transferir, barrejar) serveix per a qualsevol treball, i el sistema el pot optimitzar per a tots.

El model bàsic té dos ganxos que el programador pot substituir:

Combiner. Un mapper que processa un bloc de 128 MB de comandes.jsonl emet unes 400 000 parelles, però només hi ha tres productors diferents: 400 000 parelles viatjaran per la xarxa perquè el reducer sumi 133 000 valors per clau. El combiner és una funció reduce local al mapper que s'executa sobre la sortida de cada map abans del shuffle: agrupa les parelles del mapper per clau i les redueix a una per clau. Amb ell, el mapper envia 3 parelles en lloc de 400 000. És la "prereducció abans de moure" de 05-01. Només és vàlid quan la reducció és associativa i commutativa, i amb el mateix tipus d'entrada i sortida: sumar sí; una mitjana no, tret que es porti (suma, compte). Hadoop no garanteix que el combiner s'executi (pot córrer zero, una o diverses vegades sobre les mateixes dades), així que el resultat ha de ser idèntic amb ell o sense.

Partitioner. Decideix a quin reducer va cada clau: particio(k2) = hash(k2) mod nombre_reducers per defecte. Se substitueix quan es vol controlar el repartiment: enviar totes les claus d'un rang al mateix reducer (per obtenir una sortida globalment ordenada), o separar dos tipus de clau en dos fitxers de sortida. A la pràctica de l'apartat 8 emetem dues famílies de claus en el mateix treball (p:<productor> i m:<mercat>) i un partitioner les envia a reducers diferents, de manera que part-r-00000 contingui les vendes per productor i part-r-00001 les vendes per mercat. I el partitioner és també el lloc on s'ataca el biaix: un partitioner que conegui les claus calentes pot repartir-les entre diversos reducers, amb una segona passada per combinar.

L'article original descriu una arquitectura amb un mestre i molts treballadors, que Hadoop conserva amb altres noms (apartat 4):

1. El client envia el treball: el codi (jar o scripts), la configuració i les rutes d'entrada i sortida.
2. El mestre demana a HDFS els blocs de l'entrada i crea una tasca map per split, anotant en quins nodes viu cada bloc. Crea també R tasques reduce, amb R configurat per l'usuari.
3. Assigna tasques map a treballadors lliures, preferint el que té el bloc al seu disc; si no pot, un del mateix rack; si no, qualsevol.
4. Cada map escriu la seva sortida, particionada i ordenada, al disc local del treballador, i informa el mestre d'on és.
5. Quan tots els maps han acabat, el mestre assigna les tasques reduce; cada reducer recull per xarxa la seva partició de la sortida de tots els maps, la barreja (merge) mantenint l'ordre, i executa reduce clau a clau.
6. Cada reducer escriu el seu fitxer de sortida a HDFS. Quan tots acaben, el treball està complet.

La tolerància a fallades es recolza en el que ja sabem:

• Fallada d'un treballador. El mestre la detecta per heartbeats perduts. Les tasques map completades en aquell node es reexecuten encara que haguessin acabat, perquè la seva sortida era al disc local del node caigut i els reducers que encara no l'havien recollit la necessiten. Les tasques reduce completades no es reexecuten: la seva sortida és a HDFS. Les tasques en curs tornen a la cua. Exactament el que feia cua_treball.py, amb la subtilesa afegida de la sortida intermèdia local.
• Fallada d'una tasca (excepció al codi, registre corrupte). Es reintenta fins a quatre vegades (mapreduce.map.maxattempts); si continua fallant, el treball falla (o, si es configura, es tolera un percentatge de tasques fallides per saltar-se registres enverinats, la DLQ de 02-05 en versió batch).
• Ressagats. Execució especulativa (05-01): quan la fase és a prop d'acabar, es llança una còpia de les tasques lentes.
• Sortida atòmica. Cada tasca escriu en un directori temporal (_temporary/attempt_.../) i només quan la tasca acaba el framework fa el commit: reanomena el seu fitxer a part-r-00001 al directori de sortida. Si dos intents de la mateixa tasca (reexecució, especulació) acaben, només el primer fa commit. En acabar el treball es crea un fitxer buit _SUCCESS com a senyal per a qui consumeixi la sortida (el sensor de 05-05 l'esperarà). És l'os.replace de 05-01 institucionalitzat.
• El mestre com a punt únic. En el disseny original, si el mestre queia, el treball sencer s'avortava i el client el rellançava: es va acceptar com un compromís raonable perquè un mestre és una màquina entre milers i un treball es pot rellançar. Hadoop 2 ho va millorar creant un mestre per treball (l'ApplicationMaster de l'apartat següent) que YARN pot reiniciar, i el ResourceManager té alta disponibilitat amb ZooKeeper (03-03).

Hadoop és, des de la versió 2, tres projectes apilats:

A Hadoop 1, el mestre de MapReduce (el JobTracker) feia dues coses alhora: gestionar els recursos del clúster i coordinar cada treball. Això el feia un coll d'ampolla (uns 4 000 nodes de límit) i lligava el clúster a MapReduce: no s'hi podia executar res més. YARN va separar les dues funcions:

• ResourceManager (RM). Un per clúster (amb alta disponibilitat). Coneix els recursos de cada node i arbitra entre aplicacions amb un scheduler (Capacity o Fair Scheduler, amb cues per equip: analitica té la seva cua amb el 40 % del clúster garantit). No sap res de maps ni de reduces.
• NodeManager (NM). Un per node. Informa l'RM de la seva CPU i memòria disponibles, llança i supervisa contenidors (un procés amb una quota de CPU i memòria, avui implementat amb cgroups) i serveix la sortida intermèdia dels maps als reducers (el shuffle service).
• ApplicationMaster (AM). Un per aplicació (per treball MapReduce), que corre en un contenidor normal. És el mestre de l'apartat 3: negocia contenidors amb l'RM, demana als NM que hi llancin tasques, en segueix el progrés i reexecuta les fallides. Si l'AM mor, l'RM el reinicia (fins a yarn.resourcemanager.am.max-attempts vegades) i el nou AM recupera el progrés des del registre de tasques completades.
• Contenidor. La unitat d'assignació: "1 nucli i 2 GB al node dn-07". Cada tasca map o reduce s'executa com una JVM dins d'un contenidor.

La conseqüència arquitectònica de YARN és que el clúster és un recurs compartit en què conviuen aplicacions diferents (un MapReduce d'analitica, un Spark de l'equip de recomanacions, un servei de llarga durada), i que el model de còmput és intercanviable: quan a 05-03 llancem Spark sobre YARN, el driver de Spark serà l'ApplicationMaster i els executors correran en contenidors, amb el mateix RM arbitrant. Kubernetes juga avui aquest mateix paper de gestor de recursos genèric (07-05).

Val la pena seguir amb detall el que passa dins d'una tasca, perquè els noms reapareixen a la interfície web de Hadoop, als comptadors del treball i a les explicacions de per què un job és lent.

Costat del map.

1. L'InputFormat (apartat 6) calcula els splits: per defecte, un per bloc HDFS, ajustat al final de línia com feia trossos_per_bytes a 05-01. Amb comandes.jsonl de 150 MB, dos splits, dos maps.
2. El RecordReader lliura registres al map: per a text, (offset del byte, línia).
3. map emet parelles a un buffer circular en memòria (100 MB per defecte, mapreduce.task.io.sort.mb). Quan s'omple al 80 %, un fil el buida a disc (spill): particiona per reducer, ordena per clau dins de cada partició, aplica el combiner si n'hi ha, i escriu un fitxer d'spill.
4. En acabar el map, els fitxers d'spill es barregen en un de sol, particionat i ordenat, amb un índex que diu on comença cada partició. Si hi va haver diversos spills, el combiner es torna a aplicar en la barreja.

Costat del reduce.

5. Còpia (fetch). Tan bon punt acaba un map, cada reducer demana a l'shuffle service del NodeManager d'aquell map la seva partició, per HTTP, amb diversos fils en paral·lel. No espera que acabin tots els maps per començar a copiar (però sí per començar a reduir).
6. Barreja i ordenació. Els fragments rebuts, cadascun ja ordenat, es barregen (en memòria si hi caben, a disc en rondes si no) en una única seqüència ordenada per clau.
7. Reduce. Es recorre la seqüència; cada vegada que canvia la clau, es crida reduce(clau, iterador de valors). Per això el reducer rep un iterador, no una llista: els valors d'una clau poden no cabre en memòria (els 133 000 imports de Formatgeria Montblanc sense combiner).
8. Commit. La sortida va a _temporary/, i en acabar es reanomena a part-r-0000N. Quan l'AM confirma que tots els reduces han fet commit, escriu _SUCCESS.

Els comptadors del treball resumeixen tot això. Per a un job real d'un dia de campanya (250 000 esdeveniments, 150 MB, 2 maps, 2 reduces), amb combiner i sense:

Dues lectures: els bytes de shuffle baixen quatre ordres de magnitud amb el combiner, i tot i així el treball triga 52 s per al que vendes_scatter_gather.py feia en 1 s. Aquest minut és el cost fix del framework: arrencar l'AM, demanar contenidors, llançar quatre JVM, escriure spills i barrejar, fer commit a HDFS. Un treball MapReduce no té sentit per sota dels gigabytes, i aquest és el punt que Spark ataca.

L'InputFormat decideix dues coses: com es divideix l'entrada en splits i com es llegeixen els registres de cada split. L'OutputFormat, com s'escriuen els resultats.

Dos advertiments pràctics. Primer: un format és divisible només si es pot començar a llegir des de la meitat, i això depèn també de la compressió: gzip no és divisible (un fitxer d'1 GB en gzip és un únic split i un únic map, per gran que sigui), mentre que bzip2, LZO amb índex, o els formats de contenidor (Avro, Parquet, ORC, SequenceFile) sí que ho són. Segon: HDFS i MapReduce pateixen amb els fitxers petits (04-02): 10 000 fitxers de 50 KB són 10 000 maps d'un instant cadascun, i el cost de planificació domina; cal consolidar-los (CombineFileInputFormat, o millor, un pas previ de compactació).

La sortida segueix la mateixa lògica: TextOutputFormat escriu clau<TAB>valor per línia en un part-r-NNNNN per reducer; amb LazyOutputFormat no es creen fitxers buits; MultipleOutputs permet que un reducer escrigui en diversos fitxers amb nom (vendes per productor a productors-r-00000, per mercat a mercats-r-00000). El nombre de fitxers de sortida és sempre el nombre de reducers, i aquesta és la raó de triar-lo amb cura: pocs reducers fan fitxers grans i tasques llargues; molts, fitxers petits que seran un problema per al treball següent.

Les decisions que van fer robust MapReduce són les que el fan lent:

• Tot passa per disc. La sortida de cada map s'escriu a disc local; el reducer la copia i la torna a escriure a disc en barrejar-la; la sortida del reduce va a HDFS amb tres rèpliques. Un treball amb un shuffle són com a mínim quatre escriptures del volum de dades intermedi. Es va fer així perquè qualsevol tasca es pogués reexecutar llegint la seva entrada de disc sense dependre de la memòria d'un node que pot morir.
• Els treballs s'encadenen per HDFS. Un càlcul amb diverses agrupacions (vendes per productor, després rànquing per mercat, després unir amb el catàleg) són tres jobs, i entre l'un i l'altre la sortida completa va a HDFS amb replicació i el següent la torna a llegir. Els algorismes iteratius (recomanacions per descens de gradient, PageRank, el BSP de 05-01) són deu o cent jobs encadenats, cadascun rellegint el conjunt sencer.
• Cost fix per treball i per tasca. Arrencar una JVM per tasca, negociar contenidors, el commit. Desenes de segons que en un treball de tres hores no importen i en una consulta interactiva ho són tot.
• Model rígid. Només map i reduce; una unió (join) entre dos conjunts cal expressar-la emetent tots dos costats amb la mateixa clau i distingint-los al reducer (reduce-side join), o carregant el petit en memòria de cada mapper (map-side join, l'antecedent del broadcast join de 05-03). Res d'això no ho fa el framework per tu.

El seu lloc avui és el de base històrica i model mental: el vocabulari (map, shuffle, reduce, combiner, partitioner, split, comptadors) és el que fan servir Spark i Flink; YARN i HDFS continuen en producció a moltes empreses com a capa de recursos i d'emmagatzematge, amb Spark a sobre; i Hive, el magatzem de dades SQL sobre Hadoop que Facebook va crear per no escriure MapReduce a mà, continua existint però executant les seves consultes sobre Tez o Spark en lloc de sobre MapReduce, conservant el seu catàleg de taules (el metastore) com a peça central de molts llacs de dades. Veure un job MapReduce nou el 2026 és estrany; entendre com funcionava és el que permet llegir un pla de Spark o el panell de Flink sense sorpreses.

El treball és el mateix en les tres versions: llegir /km0/esdeveniments/2026-09-14/comandes.jsonl (el fitxer que pujar_esdeveniments_hdfs.py va deixar a HDFS a 04-02, o el generat per vendes_scatter_gather.py a 05-01), i produir les vendes totals per productor i per mercat. Per fer-ho en un sol job, el mapper emet dues claus per línia de comanda, amb prefix: p:formatgeria-montblanc i m:girona. Un partitioner envia les p: al reducer 0 i les m: al reducer 1, així que la sortida queda en dos fitxers nets.

Hadoop Streaming permet escriure mapper i reducer en qualsevol llenguatge que llegeixi de l'entrada estàndard i escrigui a la sortida estàndard. Hadoop llança el procés, li passa els registres línia a línia i recull el que emet; la clau i el valor se separen amb un tabulador.

El reducer de Streaming no rep (clau, llista de valors) com en Java, sinó la seqüència ordenada de parelles: és el mateix script qui detecta el canvi de clau. Aquesta és la raó que el shuffle ordeni i no només agrupi: amb l'entrada ordenada, agrupar és comparar amb la línia anterior, sense memòria. I com que el reducer suma, serveix també de combiner sense canvis.

Abans de tocar el clúster, la canonada d'Unix reprodueix el treball sencer, amb sort en el paper del shuffle:

(Sobre les tres línies d'exemple de 05-01: l'Anna a Girona va comprar 2 × 3,90 + 12,50 = 20,30; en Marc a Lleida 6 × 9,80 = 58,80; la Llúcia a València 3 × 4,20 + 4 × 1,60 = 19,00; i per productor, Horta La Vega 7,80 + 6,40 = 14,20, Formatgeria Montblanc 12,50 + 12,60 = 25,10, Celler Roure Alt 58,80.) Aquesta prova local val or: la majoria dels errors d'un job de Streaming (un split que falla, un valor no numèric) es veuen aquí en un segon en lloc d'en un minut de job fallit.

El llançament al clúster del docker-compose.yml de 04-02 (amb YARN afegit: un resourcemanager i un nodemanager de la imatge apache/hadoop):

Línia a línia: -files copia els scripts a cada contenidor (la memòria cau distribuïda: el codi viatja a les dades); -combiner reutilitza el reducer localment a cada map; -partitioner amb KeyFieldBasedPartitioner i l'opció -k1.1,1.1 particiona pel primer caràcter de la clau (p o m), de manera que amb dos reducers cada família va a un (amb el hash de p i m mòdul 2 cauen en reducers diferents; si no fos així, n'hi hauria prou amb un partitioner propi, que a Streaming només es pot escriure en Java). El directori de sortida no ha d'existir; MapReduce es nega a sobreescriure, precisament per protegir la sortida atòmica. El resultat:

I l'estat del treball, amb els seus comptadors, es consulta amb yarn application -list -appStates ALL, mapred job -status <job_id> o a la interfície web del ResourceManager (http://localhost:8088).

Java és el llenguatge natiu de Hadoop, i un job en Java evita el cost de llançar un intèrpret per tasca i de serialitzar-ho tot com a text. Aquest és l'exemple canònic, línia a línia:

El que cal entendre de cada bloc:

• Els tipus Writable. Hadoop no fa servir String ni double a les interfícies, sinó Text, DoubleWritable, LongWritable: tipus serialitzables de manera compacta i comparables byte a byte, que el shuffle pot ordenar sense deserialitzar. Reutilitzar les instàncies (clau.set(...) en comptes de new Text(...)) és l'optimització més citada de MapReduce: un mapper emet milions de parelles i la creació d'objectes dispararia la recollida d'escombraries.
• Mapper<K1, V1, K2, V2> i Reducer<K2, V2, K3, V3>. Els genèrics documenten el contracte de l'apartat 1. Context és el canal pel qual la tasca emet parelles (ctx.write) i incrementa comptadors.
• L'Iterable del reducer es pot recórrer una sola vegada: Hadoop l'alimenta des de la seqüència ordenada a disc, i reutilitza l'objecte DoubleWritable a cada iteració (desar referències als valors és un error clàssic: tots apunten al mateix objecte).
• Job és la descripció declarativa: quines classes, quants reducers, quins tipus, quines rutes. waitForCompletion l'envia al ResourceManager i bloqueja fins que acaba. setJarByClass diu a Hadoop quin jar conté el codi, que YARN copiarà a cada contenidor.
• setCombinerClass(SumaReducer.class) és vàlid perquè entrada i sortida del reducer són del mateix tipus (Text, DoubleWritable) i la suma és associativa. Si el reducer emetés una altra cosa (un rànquing, una mitjana) caldria un combiner diferent o cap.

Compilació i llançament:

La sortida és idèntica a la de Streaming (amb part-r-00000 i part-r-00001, la r indicant que la va escriure un reducer) i triga uns segons menys per l'absència d'intèrprets de Python i de text intermedi. En un treball de terabytes, aquesta diferència és de desenes de minuts.

Per veure el que el framework amaga, simulacions/mapreduce_local.py implementa les tres fases en un sol procés, amb el shuffle explícit: particionar, ordenar per clau, agrupar.

Amb el fitxer de 400 000 comandes de 05-01 i --sense-combiner, cada reducer rep centenars de milers de parelles; amb combiner, en rep 3 × n_maps per als productors i 4 × n_maps per als mercats. El groupby d'itertools sobre la seqüència ordenada és literalment el que fa el reducer de Streaming en detectar el canvi de clau, i sorted sobre els fragments és el merge de l'apartat 5 (a Hadoop, una barreja de seqüències ja ordenades, més barata que un sort complet).

• Provar directament al clúster. Un job de Streaming es prova amb cat | mapper | sort | reducer en un segon; al clúster, cada intent fallit costa un minut i un log de YARN que cal anar a buscar amb yarn logs -applicationId.
• Un combiner que no és associatiu. Calcular mitjanes, o "el primer", o emetre un tipus diferent al combiner, produeix resultats que canvien segons quantes vegades s'hagi executat. Regla: el combiner ha de ser una funció tal que executar-la 0, 1 o N vegades doni el mateix resultat final.
• Desar referències als valors de l'Iterable. Hadoop reutilitza l'objecte; si el reducer fa llista.add(v) per ordenar després, la llista acaba amb N còpies de l'últim valor. Cal copiar (new DoubleWritable(v.get())).
• Directori de sortida existent. El job falla abans de començar. És intencionat: la sortida atòmica exigeix un directori net. Esborrar-lo forma part del pipeline (05-05), no del job.
• gzip a l'entrada. Un comandes.jsonl.gz de 2 GB és un split i un map de vint minuts. Fes servir bzip2, LZO indexat, o millor Parquet.
• Massa reducers o massa pocs. Un de sol converteix el reduce en seqüencial (Amdahl); mil sobre 20 MB de dades creen mil fitxers minúsculs. Orientació: que cada reducer processi entre 1 i 5 GB de shuffle, i mai més reducers que claus diferents útils.
• Ignorar els comptadors. Reduce shuffle bytes i Spilled records són el termòmetre del treball. Si el shuffle és de la mida de l'entrada, falta el combiner; si els spills són diverses vegades la sortida del map, falta memòria al buffer d'ordenació.
• Barrejar encadenament de jobs amb lògica. Tres jobs encadenats a mà amb rutes intermèdies a HDFS es converteixen en un pipeline fràgil. És feina del planificador de 05-05, i una raó de pes per passar a Spark, on les tres fases són un sol programa.

Dissenya un treball (o cadena de treballs) MapReduce que produeixi, per a cada mercat, el productor amb més vendes i el seu import. Indica les claus i valors intermedis de cada fase, si pots fer servir combiner, i quants jobs calen. Després escriu el mapper.py i el reducer.py de Streaming del primer job.

El job de l'apartat 8.1 té 2 maps i 2 reduces. Descriu què fa l'ApplicationMaster en cadascun d'aquests casos i quines tasques es reexecuten: (a) el NodeManager dn-07 mor quan el map 1, que s'hi executava, ja havia acabat i el reduce 0 havia copiat la seva partició però el reduce 1 encara no; (b) el reduce 1 llança una excepció per un valor no numèric a la seva tercera línia; (c) el reduce 0 triga cinc vegades la mediana i l'AM llança una còpia especulativa que acaba primer. Quins fitxers hi ha a /km0/agregats/2026-09-14/vendes/ durant i després de cada cas?

Durant la Setmana del Formatge Artesà, p:formatgeria-montblanc concentra el 50 % de les parelles. Amb el combiner activat, és un problema? I sense combiner, o si el reduce fos "llista de les 100 comandes més grans de cada productor" (on el combiner no redueix tant el volum)? Proposa un partitioner de Java i la lògica d'un segon job per repartir la clau calenta entre quatre reducers i combinar després, seguint el salting de 05-01.

Exercici 1.

Calen dos jobs, perquè hi ha dues agrupacions encadenades: primer sumar per (mercat, productor) i després, per mercat, triar el màxim.

• Job 1. map: per cada línia de comanda, (mercat|productor, import). Combiner: suma (associativa). reduce: suma. Sortida: girona|formatgeria-montblanc<TAB>18240.50.
• Job 2. map: llegeix la sortida del job 1 i emet (mercat, productor|total). Combiner: sí, "quedar-se amb el màxim" és associatiu i commutatiu i no canvia el tipus. reduce: recórrer els valors de cada mercat i emetre el productor amb el total més gran.

Job 1 en Streaming:

Un sol job seria possible amb una clau composta mercat i valors productor|import, sumant per productor dins del reducer amb un diccionari en memòria: funciona si els productors d'un mercat caben en memòria (aquí sí, són tres), però perd el combiner i carrega tot el volum al reduce. Amb Spark serà un groupBy seguit d'una finestra, en un sol programa (05-03).

Exercici 2.

(a) L'AM deixa de rebre heartbeats de dn-07 i marca com a perdudes les seves tasques. El map 1 estava completat, però la seva sortida vivia al disc local de dn-07, i el reduce 1 encara no l'havia copiat, així que l'AM reexecuta el map 1 en un altre node (demanant un contenidor a l'RM, preferiblement en un node amb rèplica del bloc). El reduce 0 conserva la seva còpia i no se'n veu afectat; el reduce 1 espera i copia de la nova ubicació. Durant aquest temps, al directori de sortida només existeix _temporary/ amb els intents en curs.

(b) La tasca reduce 1 falla amb excepció; l'AM la reintenta en un altre contenidor (fins a 4 intents per defecte). Com que l'error és determinista (el mateix valor corrupte), fallarà les quatre vegades i el job sencer falla; no s'escriu _SUCCESS, i _temporary/ es neteja. Solució: que el reducer toleri el valor (try/except amb comptador km0,valors_no_numerics) o configurar mapreduce.reduce.failures.maxpercent. El reduce 0, que havia acabat i fet commit, deixa el seu part-00000 al directori, però sense _SUCCESS cap consumidor no l'ha de llegir.

(c) Les dues còpies del reduce 0 escriuen a _temporary/attempt_..._r_000000_0/ i _temporary/attempt_..._r_000000_1/. La còpia especulativa acaba primer i demana el commit; l'AM el concedeix i reanomena el seu fitxer a part-00000; després mata l'intent original i descarta el seu directori temporal. Al final: _SUCCESS, part-00000 (de l'intent 1) i part-00001. Mai no hi ha dos part-00000 perquè el commit és exclusiu per tasca.

Exercici 3.

Amb combiner i una suma, no és un problema: cada mapper redueix les seves 200 000 parelles de Montblanc a una, i el reducer en rep una per mapper. El biaix és a l'entrada dels mappers, que ja està repartida per blocs, no per clau. Sense combiner, el reducer 0 rep la meitat de les parelles del treball i triga el doble que el reducer dels mercats; amb un "top 100 per productor", el combiner només redueix cada mapper a 100 parelles per productor, que continua sent poc, així que tampoc no és greu; el biaix importa de debò quan el reducer necessita tots els valors (llista completa de comandes del productor, mediana exacta).

Partitioner de salting:

El mapper afegeix el sufix #(hash(comanda_id) % 4) només a la clau calenta (determinista: reexecutable). El job 1 produeix quatre parcials p:formatgeria-montblanc#0..3; un job 2 (o un pas final lleuger fora de MapReduce, perquè són quatre números) treu el sufix i suma. Per al "top 100", el job 2 barreja quatre llistes de 100 i es queda amb les 100 millors: correcte perquè el top-K global està contingut en la unió dels top-K parcials.

MapReduce va convertir els patrons de 05-01 en un contracte de dues funcions: el programador escriu map, que extreu una clau de cada registre, i reduce, que agrega els valors d'una clau, i el sistema aporta la localitat (un map per bloc, executat on viu el bloc), el shuffle & sort que agrupa per clau, la reexecució de tasques fallides, l'execució especulativa i la sortida atòmica amb _temporary i _SUCCESS. El combiner és la prereducció que estalvia quatre ordres de magnitud de shuffle al job de vendes, i el partitioner és la palanca per separar famílies de claus o trencar una clau calenta. Hadoop ho implementa sobre HDFS amb YARN com a gestor de recursos genèric, amb un ResourceManager que arbitra, NodeManagers que llancen contenidors i un ApplicationMaster per treball que fa de mestre reiniciable. Ho hem executat tres vegades: en Python amb Hadoop Streaming, provat abans amb cat | mapper | sort | reducer; en Java, l'exemple canònic amb els seus Writable, el seu Job i el seu iterador d'un sol ús; i en una simulació que fa visibles les particions, el sort i el groupby.

També n'hem vist el preu: cada fase escriu a disc, cada agrupació addicional és un altre job que passa per HDFS, cada tasca arrenca una JVM, i un càlcul que un procés de Python resol en un segon triga un minut al clúster. Per a un lot nocturn de terabytes és un preu acceptable; per encadenar la suma per productor amb la unió al catàleg i el rànquing per mercat, o per entrenar les recomanacions en cent iteracions sobre els clics, no ho és. La lliçó següent presenta Spark, que conserva el model (particions, shuffle, tasques reexecutables) però l'expressa com un DAG d'operadors que s'executa en memòria, amb un optimitzador que decideix les fases: el vendes_diaries.py d'analitica passarà de dos scripts i un hadoop jar a un programa de trenta línies amb DataFrames.