lección 3

Instalar PySpark: levantar un cluster Spark con Docker

Docker Compose con Spark master + worker + Jupyter. Tu laboratorio de procesamiento distribuido en 5 minutos.

⏱ 40 min

## Tu laboratorio personal de Big Data

Hasta ahora hemos hablado de clusters de 10, 50, 100 nodos. Pero para aprender no necesitas una empresa con 200 servidores. Docker te permite simular un cluster Spark completo en tu portátil: un nodo master que coordina, workers que procesan, y un notebook Jupyter donde escribes tu código PySpark. Todo encapsulado en contenedores que puedes levantar y destruir en segundos.

Este docker-compose.yml es tu entorno de desarrollo para TODA esta skill. Lo levantarás al inicio de cada sesión de práctica y lo pararás cuando termines. Los datasets de ejemplo los montamos como volúmenes para que persistan entre sesiones.

Requisito previo: Docker debe estar instalado y funcionando. Si seguiste la Skill 6 (Docker), ya lo tienes. Si no, instala Docker Desktop desde docker.com — es gratuito para uso educativo.

## Paso 1: Crear la estructura de directorios

Crea una carpeta para todo tu trabajo con PySpark. Dentro tendrás el docker-compose, tus notebooks y los datasets de ejemplo:

1# Crear la estructura del proyecto
2mkdir -p pyspark-lab/notebooks
3mkdir -p pyspark-lab/data
4mkdir -p pyspark-lab/output
5cd pyspark-lab
6
7# Verificar que Docker está funcionando
8docker --version
9docker compose version

Estructura básica: notebooks para tu código, data para datasets, output para resultados

## Paso 2: El docker-compose.yml

Este archivo define tres servicios: spark-master (el coordinador), spark-worker (el que procesa datos) y jupyter (tu interfaz para escribir código). El master y el worker forman el cluster Spark. Jupyter se conecta al master para enviar jobs.

1# docker-compose.yml
2
3services:
4  spark-master:
5    image: bitnami/spark:3.5
6    container_name: spark-master
7    environment:
8      - SPARK_MODE=master
9      - SPARK_MASTER_HOST=spark-master
10      - SPARK_MASTER_PORT=7077
11    ports:
12      - "8080:8080"   # Spark UI del master
13      - "7077:7077"   # Puerto del cluster
14    volumes:
15      # MISMA ruta que en jupyter: en un cluster el fichero lo abre el executor,
16      # no tu notebook, y la ruta tiene que existir igual en los dos sitios
17      - ./data:/home/jovyan/data
18      - ./output:/home/jovyan/output
19
20  spark-worker:
21    image: bitnami/spark:3.5
22    environment:
23      - SPARK_MODE=worker
24      - SPARK_MASTER_URL=spark://spark-master:7077
25      - SPARK_WORKER_MEMORY=2G
26      - SPARK_WORKER_CORES=2
27    depends_on:
28      - spark-master
29    volumes:
30      # el worker es quien lee y escribe de verdad: misma ruta, sin excepción
31      - ./data:/home/jovyan/data
32      - ./output:/home/jovyan/output
33
34  jupyter:
35    image: quay.io/jupyter/pyspark-notebook:spark-3.5.0
36    container_name: pyspark-jupyter
37    environment:
38      - SPARK_MASTER=spark://spark-master:7077
39    ports:
40      - "8888:8888"   # Jupyter Notebook
41      - "4040:4040"   # Spark UI del driver (para diagnóstico)
42    volumes:
43      - ./notebooks:/home/jovyan/work
44      - ./data:/home/jovyan/data
45      - ./output:/home/jovyan/output
46    depends_on:
47      - spark-master
48      - spark-worker

docker-compose.yml: un cluster Spark completo con Jupyter incluido

Analicemos qué hace cada servicio:

spark-master: el coordinador del cluster. Escucha en el puerto 7077 para que los workers se conecten. La UI web está en el puerto 8080.
spark-worker: un nodo trabajador con 2 GB de RAM y 2 cores asignados. Procesa las particiones de datos que el master le asigne.
jupyter: un notebook con PySpark preinstalado. Se conecta al master para enviar jobs al cluster. Tu interfaz de trabajo diaria.

Fíjate en que ./data se monta en la MISMA ruta (/home/jovyan/data) en los tres servicios. Esto es lo que más quebraderos de cabeza da en un cluster: cuando escribes spark.read.csv("/home/jovyan/data/ventas.csv"), el fichero no lo abre tu notebook, lo abren los executors, que viven dentro del contenedor del worker. Si ahí la ruta no existe, el job no falla al leer la línea: falla más tarde, dentro de las tasks, con un FileNotFoundException que parece no tener nada que ver. Regla: en un cluster, toda ruta local que le pases a Spark tiene que existir igual en todos los nodos.

