La lección anterior terminó con un coordinador de cuarenta líneas que repartía trozos de pedidos.jsonl entre procesos, sumaba parciales y reencolaba la tarea de un trabajador muerto. Funcionaba en un portátil; en un clúster de mil nodos haría falta resolver, para cada trabajo, la localidad de datos, el shuffle entre nodos, la detección de fallos, la ejecución especulativa y la salida atómica. MapReduce es el modelo de programación con el que Google (Dean y Ghemawat, 2004) resolvió todo eso una sola vez, de forma que el programador escribe únicamente dos funciones, map y reduce, y el sistema se encarga del resto. Hadoop es su implementación de código abierto, junto con HDFS (04-02) y el gestor de recursos YARN, y durante una década fue sinónimo de "big data". Hoy ya casi nadie escribe trabajos MapReduce a mano, pero todo lo que vino después (Spark, Flink, los motores SQL distribuidos) usa su vocabulario y sus fases, y sus decisiones de diseño (escribir siempre en disco, tareas independientes, un maestro que reasigna) siguen siendo la referencia contra la que se explican las mejoras. Esta lección presenta el modelo con el cálculo de ventas por productor y por mercado de Kilómetro Cero, recorre la ejecución de un trabajo dentro de YARN, y lo implementa tres veces: en Python con Hadoop Streaming, en Java como ejemplo canónico, y como simulación local del shuffle para ver con las manos lo que el framework esconde.

1. El modelo de programación: map, shuffle & sort, reduce
2. Combiner y partitioner
3. Flujo de ejecución y tolerancia a fallos
4. Hadoop: HDFS, YARN y MapReduce v2
5. Anatomía de un job: de submit a _SUCCESS
6. Formatos de entrada y salida
7. Por qué MapReduce es lento y cuál es su lugar hoy
8. Práctica: ventas por productor y mercado en Streaming, en Java y en simulación
9. Errores Comunes y Consejos
10. Ejercicios
11. Conclusión

MapReduce toma prestados dos nombres de la programación funcional y les da un significado preciso para datos distribuidos. Todo trabajo procesa parejas clave/valor y pasa por tres fases:

1. Map. El sistema divide la entrada en trozos (splits) y ejecuta, para cada registro de cada trozo, la función map(k1, v1) → lista de (k2, v2). El programador decide qué es la clave intermedia k2: es la clave por la que quiere agrupar. Para "ventas por productor", map recibe una línea de pedidos.jsonl y emite, por cada línea del pedido, (productor, cantidad × precio).
2. Shuffle & sort. El sistema recoge todas las parejas (k2, v2) de todos los mappers, las envía al reducer responsable de cada k2, y las entrega agrupadas por clave y ordenadas: (k2, [v2, v2, v2, ...]). Es la fase que el programador no escribe y la que más cuesta (05-01, apartado 5).
3. Reduce. Para cada clave, reduce(k2, lista de v2) → lista de (k3, v3). Para las ventas, suma la lista y emite (productor, total).

El ejemplo con el que siempre se presenta MapReduce es el contador de palabras: map emite (palabra, 1) por cada palabra de una línea y reduce suma los unos. Es el mismo esqueleto que el nuestro cambiando "palabra" por "productor" y "1" por "importe", y por eso se dice que MapReduce es un contador de palabras generalizado: cualquier cálculo que se pueda expresar como "extraer una clave de cada registro y agregar por clave" encaja directamente; los que necesitan varias agrupaciones encadenadas (ventas por productor y, después, el productor con más ventas por mercado) se expresan como varios trabajos en cadena, cada uno leyendo la salida del anterior desde HDFS, que es el origen de la lentitud del apartado 7.

Tres propiedades del modelo explican su éxito:

