11  Taller: MapReduce Streaming sobre el clúster HDFS/YARN

big-data
hadoop
mapreduce
streaming
python
taller
Autor/a

Wilder Ramírez Delgado

Fecha de publicación

2 de octubre de 2026

Este taller es la práctica del capítulo anterior, MapReduce: el modelo de procesamiento de Hadoop.

El clúster (HDFS, YARN y el cliente Spark) ya está: no se vuelve a explicar. El wordcount en Java del capítulo de teoría se envió desde el NameNode; aquí el cliente es spark-client. Aquí programas mapper y reducer. Después: teoría de Spark, luego PySpark.

Qué vamos a hacer y dónde se ejecuta

Lee primero MapReduce: el modelo de procesamiento de Hadoop. Aquí programas mapper y reducer, pruebas la tubería en el cliente Hadoop y envías el trabajo con Hadoop Streaming. YARN ejecuta las tareas en los NodeManagers. El cálculo distribuido es yarn jar. No montes un clúster nuevo.

Del texto a un conteo: map, shuffle, reduce

MapReduce es un modelo para procesar un archivo grande partiéndolo en trozos. Tres fases, y cada una tiene un nombre que vas a volver a encontrar en Spark:

  1. Map. Un programa —el mapper— lee un trozo y emite pares (clave, valor). En el wordcount la clave es la palabra y el valor es 1.
  2. Shuffle. Hadoop ordena esos pares por clave y junta los que coinciden. La misma palabra termina junta, y siempre en el mismo reducer.
  3. Reduce. El reducer recibe una clave y la lista de valores que le tocó, y emite el resultado. Aquí, la suma.

Con el texto del laboratorio de YARN el recorrido se ve así:

 entrada                    MAP emite              SHUFFLE agrupa         REDUCE
 hadoop hdfs yarn           hadoop 1               hadoop → 1, 1, 1       hadoop     3
 hadoop mapreduce yarn      hdfs 1                 hdfs → 1               hdfs       1
 spark hadoop datos         yarn 1                 yarn → 1, 1, 1         yarn       3
 yarn datos datos           hadoop 1               mapreduce → 1          mapreduce  1
                            mapreduce 1            spark → 1              spark      1
                            yarn 1                 datos → 1, 1, 1        datos      3
                            spark 1
                            hadoop 1
                            datos 1
                            yarn 1
                            datos 1
                            datos 1

En Dominando la terminal para Big Data ya usaste esta tubería:

tr ' ' '\n' < texto.txt | sort | uniq -c

tr parte la línea en palabras: hace de map. sort agrupa las iguales: hace de shuffle. uniq -c suma cada grupo: hace de reduce. Hadoop Streaming es esa tubería, repartida entre los nodos del clúster y con HDFS en los dos extremos.

Hadoop Streaming y las dos clases de contenedor

Hadoop Streaming conecta programas que leen stdin y escriben stdout con un trabajo MapReduce. Nuestro cliente incluye Hadoop 3.4.1 y su JAR Streaming; los scripts usan /opt/python/bin/python3 tanto en el cliente como en los workers.

Los contenedores Docker alojan los servicios del laboratorio. Los contenedores YARN son las asignaciones de recursos donde corren las tareas dentro de los NodeManagers. Enviar un trabajo no crea nuevos contenedores Docker.

El cliente distribuye los scripts con -files, lee la entrada de HDFS y guarda allí la salida. El envío se realiza con yarn jar directamente desde spark-client.

Antes de empezar: encender el clúster y Jupyter

Este taller se ejecuta en Jupyter dentro de spark-client, con el kernel Python del clúster. El arranque es el mismo del paso 11 del Laboratorio: cliente Spark. En una terminal de WSL:

cd ~/hdfs-docker-lab
dc() { docker compose -f compose.yaml -f compose.spark.yaml "$@"; }
dc up -d --no-deps spark-client
dc exec spark-client /opt/python/bin/jupyter notebook \
  --ip=0.0.0.0 --port=8888 --no-browser --allow-root \
  --ServerApp.root_dir=/opt/spark/work-dir
Línea Qué hace
dc() { … } Atajo para usar los dos archivos Compose (el clúster y el cliente Spark) sin escribirlos cada vez.
dc up -d --no-deps spark-client Levanta solo el cliente. Supone que HDFS y YARN ya están arriba. Si el clúster está apagado, enciéndelo antes con dc up -d y espera Safe mode is OFF y dos NodeManagers RUNNING (paso 6 del laboratorio).
dc exec spark-client … jupyter notebook … Arranca Jupyter dentro del cliente. La terminal queda ocupada mientras Jupyter esté abierto.

