15 Taller: fundamentos de PySpark sobre el clúster YARN

big-data

spark

pyspark

taller

Autor/a

Wilder Ramírez Delgado

Fecha de publicación

6 de octubre de 2026

Este taller es la práctica del capítulo anterior, PySpark: la API de Spark en Python. Léelo antes: aquí ejecutas sus ejemplos sobre el clúster.

Antes: clúster (HDFS, YARN, cliente Spark), MapReduce y la teoría de Spark. Aquí no se vuelve a montar HDFS ni el cliente.

El orden del libro: clúster (HDFS, YARN, cliente Spark) → MapReduce → Spark → PySpark. Este taller cierra ese recorrido.

Al terminar: SparkSession en YARN, transformaciones frente a acciones, leer HDFS, DataFrames y el wordcount de las cuatro líneas, escrito de vuelta a HDFS.

Cómo está armado este taller

El motor (DAG, driver, executors, tipo SPARK) está en Spark: el motor de procesamiento sobre YARN. PySpark (SparkSession, DataFrame, pereza) está en PySpark: la API de Spark en Python. Aquí programas.

HDFS guarda (hdfs://namenode:9000). YARN reparte (http://localhost:8088). El driver vive en spark-client; los executors son contenedores YARN. Si no aparece la aplicación en :8088, la sesión no nació. Si la ruta no se lee, el problema es HDFS.

Antes de ejecutar la primera celda

Completa el Laboratorio: cliente Spark y selecciona Python del clúster. Deja preparados: spark-client, dos NodeManagers, Python 3.11 y /yarn/entrada/texto.txt. El CSV de ventas lo subimos aquí.

Ejecuta en orden. Si reinicias el kernel, recrea la sesión. YARN: http://localhost:8088. UI Spark: http://localhost:4040 mientras la sesión viva.

from pathlib import Path
import sys, os

assert str(Path(sys.executable).resolve()).startswith('/opt/python/'), 'Abre Jupyter en spark-client'
assert os.environ.get('SPARK_CONF_DIR') == '/opt/cluster-conf'
assert Path('/opt/hadoop/bin/hdfs').is_file(), 'Falta la imagen de cliente del laboratorio'
print('Kernel dentro del contenedor:', sys.executable)

La SparkSession: el punto de entrada

La SparkSession da acceso a DataFrames, SQL y al contexto de los RDD. La celda utiliza la configuración del laboratorio en spark-defaults.conf: YARN, memoria de los executors, dirección del driver y Python compartido. Aquí solo damos nombre a la aplicación.

Crear la sesión inicia la aplicación; las acciones posteriores solicitan cálculos. Comprueba que master es yarn y busca pyspark-01-fundamentos en el ResourceManager. Reutilizaremos esta sesión hasta el cierre del taller.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("pyspark-01-fundamentos").getOrCreate()
spark.sparkContext.setLogLevel("WARN")
assert spark.sparkContext.master == "yarn", "Revisa el kernel y spark-defaults.conf"
print("master:", spark.sparkContext.master)
print("appId:", spark.sparkContext.applicationId)
print("version:", spark.version)
print("ui:", spark.sparkContext.uiWebUrl)

RDD y DataFrame, en miniatura

Para qué: sentir las dos APIs (detalle en PySpark: la API de Spark en Python). El resto usa DataFrame. collect con cuatro palabras es correcto; con un archivo grande no.

palabras = spark.sparkContext.parallelize(
    ["hadoop", "yarn", "spark", "hadoop"],
    numSlices=2,
)
conteo_rdd = palabras.map(lambda palabra: (palabra, 1)).reduceByKey(lambda a, b: a + b)
print(sorted(conteo_rdd.collect()))

collect trae el resultado al driver. Con cuatro palabras es inofensivo. Con un archivo grande llenarías la memoria de spark-client: por eso, en adelante, preferimos show, que enseña unas filas y deja el resto en el clúster.

cursos = spark.createDataFrame(
    [("hadoop", 3), ("yarn", 3), ("spark", 1)],
    ["palabra", "n"],
)
cursos.printSchema()
cursos.show(truncate=False)

Pereza: filter no calcula hasta count

Para qué: ver el plan antes del número. filter registra; count dispara el job. Debes ver filas con hadoop: 2.

from pyspark.sql.functions import col

lineas = spark.read.text("hdfs://namenode:9000/yarn/entrada/texto.txt")
filtradas = lineas.filter(col("value").contains("hadoop"))
print("transformacion registrada; el calculo todavia no corre")
filtradas.explain(mode="simple")
print("filas con hadoop:", filtradas.count())

Leer un CSV desde HDFS

Para qué: esquema (nombres y tipos) en lugar de partir strings. La primera celda crea y sube ventas.csv. Debes ver producto string, unidades entero y tres filas.

from pathlib import Path
import subprocess

csv_local = "/tmp/ventas.csv"
Path(csv_local).write_text("producto,unidades\nteclado,1\nteclado,3\nraton,2\n", encoding="utf-8")
subprocess.run(["/opt/hadoop/bin/hdfs", "dfs", "-mkdir", "-p", "/pyspark/entrada"], check=True)
subprocess.run(
    ["/opt/hadoop/bin/hdfs", "dfs", "-put", "-f", csv_local, "/pyspark/entrada/ventas.csv"],
    check=True,
)

ventas = (
    spark.read
    .option("header", True)
    .option("inferSchema", True)
    .csv("hdfs://namenode:9000/pyspark/entrada/ventas.csv")
)
ventas.printSchema()
ventas.show(truncate=False)

select, filter, groupBy

Para qué: las tres operaciones que más se repiten. count() aquí cuenta filas, no suma unidades. Debes ver raton 1 fila, teclado 1 fila (la fila teclado,1 cae con unidades >= 2).

from pyspark.sql.functions import col

(
    ventas.select("producto", "unidades")
    .filter(col("unidades") >= 2)
    .groupBy("producto")
    .count()
    .orderBy("producto")
    .show(truncate=False)
)

Wordcount con DataFrames

Para qué: el mismo texto de MapReduce, sin mapper/reducer a mano. explode es el map; groupBy().count() es shuffle + reduce. Debes ver: datos 3, hadoop 3, hdfs 1, mapreduce 1, spark 1, yarn 3.

from pyspark.sql.functions import col, explode, split

conteo = (
    lineas.select(explode(split(col("value"), r"\s+")).alias("palabra"))
    .where(col("palabra") != "")
    .groupBy("palabra")
    .count()
    .orderBy("palabra")
)
conteo.show(truncate=False)

Escribir en HDFS

Para qué: un show desaparece al cerrar la sesión. write es una acción. coalesce(1) deja un solo archivo en este ejemplo; en datos grandes no lo uses. Debes ver _SUCCESS, un part-00000-… y el mismo conteo al releer.

import subprocess

(
    conteo.coalesce(1).write.mode("overwrite").option("header", True)
    .csv("hdfs://namenode:9000/yarn/salida_pyspark_wc")
)
subprocess.run(["/opt/hadoop/bin/hdfs", "dfs", "-ls", "/yarn/salida_pyspark_wc"], check=True)
leido = spark.read.option("header", True).csv("hdfs://namenode:9000/yarn/salida_pyspark_wc")
leido.show(truncate=False)

esperado = {"datos": 3, "hadoop": 3, "hdfs": 1, "mapreduce": 1, "spark": 1, "yarn": 3}
# Seis filas conocidas: collect se utiliza solo para verificar este ejemplo pequeño.
obtenido = {fila["palabra"]: fila["count"] for fila in conteo.collect()}
assert obtenido == esperado, obtenido
assert sum(obtenido.values()) == 12
print("Conteo verificado: 6 palabras distintas, 12 apariciones.")

Práctica breve

Muestra productos con unidades > 2. Predice cuántas filas y comprueba. Luego compara contar filas por producto con sumar unidades.

Esperado: el filtro deja una fila (teclado, 3). La suma es teclado 4, ratón 2.

from pyspark.sql.functions import sum as spark_sum

ventas.filter(col("unidades") > 2).show()
ventas.groupBy("producto").agg(spark_sum("unidades").alias("total")).show()

Cerrar

Para qué: devolver memoria a YARN. En :8088 la app pasa a FINISHED / SUCCEEDED.

spark.stop()
print("sesion cerrada")

Referencias

Teoría: PySpark: la API de Spark en Python.
PySpark 3.5.3.
Spark SQL, DataFrames.