• Las funciones son locales. map ve un registro; reduce ve una clave y sus valores. Ninguna necesita saber cuántos nodos hay ni dónde están los datos. El programador escribe lógica de negocio, no distribución.
• Las tareas son independientes. Cada map sobre un split y cada reduce sobre una partición de claves se pueden ejecutar en cualquier nodo, en cualquier orden, y repetir: es la reejecución determinista de 05-01.
• El shuffle es genérico. Un único mecanismo (particionar por clave, ordenar, transferir, mezclar) sirve para cualquier trabajo, y el sistema puede optimizarlo por todos.

El modelo básico tiene dos ganchos que el programador puede sustituir:

Combiner. Un mapper que procesa un bloque de 128 MB de pedidos.jsonl emite unas 400 000 parejas, pero solo hay tres productores distintos: 400 000 parejas viajarán por la red para que el reducer sume 133 000 valores por clave. El combiner es una función reduce local al mapper que se ejecuta sobre la salida de cada map antes del shuffle: agrupa las parejas del mapper por clave y las reduce a una por clave. Con él, el mapper envía 3 parejas en lugar de 400 000. Es la "prerreducción antes de mover" de 05-01. Solo es válido cuando la reducción es asociativa y conmutativa, y con el mismo tipo de entrada y salida: sumar sí; una media no, a menos que se lleve (suma, cuenta). Hadoop no garantiza que el combiner se ejecute (puede correr cero, una o varias veces sobre los mismos datos), así que el resultado debe ser idéntico con o sin él.

Partitioner. Decide a qué reducer va cada clave: particion(k2) = hash(k2) mod numero_reducers por defecto. Se sustituye cuando se quiere controlar el reparto: enviar todas las claves de un rango al mismo reducer (para obtener una salida globalmente ordenada), o separar dos tipos de clave en dos ficheros de salida. En la práctica del apartado 8 emitimos dos familias de claves en el mismo trabajo (p:<productor> y m:<mercado>) y un partitioner las envía a reducers distintos, de modo que part-r-00000 contenga las ventas por productor y part-r-00001 las ventas por mercado. Y el partitioner es también el lugar donde se ataca el sesgo: un partitioner que conozca las claves calientes puede repartirlas entre varios reducers, con una segunda pasada para combinar.

El artículo original describe una arquitectura con un maestro y muchos trabajadores, que Hadoop conserva con otros nombres (apartado 4):

1. El cliente envía el trabajo: el código (jar o scripts), la configuración y las rutas de entrada y salida.
2. El maestro pide a HDFS los bloques de la entrada y crea una tarea map por split, anotando en qué nodos vive cada bloque. Crea también R tareas reduce, con R configurado por el usuario.
3. Asigna tareas map a trabajadores libres, prefiriendo el que tiene el bloque en su disco; si no puede, uno del mismo rack; si no, cualquiera.
4. Cada map escribe su salida, particionada y ordenada, en el disco local del trabajador, e informa al maestro de dónde está.
5. Cuando todos los maps han terminado, el maestro asigna las tareas reduce; cada reducer recoge por red su partición de la salida de todos los maps, la mezcla (merge) manteniendo el orden, y ejecuta reduce clave a clave.
6. Cada reducer escribe su fichero de salida en HDFS. Cuando todos terminan, el trabajo está completo.

La tolerancia a fallos se apoya en lo que ya sabemos:

• Fallo de un trabajador. El maestro lo detecta por heartbeats perdidos. Las tareas map completadas en ese nodo se reejecutan aunque hubieran terminado, porque su salida estaba en el disco local del nodo caído y los reducers que aún no la habían recogido la necesitan. Las tareas reduce completadas no se reejecutan: su salida está en HDFS. Las tareas en curso vuelven a la cola. Exactamente lo que hacía cola_trabajo.py, con la sutileza añadida de la salida intermedia local.
• Fallo de una tarea (excepción en el código, registro corrupto). Se reintenta hasta cuatro veces (mapreduce.map.maxattempts); si sigue fallando, el trabajo falla (o, si se configura, se tolera un porcentaje de tareas fallidas para saltarse registros envenenados, la DLQ de 02-05 en versión batch).
• Rezagados. Ejecución especulativa (05-01): cuando la fase está cerca de terminar, se lanza una copia de las tareas lentas.
• Salida atómica. Cada tarea escribe en un directorio temporal (_temporary/attempt_.../) y solo cuando la tarea termina el framework hace el commit: renombra su fichero a part-r-00001 en el directorio de salida. Si dos intentos de la misma tarea (reejecución, especulación) terminan, solo el primero hace commit. Al terminar el trabajo se crea un fichero vacío _SUCCESS como señal para quien consuma la salida (el sensor de 05-05 lo esperará). Es el os.replace de 05-01 institucionalizado.
• El maestro como punto único. En el diseño original, si el maestro caía, el trabajo entero se abortaba y el cliente lo relanzaba: se aceptó como un compromiso razonable porque un maestro es una máquina entre miles y un trabajo se puede relanzar. Hadoop 2 lo mejoró creando un maestro por trabajo (el ApplicationMaster del apartado siguiente) que YARN puede reiniciar, y el ResourceManager tiene alta disponibilidad con ZooKeeper (03-03).

Hadoop es, desde la versión 2, tres proyectos apilados:

En Hadoop 1, el maestro de MapReduce (el JobTracker) hacía dos cosas a la vez: gestionar los recursos del clúster y coordinar cada trabajo. Eso lo hacía un cuello de botella (unos 4 000 nodos de límite) y ataba el clúster a MapReduce: no se podía ejecutar otra cosa. YARN separó las dos funciones:

• ResourceManager (RM). Uno por clúster (con alta disponibilidad). Conoce los recursos de cada nodo y arbitra entre aplicaciones con un scheduler (Capacity o Fair Scheduler, con colas por equipo: analitica tiene su cola con el 40 % del clúster garantizado). No sabe nada de maps ni reduces.
• NodeManager (NM). Uno por nodo. Informa al RM de su CPU y memoria disponibles, lanza y supervisa contenedores (un proceso con una cuota de CPU y memoria, hoy implementado con cgroups) y sirve la salida intermedia de los maps a los reducers (el shuffle service).
• ApplicationMaster (AM). Uno por aplicación (por trabajo MapReduce), que corre en un contenedor normal. Es el maestro del apartado 3: negocia contenedores con el RM, pide a los NM que lancen tareas en ellos, sigue su progreso y reejecuta las fallidas. Si el AM muere, el RM lo reinicia (hasta yarn.resourcemanager.am.max-attempts veces) y el nuevo AM recupera el progreso desde el registro de tareas completadas.
• Contenedor. La unidad de asignación: "1 núcleo y 2 GB en el nodo dn-07". Cada tarea map o reduce se ejecuta como una JVM dentro de un contenedor.

La consecuencia arquitectónica de YARN es que el clúster es un recurso compartido en el que conviven aplicaciones distintas (un MapReduce de analitica, un Spark del equipo de recomendaciones, un servicio de larga duración), y que el modelo de cómputo es intercambiable: cuando en 05-03 lancemos Spark sobre YARN, el driver de Spark será el ApplicationMaster y los executors correrán en contenedores, con el mismo RM arbitrando. Kubernetes juega hoy ese mismo papel de gestor de recursos genérico (07-05).

Vale la pena seguir con detalle lo que ocurre dentro de una tarea, porque los nombres reaparecen en la interfaz web de Hadoop, en los contadores del trabajo y en las explicaciones de por qué un job es lento.

Lado del map.

1. El InputFormat (apartado 6) calcula los splits: por defecto, uno por bloque HDFS, ajustado al final de línea como hacía trozos_por_bytes en 05-01. Con pedidos.jsonl de 150 MB, dos splits, dos maps.
2. El RecordReader entrega registros al map: para texto, (offset del byte, línea).
3. map emite parejas a un buffer circular en memoria (100 MB por defecto, mapreduce.task.io.sort.mb). Cuando se llena al 80 %, un hilo lo vuelca a disco (spill): particiona por reducer, ordena por clave dentro de cada partición, aplica el combiner si lo hay, y escribe un fichero de spill.
4. Al terminar el map, los ficheros de spill se mezclan en uno solo, particionado y ordenado, con un índice que dice dónde empieza cada partición. Si hubo varios spills, el combiner se vuelve a aplicar en la mezcla.