Salida esperada al final del arranque (el token cambia cada vez):

    Or copy and paste one of these URLs:
        http://spark-client:8888/tree?token=…
        http://127.0.0.1:8888/tree?token=…

Abre en el navegador la URL de 127.0.0.1. La carpeta ~/hdfs-docker-lab/notebooks/ del host es /opt/spark/work-dir dentro del contenedor: ahí debe estar este notebook. Ábrelo y comprueba arriba a la derecha que el kernel sea Python del clúster. Al terminar, guarda con Ctrl+S y cierra Jupyter con Ctrl+C en su terminal.

1. Comprobar el entorno de Jupyter

Si esta celda falla, vuelve al Laboratorio: cliente Spark. El cliente Hadoop debe resolver el NameNode y el ResourceManager mediante los XML compartidos. El reporte de YARN debe mostrar dos NodeManagers en RUNNING. Ejecuta primero todas las comprobaciones de ese laboratorio.

from pathlib import Path
import os, sys, subprocess

assert Path('/opt/spark/work-dir').is_dir(), 'Abre el notebook dentro de spark-client'
assert str(Path(sys.executable).resolve()).startswith('/opt/python/'), 'Selecciona el kernel del clúster'
assert os.environ.get('HADOOP_CONF_DIR') == '/opt/cluster-conf'
HDFS = '/opt/hadoop/bin/hdfs'
YARN = '/opt/hadoop/bin/yarn'
STREAMING = '/opt/hadoop/share/hadoop/tools/lib/hadoop-streaming-3.4.1.jar'
assert all(Path(p).is_file() for p in (HDFS, YARN, STREAMING)), 'Falta construir el cliente completo del laboratorio'
trabajo = Path('/opt/spark/work-dir/mapreduce')
trabajo.mkdir(parents=True, exist_ok=True)

def ejecutar(*argumentos):
    proceso = subprocess.run(list(argumentos), text=True, stdout=subprocess.PIPE,
                             stderr=subprocess.STDOUT, cwd=trabajo)
    print(proceso.stdout)
    proceso.check_returncode()
    return proceso.stdout

print('Python:', sys.executable)
ejecutar(HDFS, 'getconf', '-confKey', 'fs.defaultFS')
ejecutar(YARN, 'node', '-list')

2. El mapper: emitir pares

Cada palabra se convierte en palabra<TAB>1. split() reconoce espacios y tabuladores; no normalizamos mayúsculas ni puntuación en este primer ejercicio. Guardamos los scripts en una carpeta persistente de Jupyter.

mapper = r"""#!/opt/python/bin/python3
# Mapper de wordcount: cada palabra de stdin sale como palabra<TAB>1.
import sys

for linea in sys.stdin:
    for palabra in linea.strip().split():
        if palabra:
            print(f"{palabra}\t1")
"""
(trabajo / "mapper.py").write_text(mapper, encoding="utf-8")
print(mapper)

3. El reducer: acumular una clave a la vez

Hadoop entrega pares ordenados por clave. El reducer acumula mientras la palabra no cambie y emite el total al pasar a otra. Al terminar debe emitir también la última clave. El shuffle hace que todas las apariciones de una palabra lleguen al mismo reducer.

Fijaremos un reducer en el trabajo inicial. Las ampliaciones al final permiten estudiar varios reducers sin cambiar el recorrido base.

reducer = r"""#!/opt/python/bin/python3
# Reducer de wordcount: la entrada llega ordenada por palabra. Suma y emite.
import sys

actual = None
total = 0

for linea in sys.stdin:
    linea = linea.strip()
    if not linea:
        continue
    palabra, cuenta = linea.split("\t", 1)
    cuenta = int(cuenta)
    if palabra == actual:
        total += cuenta
    else:
        if actual is not None:
            print(f"{actual}\t{total}")
        actual = palabra
        total = cuenta

if actual is not None:
    print(f"{actual}\t{total}")
"""
(trabajo / "reducer.py").write_text(reducer, encoding="utf-8")
print(reducer)

4. Probar dentro del contenedor cliente

Esta celda ejecuta los scripts con el Python del kernel y ordena los pares para simular el shuffle. Contrasta el resultado con un conteo independiente. Todavía no utiliza YARN.