Si tienes menos de 8 GB de RAM en tu máquina, reduce SPARK_WORKER_MEMORY a 1G. Docker + los tres contenedores necesitan al menos 4-5 GB para funcionar sin problemas. Si tu máquina es potente (16+ GB), puedes añadir un segundo worker copiando el servicio spark-worker.

## Paso 3: Levantar el cluster

1# Levantar todos los servicios en background
2docker compose up -d
3
4# Ver el estado de los contenedores
5docker compose ps
6
7# Deberías ver los tres servicios con STATUS "Up". Algo así:
8# NAME                        STATUS         PORTS
9# spark-master                Up 2 minutes   0.0.0.0:7077->7077, 0.0.0.0:8080->8080
10# pyspark-lab-spark-worker-1  Up 2 minutes
11# pyspark-jupyter             Up 2 minutes   0.0.0.0:4040->4040, 0.0.0.0:8888->8888
12#
13# El worker lleva el prefijo de la carpeta del proyecto (pyspark-lab-) y un -1
14# al final porque no le hemos puesto container_name: es lo que permite escalarlo
15# a varios workers más adelante.
16
17# Ver los logs del master para confirmar que el worker se conectó
18docker compose logs spark-master | grep -i "registered"
19# Deberías ver: "Registered worker ... with 2 cores, 2048.0 MiB RAM"

Levantar y verificar que el cluster funciona

## Paso 4: Verificar la Spark UI

Abre tu navegador y ve a http://localhost:8080. Deberías ver la Spark Master UI con información sobre el cluster: workers conectados, memoria disponible, cores totales. Esta UI será tu mejor amiga para diagnosticar problemas (la exploraremos en profundidad en la lección 8).

## Paso 5: Tu primer programa PySpark

Abre Jupyter en http://localhost:8888 (puede pedir un token — míralo en los logs con docker compose logs jupyter). Crea el notebook DENTRO de la carpeta work: es la única que está montada contra tu máquina, así que es la única donde el notebook sobrevive a un docker compose down. Ahí escribe tu primer programa Spark:

1from pyspark.sql import SparkSession
2
3# Crear la sesión Spark conectada al cluster
4spark = (
5    SparkSession.builder
6    .appName("mi-primer-spark")
7    .master("spark://spark-master:7077")
8    .config("spark.executor.memory", "1g")
9    .getOrCreate()
10)
11
12# Verificar la conexión
13print(f"Spark version: {spark.version}")
14print(f"App name: {spark.sparkContext.appName}")
15print(f"Master: {spark.sparkContext.master}")
16
17# Crear un DataFrame simple para probar
18datos = [
19    ("Ana", "Madrid", 28),
20    ("Carlos", "Barcelona", 34),
21    ("Elena", "Sevilla", 25),
22    ("David", "Valencia", 31),
23]
24
25df = spark.createDataFrame(datos, ["nombre", "ciudad", "edad"])
26df.show()
27# +------+---------+----+
28# |nombre|   ciudad|edad|
29# +------+---------+----+
30# |   Ana|   Madrid|  28|
31# |Carlos|Barcelona|  34|
32# | Elena|  Sevilla|  25|
33# | David| Valencia|  31|
34# +------+---------+----+
35
36print(f"Número de particiones: {df.rdd.getNumPartitions()}")
37print("[OK] ¡Spark funciona correctamente!")

Tu primer programa PySpark: conexión al cluster y DataFrame básico

Si obtienes un error de conexión al master, espera 30 segundos y reintenta. El master tarda un poco en estar listo. También verifica con docker compose ps que todos los contenedores están en estado "running".

## Paso 6: Crear los datasets de práctica

Para las siguientes lecciones necesitamos datos realistas. Vamos a generar un dataset de ventas de un e-commerce ficticio que usaremos en todos los ejercicios:

1import csv
2import random
3from datetime import datetime, timedelta
4
5# Generar dataset de ventas (1 millón de filas para que sea interesante)
6random.seed(42)
7
8productos = ["Laptop", "Monitor", "Teclado", "Mouse", "Auriculares",
9             "Webcam", "SSD", "RAM", "Cable USB", "Hub USB"]
10# Pesos: los periféricos baratos se venden mucho más que los portátiles
11pesos_producto = [2, 5, 15, 20, 8, 3, 5, 5, 27, 10]
12
13ciudades = ["Madrid", "Barcelona", "Valencia", "Sevilla", "Bilbao",
14            "Málaga", "Zaragoza", "Murcia", "Palma", "Las Palmas"]
15# Madrid y Barcelona concentran la mitad del mercado
16pesos_ciudad = [30, 18, 10, 8, 7, 7, 6, 5, 5, 4]
17
18categorias = {"Laptop": "Portátiles", "Monitor": "Pantallas",
19              "Teclado": "Periféricos", "Mouse": "Periféricos",
20              "Auriculares": "Audio", "Webcam": "Video",
21              "SSD": "Almacenamiento", "RAM": "Componentes",
22              "Cable USB": "Accesorios", "Hub USB": "Accesorios"}
23
24# Ruta absoluta: es la carpeta montada como volumen, la misma en los tres
25# servicios. Con una ruta relativa el fichero acabaría dentro de work/ y Spark
26# no lo encontraría.
27filas_total = 1_000_000
28
29with open("/home/jovyan/data/ventas.csv", "w", newline="") as f:
30    writer = csv.writer(f)
31    writer.writerow(["id", "fecha", "producto", "categoria",
32                     "ciudad", "cantidad", "precio_unitario", "cliente_id"])
33
34    fecha_base = datetime(2024, 1, 1)
35    ultima_fila = None
36
37    for i in range(filas_total):
38        fecha = fecha_base + timedelta(days=random.randint(0, 364))
39        producto = random.choices(productos, weights=pesos_producto)[0]
40        ciudad = random.choices(ciudades, weights=pesos_ciudad)[0]
41
42        # Se construye primero la fila limpia...
43        fila = [i + 1, fecha.strftime("%Y-%m-%d"), producto,
44                categorias[producto], ciudad,
45                random.randint(1, 5),
46                round(random.uniform(5, 1500), 2),
47                random.randint(1, 50000)]
48
49        # ...y cada tramo ensucia UN campo. 16.000 filas defectuosas (1,6%)
50        # repartidas en cinco defectos distintos más los duplicados.
51        if i < 4000:
52            fila[1] = ""                            # fecha vacía
53        elif i < 6000:
54            fila[1] = "31/13/2024"                  # fecha imposible
55        elif i < 9000:
56            fila[5] = ""                            # cantidad vacía
57        elif i < 10000:
58            fila[6] = -fila[6]                      # precio negativo
59        elif i < 15000:
60            fila[4] = f" {ciudad} "                 # espacios alrededor
61        elif i < 17000:
62            # 1.000 parejas de filas idénticas, el id incluido: duplicado
63            # exacto, que es lo que hay que deduplicar
64            if (i - 15000) % 2 == 1:
65                fila = list(ultima_fila)
66
67        writer.writerow(fila)
68        ultima_fila = fila
69
70print("[OK] Dataset ventas.csv generado: 1,000,000 filas")
71print("  Con pesos: Cable USB ~27%, Mouse ~20%, Laptop ~2%")
72print("  Con suciedad: 16.000 filas defectuosas (1,6%)")
73print("    4.000 fecha vacía · 2.000 fecha 31/13/2024 · 3.000 cantidad vacía")
74print("    1.000 precio negativo · 5.000 ciudad con espacios")
75print("    1.000 parejas de filas idénticas (2.000 filas, 1.000 de sobra)")

Generamos 1 millón de ventas ficticias para practicar con datos realistas

## Paso 7: Leer el dataset con Spark

1# Leer el CSV con Spark
2ventas = (
3    spark.read
4    .option("header", "true")
5    .option("inferSchema", "true")
6    .csv("/home/jovyan/data/ventas.csv")
7)
8
9# Verificar esquema y primeras filas
10ventas.printSchema()
11# root
12#  |-- id: integer
13#  |-- fecha: string
14#  |-- producto: string
15#  |-- categoria: string
16#  |-- ciudad: string
17#  |-- cantidad: integer
18#  |-- precio_unitario: double
19#  |-- cliente_id: integer
20
21ventas.show(5)
22print(f"Total filas: {ventas.count():,}")
23print(f"Particiones: {ventas.rdd.getNumPartitions()}")

Spark lee el CSV, infiere tipos y lo particiona automáticamente

## Comandos útiles del día a día

1# Levantar el cluster
2docker compose up -d
3
4# Ver logs en tiempo real
5docker compose logs -f
6
7# Parar el cluster (conserva datos)
8docker compose stop
9
10# Destruir el cluster (elimina contenedores, conserva volúmenes)
11docker compose down
12
13# Destruir TODO incluyendo volúmenes
14docker compose down -v
15
16# Reiniciar un servicio específico
17docker compose restart spark-worker
18
19# Escalar workers (ej: 3 workers)
20docker compose up -d --scale spark-worker=3

Comandos Docker Compose que usarás constantemente

Consejo de senior: añade estos alias a tu .bashrc o .zshrc para no escribir "docker compose" mil veces al día: alias spark-up="docker compose up -d" / alias spark-down="docker compose down" / alias spark-logs="docker compose logs -f"

Regístrate para guardar tu progreso.