Lado del reduce.

5. Copia (fetch). En cuanto termina un map, cada reducer pide al shuffle service del NodeManager de ese map su partición, por HTTP, con varios hilos en paralelo. No espera a que terminen todos los maps para empezar a copiar (pero sí para empezar a reducir).
6. Mezcla y ordenación. Los fragmentos recibidos, cada uno ya ordenado, se mezclan (en memoria si caben, en disco en rondas si no) en una única secuencia ordenada por clave.
7. Reduce. Se recorre la secuencia; cada vez que cambia la clave, se llama a reduce(clave, iterador de valores). Por eso el reducer recibe un iterador, no una lista: los valores de una clave pueden no caber en memoria (los 133 000 importes de Quesería Montblanc sin combiner).
8. Commit. La salida va a _temporary/, y al terminar se renombra a part-r-0000N. Cuando el AM confirma que todos los reduces han hecho commit, escribe _SUCCESS.

Los contadores del trabajo resumen todo esto. Para un job real de un día de campaña (250 000 eventos, 150 MB, 2 maps, 2 reduces), con y sin combiner:

Dos lecturas: los bytes de shuffle bajan cuatro órdenes de magnitud con el combiner, y aun así el trabajo tarda 52 s para lo que ventas_scatter_gather.py hacía en 1 s. Ese minuto es el coste fijo del framework: arrancar el AM, pedir contenedores, lanzar cuatro JVM, escribir spills y mezclar, hacer commit en HDFS. Un trabajo MapReduce no tiene sentido por debajo de los gigabytes, y ese es el punto que Spark ataca.

El InputFormat decide dos cosas: cómo se divide la entrada en splits y cómo se leen los registros de cada split. El OutputFormat, cómo se escriben los resultados.

Dos advertencias prácticas. Primera: un formato es divisible solo si se puede empezar a leer desde la mitad, y eso depende también de la compresión: gzip no es divisible (un fichero de 1 GB en gzip es un único split y un único map, por grande que sea), mientras que bzip2, LZO con índice, o los formatos de contenedor (Avro, Parquet, ORC, SequenceFile) sí lo son. Segunda: HDFS y MapReduce sufren con los ficheros pequeños (04-02): 10 000 ficheros de 50 KB son 10 000 maps de un instante cada uno, y el coste de planificación domina; hay que consolidarlos (CombineFileInputFormat, o mejor, un paso previo de compactación).

La salida sigue la misma lógica: TextOutputFormat escribe clave<TAB>valor por línea en un part-r-NNNNN por reducer; con LazyOutputFormat no se crean ficheros vacíos; MultipleOutputs permite que un reducer escriba en varios ficheros con nombre (ventas por productor en productores-r-00000, por mercado en mercados-r-00000). El número de ficheros de salida es siempre el número de reducers, y esa es la razón de elegirlo con cuidado: pocos reducers hacen ficheros grandes y tareas largas; muchos, ficheros pequeños que serán un problema para el siguiente trabajo.

Las decisiones que hicieron robusto a MapReduce son las que lo hacen lento:

• Todo pasa por disco. La salida de cada map se escribe en disco local; el reducer la copia y la vuelve a escribir en disco al mezclar; la salida del reduce va a HDFS con tres réplicas. Un trabajo con un shuffle son al menos cuatro escrituras del volumen de datos intermedio. Se hizo así para que cualquier tarea se pudiera reejecutar leyendo su entrada de disco sin depender de la memoria de un nodo que puede morir.
• Los trabajos se encadenan por HDFS. Un cálculo con varias agrupaciones (ventas por productor, luego ranking por mercado, luego unir con el catálogo) son tres jobs, y entre uno y otro la salida completa va a HDFS con replicación y el siguiente la vuelve a leer. Los algoritmos iterativos (recomendaciones por descenso de gradiente, PageRank, el BSP de 05-01) son diez o cien jobs encadenados, cada uno releyendo el conjunto entero.
• Coste fijo por trabajo y por tarea. Arrancar una JVM por tarea, negociar contenedores, el commit. Decenas de segundos que en un trabajo de tres horas no importan y en una consulta interactiva lo son todo.
• Modelo rígido. Solo map y reduce; una unión (join) entre dos conjuntos hay que expresarla emitiendo ambos lados con la misma clave y distinguiéndolos en el reducer (reduce-side join), o cargando el pequeño en memoria de cada mapper (map-side join, el antecedente del broadcast join de 05-03). Nada de eso lo hace el framework por ti.