from collections import Counter

texto = "hadoop hdfs yarn\nhadoop mapreduce yarn\nspark hadoop datos\nyarn datos datos\n"
mapeado = subprocess.run([sys.executable, str(trabajo / 'mapper.py')],
                         input=texto, text=True, capture_output=True, check=True)
ordenado = '\n'.join(sorted(mapeado.stdout.splitlines())) + '\n'
reducido = subprocess.run([sys.executable, str(trabajo / 'reducer.py')],
                         input=ordenado, text=True, capture_output=True, check=True)
obtenido = {p: int(n) for p, n in (fila.split('\t') for fila in reducido.stdout.splitlines())}
esperado = dict(Counter(texto.split()))
assert obtenido == esperado, obtenido
print(reducido.stdout)
print('Lógica correcta dentro de spark-client.')

5. Subir la entrada a HDFS

El archivo local que creamos está dentro de spark-client, en la carpeta compartida de notebooks. hdfs dfs -put lo incorpora al sistema distribuido. Conservamos la ruta del curso para compararlo después con PySpark. -f reemplaza únicamente este texto de práctica.

entrada_local = trabajo / 'texto.txt'
entrada_local.write_text(texto, encoding='utf-8')
ejecutar(HDFS, 'dfs', '-mkdir', '-p', '/yarn/entrada')
ejecutar(HDFS, 'dfs', '-put', '-f', str(entrada_local), '/yarn/entrada/texto.txt')
ejecutar(HDFS, 'dfs', '-cat', '/yarn/entrada/texto.txt')

6. Enviar el trabajo a YARN

Cada ejecución utiliza una carpeta de salida nueva; no necesitamos borrar resultados anteriores. -files distribuye los dos scripts y la ruta absoluta de Python corresponde al volumen preparado en los NodeManagers.

La celda espera a que termine el trabajo y muestra sus logs completos. En el portal http://localhost:8088, busca wordcount-python de tipo MAPREDUCE. No debe haber otra sesión Spark ocupando los recursos del laboratorio; ciérrala antes con spark.stop().

from datetime import datetime, timezone
from uuid import uuid4

marca = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') + '-' + uuid4().hex[:6]
salida_hdfs = '/curso/mapreduce/salidas/wordcount-' + marca
ejecutar(YARN, 'jar', STREAMING,
         '-D', 'mapreduce.job.name=wordcount-python',
         '-D', 'mapreduce.job.reduces=1',
         '-files', f'{trabajo / "mapper.py"},{trabajo / "reducer.py"}',
         '-mapper', '/opt/python/bin/python3 -u mapper.py',
         '-reducer', '/opt/python/bin/python3 -u reducer.py',
         '-input', '/yarn/entrada/texto.txt', '-output', salida_hdfs)
print('Resultado:', salida_hdfs)

7. Comprobar la salida distribuida

La carpeta debe contener _SUCCESS y un archivo part-00000. Comparamos el resultado con el esperado y guardamos su ruta en la carpeta de trabajo. Los contadores y la aplicación de YARN demuestran que no fue solo una ejecución local.

ejecutar(HDFS, 'dfs', '-test', '-e', salida_hdfs + '/_SUCCESS')
ejecutar(HDFS, 'dfs', '-ls', salida_hdfs)
respuesta = subprocess.run([HDFS, 'dfs', '-cat', salida_hdfs + '/part-*'],
                           text=True, capture_output=True, check=True)
distribuido = {p: int(n) for p, n in (fila.split('\t') for fila in respuesta.stdout.splitlines())}
assert distribuido == esperado, distribuido
assert sum(distribuido.values()) == 12
print(distribuido)
(trabajo / 'ultima_salida_hdfs.txt').write_text(salida_hdfs, encoding='utf-8')
ejecutar(YARN, 'application', '-list', '-appStates', 'ALL')

8. Prácticas de MapReduce

Estas cuatro prácticas se realizan después del wordcount anterior. Reserva entre 90 y 120 minutos adicionales. Usa el mismo kernel de Jupyter dentro de spark-client: las pruebas pequeñas corren en ese contenedor y los trabajos MapReduce se ejecutan en YARN.

Para cada ejercicio: completa los TODO de la cadena que contiene el programa, ejecuta la celda que guarda los archivos, pasa las pruebas y envía el trabajo al clúster. Las plantillas lanzan NotImplementedError hasta completarlas. Al editar una plantilla debes volver a ejecutar su celda para guardar el cambio. Las soluciones de referencia están al final y no se ejecutan automáticamente.

Herramientas comunes

Ejecuta primero las secciones anteriores, que definen trabajo, ejecutar, HDFS, YARN y STREAMING. Cada envío usa una ruta nueva en HDFS para conservar las ejecuciones anteriores. El ordenamiento de la prueba en el cliente representa el agrupamiento por clave; YARN realiza el shuffle real.

import ast, json, math, re
from datetime import datetime, timezone
from uuid import uuid4
practicas = trabajo / 'practicas'
practicas.mkdir(exist_ok=True)

def guardar_practica(nombre, entrada, mapper, reducer):
    assert re.fullmatch(r'p[1-4]_[a-z_]+', nombre)
    carpeta = practicas / nombre
    carpeta.mkdir(exist_ok=True)
    for archivo, contenido in [('mapper.py', mapper), ('reducer.py', reducer)]:
        ast.parse(contenido)
        (carpeta / archivo).write_text(contenido, encoding='utf-8')
    (carpeta / 'entrada.txt').write_text(entrada, encoding='utf-8')
    return carpeta

def leer_pares(texto, convertir=int):
    pares = {}
    for linea in texto.splitlines():
        if not linea.strip():
            continue
        clave, valor = linea.split('\t') if '\t' in linea else linea.split(chr(9))
        assert clave not in pares, f'Clave duplicada: {clave}'
        pares[clave] = convertir(valor)
    return pares

def probar_en_cliente(carpeta, entrada=None, convertir=int):
    if entrada is None:
        entrada = (carpeta / 'entrada.txt').read_text(encoding='utf-8')
    def correr(archivo, texto):
        r = subprocess.run([sys.executable, str(carpeta / archivo)], input=texto,
                           text=True, capture_output=True)
        if r.returncode:
            print(r.stderr)
        r.check_returncode()
        return r.stdout
    mapeado = correr('mapper.py', entrada)
    filas = sorted(mapeado.splitlines(), key=lambda x: x.split(chr(9), 1)[0])
    ordenado = ''.join(fila + chr(10) for fila in filas)
    return leer_pares(correr('reducer.py', ordenado), convertir)

def enviar_a_yarn(carpeta, reducers=1):
    assert isinstance(reducers, int) and reducers > 0
    marca = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') + '-' + uuid4().hex[:8]
    base = f'/curso/mapreduce/practicas/{carpeta.name}/{marca}'
    entrada = base + '/entrada/entrada.txt'
    salida = base + '/salida'
    ejecutar(HDFS, 'dfs', '-mkdir', '-p', base + '/entrada')
    ejecutar(HDFS, 'dfs', '-put', str(carpeta / 'entrada.txt'), entrada)
    ejecutar(YARN, 'jar', STREAMING,
             '-D', f'mapreduce.job.name={carpeta.name}-r{reducers}',
             '-D', f'mapreduce.job.reduces={reducers}',
             '-files', f'{carpeta / "mapper.py"},{carpeta / "reducer.py"}',
             '-mapper', '/opt/python/bin/python3 -u mapper.py',
             '-reducer', '/opt/python/bin/python3 -u reducer.py',
             '-input', entrada, '-output', salida)
    ejecutar(HDFS, 'dfs', '-test', '-e', salida + '/_SUCCESS')
    (carpeta / 'ultima_ejecucion.json').write_text(
        json.dumps({'entrada': entrada, 'salida': salida, 'reducers': reducers}, indent=2),
        encoding='utf-8')
    return salida

def verificar_hdfs(salida, esperado, convertir=int):
    obtenido = leer_pares(ejecutar(HDFS, 'dfs', '-cat', salida + '/part-*'), convertir)
    assert obtenido.keys() == esperado.keys(), obtenido
    assert all(math.isclose(obtenido[k], esperado[k], rel_tol=1e-6, abs_tol=1e-6)
               for k in esperado), obtenido
    print('Resultado verificado:', obtenido)
    return obtenido
reducer_suma = r"""#!/opt/python/bin/python3
# Reducer de wordcount: la entrada llega ordenada por palabra. Suma y emite.
import sys

actual = None
total = 0

for linea in sys.stdin:
    linea = linea.strip()
    if not linea:
        continue
    palabra, cuenta = linea.split("\t", 1)
    cuenta = int(cuenta)
    if palabra == actual:
        total += cuenta
    else:
        if actual is not None:
            print(f"{actual}\t{total}")
        actual = palabra
        total = cuenta

if actual is not None:
    print(f"{actual}\t{total}")
"""