Su lugar hoy es el de base histórica y modelo mental: el vocabulario (map, shuffle, reduce, combiner, partitioner, split, contadores) es el que usan Spark y Flink; YARN y HDFS siguen en producción en muchas empresas como capa de recursos y de almacenamiento, con Spark encima; y Hive, el almacén de datos SQL sobre Hadoop que Facebook creó para no escribir MapReduce a mano, sigue existiendo pero ejecutando sus consultas sobre Tez o Spark en lugar de sobre MapReduce, conservando su catálogo de tablas (el metastore) como pieza central de muchos lagos de datos. Ver un job MapReduce nuevo en 2026 es raro; entender cómo funcionaba es lo que permite leer un plan de Spark o el panel de Flink sin sorpresas.

El trabajo es el mismo en las tres versiones: leer /km0/eventos/2026-09-14/pedidos.jsonl (el fichero que subir_eventos_hdfs.py dejó en HDFS en 04-02, o el generado por ventas_scatter_gather.py en 05-01), y producir las ventas totales por productor y por mercado. Para hacerlo en un solo job, el mapper emite dos claves por línea de pedido, con prefijo: p:queseria-montblanc y m:girona. Un partitioner envía las p: al reducer 0 y las m: al reducer 1, así que la salida queda en dos ficheros limpios.

Hadoop Streaming permite escribir mapper y reducer en cualquier lenguaje que lea de la entrada estándar y escriba en la salida estándar. Hadoop lanza el proceso, le pasa los registros línea a línea y recoge lo que emite; la clave y el valor se separan con un tabulador.

El reducer de Streaming no recibe (clave, lista de valores) como en Java, sino la secuencia ordenada de parejas: es el propio script quien detecta el cambio de clave. Esa es la razón de que el shuffle ordene y no solo agrupe: con la entrada ordenada, agrupar es comparar con la línea anterior, sin memoria. Y como el reducer suma, sirve también de combiner sin cambios.

Antes de tocar el clúster, la tubería de Unix reproduce el trabajo entero, con sort en el papel del shuffle:

(Sobre las tres líneas de ejemplo de 05-01: Ana en Girona compró 2 × 3,90 + 12,50 = 20,30; Marc en Lleida 6 × 9,80 = 58,80; Lucía en Valencia 3 × 4,20 + 4 × 1,60 = 19,00; y por productor, Huerta La Vega 7,80 + 6,40 = 14,20, Quesería Montblanc 12,50 + 12,60 = 25,10, Bodega Roble Alto 58,80.) Esta prueba local vale oro: la mayoría de los errores de un job de Streaming (un split que falla, un valor no numérico) se ven aquí en un segundo en lugar de en un minuto de job fallido.

El lanzamiento en el clúster del docker-compose.yml de 04-02 (con YARN añadido: un resourcemanager y un nodemanager de la imagen apache/hadoop):

Línea a línea: -files copia los scripts a cada contenedor (la caché distribuida: el código viaja a los datos); -combiner reutiliza el reducer localmente en cada map; -partitioner con KeyFieldBasedPartitioner y la opción -k1.1,1.1 particiona por el primer carácter de la clave (p o m), de modo que con dos reducers cada familia va a uno (con el hash de p y m módulo 2 caen en reducers distintos; si no fuera así, bastaría un partitioner propio, que en Streaming solo se puede escribir en Java). El directorio de salida no debe existir; MapReduce se niega a sobrescribir, precisamente para proteger la salida atómica. El resultado:

Y el estado del trabajo, con sus contadores, se consulta con yarn application -list -appStates ALL, mapred job -status <job_id> o en la interfaz web del ResourceManager (http://localhost:8088).

Java es el lenguaje nativo de Hadoop, y un job en Java evita el coste de lanzar un intérprete por tarea y de serializar todo como texto. Este es el ejemplo canónico, línea a línea:

Lo que hay que entender de cada bloque:

• Los tipos Writable. Hadoop no usa String ni double en las interfaces, sino Text, DoubleWritable, LongWritable: tipos serializables de forma compacta y comparables byte a byte, que el shuffle puede ordenar sin deserializar. Reutilizar las instancias (clave.set(...) en vez de new Text(...)) es la optimización más citada de MapReduce: un mapper emite millones de parejas y la creación de objetos dispararía la recolección de basura.
• Mapper<K1, V1, K2, V2> y Reducer<K2, V2, K3, V3>. Los genéricos documentan el contrato del apartado 1. Context es el canal por el que la tarea emite parejas (ctx.write) e incrementa contadores.
• El Iterable del reducer se puede recorrer una sola vez: Hadoop lo alimenta desde la secuencia ordenada en disco, y reutiliza el objeto DoubleWritable en cada iteración (guardar referencias a los valores es un error clásico: todos apuntan al mismo objeto).
• Job es la descripción declarativa: qué clases, cuántos reducers, qué tipos, qué rutas. waitForCompletion la envía al ResourceManager y bloquea hasta que termina. setJarByClass le dice a Hadoop qué jar contiene el código, que YARN copiará a cada contenedor.
• setCombinerClass(SumaReducer.class) es válido porque entrada y salida del reducer son del mismo tipo (Text, DoubleWritable) y la suma es asociativa. Si el reducer emitiera otra cosa (un ranking, una media) haría falta un combiner distinto o ninguno.

Compilación y lanzamiento:

La salida es idéntica a la de Streaming (con part-r-00000 y part-r-00001, la r indicando que la escribió un reducer) y tarda unos segundos menos por la ausencia de intérpretes de Python y de texto intermedio. En un trabajo de terabytes, esa diferencia es de decenas de minutos.

Para ver lo que el framework esconde, simulaciones/mapreduce_local.py implementa las tres fases en un solo proceso, con el shuffle explícito: particionar, ordenar por clave, agrupar.

Con el fichero de 400 000 pedidos de 05-01 y --sin-combiner, cada reducer recibe cientos de miles de parejas; con combiner, recibe 3 × n_maps para los productores y 4 × n_maps para los mercados. El groupby de itertools sobre la secuencia ordenada es literalmente lo que hace el reducer de Streaming al detectar el cambio de clave, y sorted sobre los fragmentos es el merge del apartado 5 (en Hadoop, una mezcla de secuencias ya ordenadas, más barata que un sort completo).

• Probar directamente en el clúster. Un job de Streaming se prueba con cat | mapper | sort | reducer en un segundo; en el clúster, cada intento fallido cuesta un minuto y un log de YARN que hay que ir a buscar con yarn logs -applicationId.
• Un combiner que no es asociativo. Calcular medias, o "el primero", o emitir un tipo distinto en el combiner, produce resultados que cambian según cuántas veces se haya ejecutado. Regla: el combiner debe ser una función tal que ejecutarla 0, 1 o N veces dé el mismo resultado final.
• Guardar referencias a los valores del Iterable. Hadoop reutiliza el objeto; si el reducer hace lista.add(v) para ordenar después, la lista acaba con N copias del último valor. Hay que copiar (new DoubleWritable(v.get())).
• Directorio de salida existente. El job falla antes de empezar. Es intencionado: la salida atómica exige un directorio limpio. Borrarlo forma parte del pipeline (05-05), no del job.
• gzip en la entrada. Un pedidos.jsonl.gz de 2 GB es un split y un map de veinte minutos. Usa bzip2, LZO indexado, o mejor Parquet.
• Demasiados o demasiado pocos reducers. Uno solo convierte el reduce en secuencial (Amdahl); mil sobre 20 MB de datos crean mil ficheros minúsculos. Orientación: que cada reducer procese entre 1 y 5 GB de shuffle, y nunca más reducers que claves distintas útiles.
• Ignorar los contadores. Reduce shuffle bytes y Spilled records son el termómetro del trabajo. Si el shuffle es del tamaño de la entrada, falta el combiner; si los spills son varias veces la salida del map, falta memoria en el buffer de ordenación.
• Mezclar encadenamiento de jobs con lógica. Tres jobs encadenados a mano con rutas intermedias en HDFS se convierten en un pipeline frágil. Es trabajo del planificador de 05-05, y una razón de peso para pasar a Spark, donde las tres fases son un solo programa.

Diseña un trabajo (o cadena de trabajos) MapReduce que produzca, para cada mercado, el productor con más ventas y su importe. Indica las claves y valores intermedios de cada fase, si puedes usar combiner, y cuántos jobs hacen falta. Después escribe el mapper.py y el reducer.py de Streaming del primer job.

El job del apartado 8.1 tiene 2 maps y 2 reduces. Describe qué hace el ApplicationMaster en cada uno de estos casos y qué tareas se reejecutan: (a) el NodeManager dn-07 muere cuando el map 1, que se ejecutaba allí, ya había terminado y el reduce 0 había copiado su partición pero el reduce 1 aún no; (b) el reduce 1 lanza una excepción por un valor no numérico en su tercera línea; (c) el reduce 0 tarda cinco veces la mediana y el AM lanza una copia especulativa que termina primero. ¿Qué ficheros hay en /km0/agregados/2026-09-14/ventas/ durante y después de cada caso?

Durante la Semana del Queso Artesano, p:queseria-montblanc concentra el 50 % de las parejas. Con el combiner activado, ¿es un problema? ¿Y sin combiner, o si el reduce fuera "lista de los 100 pedidos más grandes de cada productor" (donde el combiner no reduce el volumen tanto)? Propón un partitioner de Java y la lógica de un segundo job para repartir la clave caliente entre cuatro reducers y combinar después, siguiendo el salting de 05-01.

Ejercicio 1.

Hacen falta dos jobs, porque hay dos agrupaciones encadenadas: primero sumar por (mercado, productor) y después, por mercado, elegir el máximo.

• Job 1. map: por cada línea de pedido, (mercado|productor, importe). Combiner: suma (asociativa). reduce: suma. Salida: girona|queseria-montblanc<TAB>18240.50.
• Job 2. map: lee la salida del job 1 y emite (mercado, productor|total). Combiner: sí, "quedarse con el máximo" es asociativo y conmutativo y no cambia el tipo. reduce: recorrer los valores de cada mercado y emitir el productor con mayor total.

Job 1 en Streaming:

Un solo job sería posible con una clave compuesta mercado y valores productor|importe, sumando por productor dentro del reducer con un diccionario en memoria: funciona si los productores de un mercado caben en memoria (aquí sí, son tres), pero pierde el combiner y carga todo el volumen en el reduce. Con Spark será un groupBy seguido de una ventana, en un solo programa (05-03).

Ejercicio 2.

(a) El AM deja de recibir heartbeats de dn-07 y marca como perdidas sus tareas. El map 1 estaba completado, pero su salida vivía en el disco local de dn-07, y el reduce 1 aún no la había copiado, así que el AM reejecuta el map 1 en otro nodo (pidiendo un contenedor al RM, preferiblemente en un nodo con réplica del bloque). El reduce 0 conserva su copia y no se ve afectado; el reduce 1 espera y copia de la nueva ubicación. Durante ese tiempo, en el directorio de salida solo existe _temporary/ con los intentos en curso.

(b) La tarea reduce 1 falla con excepción; el AM la reintenta en otro contenedor (hasta 4 intentos por defecto). Como el error es determinista (el mismo valor corrupto), fallará las cuatro veces y el job entero falla; no se escribe _SUCCESS, y _temporary/ se limpia. Solución: que el reducer tolere el valor (try/except con contador km0,valores_no_numericos) o configurar mapreduce.reduce.failures.maxpercent. El reduce 0, que había terminado y hecho commit, deja su part-00000 en el directorio, pero sin _SUCCESS ningún consumidor debe leerlo.

(c) Las dos copias del reduce 0 escriben en _temporary/attempt_..._r_000000_0/ y _temporary/attempt_..._r_000000_1/. La copia especulativa termina primero y pide el commit; el AM lo concede y renombra su fichero a part-00000; después mata al intento original y descarta su directorio temporal. Al final: _SUCCESS, part-00000 (del intento 1) y part-00001. Nunca hay dos part-00000 porque el commit es exclusivo por tarea.

Ejercicio 3.

Con combiner y una suma, no es un problema: cada mapper reduce sus 200 000 parejas de Montblanc a una, y el reducer recibe una por mapper. El sesgo está en la entrada de los mappers, que ya está repartida por bloques, no por clave. Sin combiner, el reducer 0 recibe la mitad de las parejas del trabajo y tarda el doble que el reducer de los mercados; con un "top 100 por productor", el combiner solo reduce cada mapper a 100 parejas por productor, que sigue siendo poco, así que tampoco es grave; el sesgo importa de verdad cuando el reducer necesita todos los valores (lista completa de pedidos del productor, mediana exacta).

Partitioner de salting:

El mapper añade el sufijo #(hash(pedido_id) % 4) solo a la clave caliente (determinista: reejecutable). El job 1 produce cuatro parciales p:queseria-montblanc#0..3; un job 2 (o un paso final ligero fuera de MapReduce, porque son cuatro números) quita el sufijo y suma. Para el "top 100", el job 2 mezcla cuatro listas de 100 y se queda con las 100 mejores: correcto porque el top-K global está contenido en la unión de los top-K parciales.

MapReduce convirtió los patrones de 05-01 en un contrato de dos funciones: el programador escribe map, que extrae una clave de cada registro, y reduce, que agrega los valores de una clave, y el sistema aporta la localidad (un map por bloque, ejecutado donde vive el bloque), el shuffle & sort que agrupa por clave, la reejecución de tareas fallidas, la ejecución especulativa y la salida atómica con _temporary y _SUCCESS. El combiner es la prerreducción que ahorra cuatro órdenes de magnitud de shuffle en el job de ventas, y el partitioner es la palanca para separar familias de claves o romper una clave caliente. Hadoop lo implementa sobre HDFS con YARN como gestor de recursos genérico, con un ResourceManager que arbitra, NodeManagers que lanzan contenedores y un ApplicationMaster por trabajo que hace de maestro reiniciable. Lo hemos ejecutado tres veces: en Python con Hadoop Streaming, probado antes con cat | mapper | sort | reducer; en Java, el ejemplo canónico con sus Writable, su Job y su iterador de un solo uso; y en una simulación que hace visibles las particiones, el sort y el groupby.

También hemos visto el precio: cada fase escribe en disco, cada agrupación adicional es otro job que pasa por HDFS, cada tarea arranca una JVM, y un cálculo que un proceso de Python resuelve en un segundo tarda un minuto en el clúster. Para un lote nocturno de terabytes es un precio aceptable; para encadenar la suma por productor con la unión al catálogo y el ranking por mercado, o para entrenar las recomendaciones en cien iteraciones sobre los clics, no lo es. La siguiente lección presenta Spark, que conserva el modelo (particiones, shuffle, tareas reejecutables) pero lo expresa como un DAG de operadores que se ejecuta en memoria, con un optimizador que decide las fases: el ventas_diarias.py de analitica pasará de dos scripts y un hadoop jar a un programa de treinta líneas con DataFrames.