Práctica 1. Contar únicamente palabras de cuatro o más letras

Problema: eliminar palabras cortas antes de contar. La entrada incluye las cuatro líneas del ejemplo y una línea adicional: de la y el en un.

Tu tarea: completa conservar. Reutiliza el reducer de suma. El mapper debe filtrar antes de emitir los pares.

Resultado esperado: datos=3, hadoop=3, hdfs=1, mapreduce=1, spark=1 y yarn=3. Tanto yarn como hdfs tienen cuatro letras y permanecen. De 18 apariciones se conservan 12.

entrada_p1 = """hadoop hdfs yarn
hadoop mapreduce yarn
spark hadoop datos
yarn datos datos
de la y el en un
"""
mapper_p1 = r"""import sys

def conservar(palabra):
    # TODO: devolver True si tiene cuatro o más letras.
    raise NotImplementedError('Completa el filtro')

for linea in sys.stdin:
    for palabra in linea.split():
        if conservar(palabra):
            print(f"{palabra}\t1")
"""
p1 = guardar_practica('p1_filtro', entrada_p1, mapper_p1, reducer_suma)
esperado_p1 = {'datos': 3, 'hadoop': 3, 'hdfs': 1, 'mapreduce': 1, 'spark': 1, 'yarn': 3}
print(entrada_p1)

Prueba antes de enviar: completa la plantilla y vuelve a ejecutar la celda anterior. Esta prueba usa el Python del contenedor de Jupyter.

assert probar_en_cliente(p1) == esperado_p1
assert probar_en_cliente(p1, '') == {}
assert probar_en_cliente(p1, 'sol yarn hdfs de casa casa') == {'yarn':1, 'hdfs':1, 'casa':2}
print('Pruebas del cliente superadas')

Ejecución distribuida: ejecuta la siguiente celda después de superar las pruebas. Anota el identificador application_… que aparece en el registro y consulta el trabajo en el portal de YARN.

salida_p1 = enviar_a_yarn(p1)
resultado_p1 = verificar_hdfs(salida_p1, esperado_p1)

Explica: ¿por qué filtrar en el mapper reduce los pares enviados al shuffle? ¿Cuántos pares emite este mapper para la entrada de la práctica?

Práctica 2. Sumar unidades vendidas por producto

Problema: calcular unidades vendidas, no contar filas. La entrada CSV tiene encabezado y tres registros: teclado,1; teclado,3; raton,2.

Tu tarea: completa emitir para producir producto<TAB>unidades. El mapper ya omite el encabezado y convierte las unidades a entero. Reutiliza el reducer de suma.

Resultado esperado: teclado=4 y raton=2. Hay tres registros de venta y seis unidades. Emitir siempre 1 daría un resultado incorrecto.

entrada_p2 = """producto,unidades
teclado,1
teclado,3
raton,2
"""
mapper_p2 = """import sys, csv

def emitir(producto, unidades):
    # TODO: imprimir producto y unidades separados por una tabulación.
    raise NotImplementedError('Completa la emisión')

for fila in csv.reader(sys.stdin):
    if not fila or fila[0].strip() == 'producto':
        continue
    emitir(fila[0].strip(), int(fila[1]))
"""
p2 = guardar_practica('p2_ventas', entrada_p2, mapper_p2, reducer_suma)
esperado_p2 = {'teclado': 4, 'raton': 2}
print(entrada_p2)

Prueba antes de enviar: completa la plantilla y vuelve a ejecutar la celda anterior. Esta prueba usa el Python del contenedor de Jupyter.

assert probar_en_cliente(p2) == esperado_p2
assert probar_en_cliente(p2, 'producto,unidades\n') == {}
assert probar_en_cliente(p2, 'producto,unidades\nraton,10\nraton,0\nteclado,2\n') == {'raton':10, 'teclado':2}
print('Pruebas del cliente superadas')

Ejecución distribuida: ejecuta la siguiente celda después de superar las pruebas. Anota el identificador application_… que aparece en el registro y consulta el trabajo en el portal de YARN.

salida_p2 = enviar_a_yarn(p2)
resultado_p2 = verificar_hdfs(salida_p2, esperado_p2)

Explica: ¿qué diferencia hay entre contar ventas y sumar unidades? ¿Qué cambiaría si también quisieras calcular ingresos?

Práctica 3. Calcular la temperatura media por ciudad

Problema: cada ciudad tiene una cantidad diferente de observaciones.

Tu tarea: completa el mapper para emitir ciudad<TAB>temperatura,1. En el reducer acumula suma y cantidad, y emite su cociente al cerrar cada grupo, incluido el último.

Ciudad Suma Observaciones Promedio
Bogota 90 3 30
Medellin 40 2 20
Cali 124 4 31

No configures este reducer de promedio final como combiner. Promediar promedios pierde los pesos: para Bogotá, promediar 20 (de 10 y 30) con 50 da 35, pero el promedio correcto es 30. Una agregación intermedia tendría que conservar (suma, cantidad).

entrada_p3 = """ciudad,temperatura
Bogota,10
Medellin,18
Cali,28
Bogota,30
Cali,32
Medellin,22
Bogota,50
Cali,30
Cali,34
"""
mapper_p3 = """import sys, csv

def emitir(ciudad, temperatura):
    # TODO: emitir ciudad<TAB>temperatura,1
    raise NotImplementedError('Completa la emisión')

for fila in csv.reader(sys.stdin):
    if not fila or fila[0].strip() == 'ciudad':
        continue
    emitir(fila[0].strip(), float(fila[1]))
"""
reducer_p3 = r"""import sys

def promedio(ciudad, suma, cantidad):
    # TODO: emitir ciudad<TAB>promedio, con seis decimales.
    raise NotImplementedError('Completa el promedio')

actual = None
suma = 0.0
cantidad = 0
for linea in sys.stdin:
    if not linea.strip():
        continue
    ciudad, par = linea.strip().split("\t")
    subtotal, n = par.split(',')
    if actual is not None and ciudad != actual:
        promedio(actual, suma, cantidad)
        suma, cantidad = 0.0, 0
    actual = ciudad
    # TODO: acumular float(subtotal) e int(n).
    raise NotImplementedError('Completa la acumulación')
if actual is not None:
    promedio(actual, suma, cantidad)
"""
p3 = guardar_practica('p3_promedios', entrada_p3, mapper_p3, reducer_p3)
esperado_p3 = {'Bogota': 30.0, 'Medellin': 20.0, 'Cali': 31.0}
print(entrada_p3)

Prueba antes de enviar: completa la plantilla y vuelve a ejecutar la celda anterior. Esta prueba usa el Python del contenedor de Jupyter.

assert probar_en_cliente(p3, convertir=float) == esperado_p3
assert probar_en_cliente(p3, 'ciudad,temperatura\n', float) == {}
assert probar_en_cliente(p3, 'ciudad,temperatura\nBogota,12.5\n', float) == {'Bogota':12.5}
assert probar_en_cliente(p3, 'ciudad,temperatura\nBogota,10\nBogota,30\nBogota,50\n', float) == {'Bogota':30.0}
print('Pruebas del cliente superadas')

Ejecución distribuida: ejecuta la siguiente celda después de superar las pruebas. Anota el identificador application_… que aparece en el registro y consulta el trabajo en el portal de YARN.

salida_p3 = enviar_a_yarn(p3)
resultado_p3 = verificar_hdfs(salida_p3, esperado_p3, convertir=float)

Explica: ¿por qué hay que conservar la cantidad de observaciones? ¿Qué ocurriría si olvidaras emitir el último grupo?

Práctica 4. Ejecutar las ventas con dos reducers

Problema: observar cómo se reparte la salida sin cambiar el resultado del cálculo.

Tu tarea: primero resuelve la práctica 2. Reutiliza exactamente su entrada y sus programas, y completa reducers_p4 con el número 2. Ejecuta el trabajo y compara el resultado con la ejecución de un reducer.

El particionador decide el destino de cada clave. Una clave debe aparecer en un único archivo de salida. Dos reducers producen dos archivos part-*, pero alguno puede estar vacío: no se garantiza repartir estas dos claves entre ambos reducers. Con tres filas no se puede concluir que aumentar reducers mejora el rendimiento.

reducers_p4 = None  # TODO: solicitar dos reducers.
assert reducers_p4 == 2, 'Completa reducers_p4 con 2'
assert probar_en_cliente(p2) == esperado_p2
p4 = guardar_practica('p4_dos_reducers', entrada_p2,
                      (p2 / 'mapper.py').read_text(encoding='utf-8'),
                      (p2 / 'reducer.py').read_text(encoding='utf-8'))
salida_p4 = enviar_a_yarn(p4, reducers=reducers_p4)
resultado_p4 = verificar_hdfs(salida_p4, esperado_p2)
assert resultado_p4 == resultado_p2
listado = ejecutar(HDFS, 'dfs', '-ls', salida_p4 + '/part-*')
archivos = [linea.split()[-1] for linea in listado.splitlines() if linea.startswith('-')]
assert len(archivos) == 2, archivos
reunido = {}
for archivo in archivos:
    print('Archivo:', archivo)
    parte = leer_pares(ejecutar(HDFS, 'dfs', '-cat', archivo))
    assert reunido.keys().isdisjoint(parte.keys()), 'Una clave aparece en dos archivos'
    reunido.update(parte)
assert reunido == esperado_p2
print('Dos archivos inspeccionados; resultado completo verificado')

Explica: ¿qué claves quedaron en cada archivo? ¿Aparece alguna clave en ambos? ¿Por qué leer un solo part-* puede dar un resultado incompleto?

Entrega de las cuatro prácticas

Entrega las plantillas completadas, las pruebas aprobadas, el identificador de cada aplicación YARN y su ruta de salida HDFS. Cada carpeta de práctica guarda ultima_ejecucion.json con las rutas del último envío.

Práctica Evidencia del resultado
1. Filtro Seis claves y 12 apariciones; palabras cortas ausentes
2. Ventas teclado=4, raton=2
3. Promedios Bogota=30, Medellin=20, Cali=31
4. Dos reducers Dos archivos inspeccionados y mismo resultado que la práctica 2

Responde también las preguntas de cada práctica. Superar las pruebas del cliente no demuestra que YARN haya ejecutado el trabajo: verifica _SUCCESS, la salida y el estado final de la aplicación en el portal.

9. Soluciones de referencia

Consulta esta sección después de intentar resolver los ejercicios. Sustituye la cadena del programa correspondiente en la celda de la práctica y vuelve a ejecutarla para guardar el archivo. Estas soluciones no se ejecutan automáticamente.

Práctica 1: mapper

import sys

def conservar(palabra):
    # TODO: devolver True si tiene cuatro o más letras.
    return len(palabra) >= 4

for linea in sys.stdin:
    for palabra in linea.split():
        if conservar(palabra):
            print(f"{palabra}\t1")

Práctica 2: mapper

import sys, csv

def emitir(producto, unidades):
    # TODO: imprimir producto y unidades separados por una tabulación.
    print(f"{producto}\t{unidades}")

for fila in csv.reader(sys.stdin):
    if not fila or fila[0].strip() == 'producto':
        continue
    emitir(fila[0].strip(), int(fila[1]))

Práctica 3: mapper

import sys, csv

def emitir(ciudad, temperatura):
    # TODO: emitir ciudad<TAB>temperatura,1
    print(f"{ciudad}\t{temperatura},1")

for fila in csv.reader(sys.stdin):
    if not fila or fila[0].strip() == 'ciudad':
        continue
    emitir(fila[0].strip(), float(fila[1]))

Práctica 3: reducer

import sys

def promedio(ciudad, suma, cantidad):
    # TODO: emitir ciudad<TAB>promedio, con seis decimales.
    print(f"{ciudad}\t{suma / cantidad:.6f}")

actual = None
suma = 0.0
cantidad = 0
for linea in sys.stdin:
    if not linea.strip():
        continue
    ciudad, par = linea.strip().split("\t")
    subtotal, n = par.split(',')
    if actual is not None and ciudad != actual:
        promedio(actual, suma, cantidad)
        suma, cantidad = 0.0, 0
    actual = ciudad
    # TODO: acumular float(subtotal) e int(n).
    suma += float(subtotal)
    cantidad += int(n)
if actual is not None:
    promedio(actual, suma, cantidad)

Práctica 4: configuración

En la celda correspondiente escribe reducers_p4 = 2. Reutiliza los programas resueltos de la práctica 2. La comparación se hace sobre la unión de los archivos de salida, sin exigir qué producto debe quedar en cada archivo.

Siguiente: teoría de Spark (Spark: el motor de procesamiento sobre YARN) y su taller (Taller de Spark). Después, PySpark. El clúster no se vuelve a explicar.