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.

⏱ 60 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      - ./data:/opt/spark-data
16      - ./output:/opt/spark-output
17
18  spark-worker:
19    image: bitnami/spark:3.5
20    environment:
21      - SPARK_MODE=worker
22      - SPARK_MASTER_URL=spark://spark-master:7077
23      - SPARK_WORKER_MEMORY=2G
24      - SPARK_WORKER_CORES=2
25    depends_on:
26      - spark-master
27    volumes:
28      - ./data:/opt/spark-data
29      - ./output:/opt/spark-output
30
31  jupyter:
32    image: quay.io/jupyter/pyspark-notebook:spark-3.5.0
33    container_name: pyspark-jupyter
34    environment:
35      - SPARK_MASTER=spark://spark-master:7077
36    ports:
37      - "8888:8888"   # Jupyter Notebook
38      - "4040:4040"   # Spark UI del driver (para diagnóstico)
39    volumes:
40      - ./notebooks:/home/jovyan/work
41      - ./data:/home/jovyan/data
42      - ./output:/home/jovyan/output
43    depends_on:
44      - spark-master
45      - 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.

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# Salida esperada:
8# NAME             STATUS    PORTS
9# spark-master     running   0.0.0.0:7077->7077, 0.0.0.0:8080->8080
10# spark-worker     running   
11# pyspark-jupyter  running   0.0.0.0:4040->4040, 0.0.0.0:8888->8888
12
13# Ver los logs del master para confirmar que el worker se conectó
14docker compose logs spark-master | grep -i "registered"
15# 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 un notebook nuevo y 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("✓ ¡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# Guardar en data/ventas.csv (montado como volumen Docker)
25filas_total = 1_000_000
26filas_sucias = 17_000  # 1.7% de suciedad sembrada
27with open("data/ventas.csv", "w", newline="") as f:
28    writer = csv.writer(f)
29    writer.writerow(["id", "fecha", "producto", "categoria",
30                     "ciudad", "cantidad", "precio_unitario", "cliente_id"])
31
32    fecha_base = datetime(2024, 1, 1)
33    for i in range(filas_total):
34        fecha = fecha_base + timedelta(days=random.randint(0, 364))
35        producto = random.choices(productos, weights=pesos_producto)[0]
36        ciudad = random.choices(ciudades, weights=pesos_ciudad)[0]
37
38        # Sembrar suciedad (1.7%): el alumno la limpiará en la L07
39        if i < 4000:
40            fecha_str = ""                          # fecha vacía
41        elif i < 6000:
42            fecha_str = "31/13/2024"               # fecha mal formada
43        elif i < 9000:
44            cantidad = ""                          # cantidad vacía
45            writer.writerow([i+1, fecha.strftime("%Y-%m-%d"),
46                producto, categorias[producto], ciudad,
47                cantidad, round(random.uniform(5, 1500), 2),
48                random.randint(1, 50000)])
49            continue
50        elif i < 10000:
51            precio = -round(random.uniform(5, 500), 2)  # precio negativo
52            writer.writerow([i+1, fecha.strftime("%Y-%m-%d"),
53                producto, categorias[producto], ciudad,
54                random.randint(1, 5), precio,
55                random.randint(1, 50000)])
56            continue
57        elif i < 15000:
58            ciudad = f" {ciudad} "                 # espacios extra
59        elif i < 17000:
60            # duplicado exacto de la fila anterior
61            pass
62        else:
63            fecha_str = fecha.strftime("%Y-%m-%d")
64            writer.writerow([i+1, fecha_str, producto,
65                categorias[producto], ciudad,
66                random.randint(1, 5),
67                round(random.uniform(5, 1500), 2),
68                random.randint(1, 50000)])
69            continue
70
71        if i < 4000:
72            writer.writerow([i+1, "", producto,
73                categorias[producto], ciudad,
74                random.randint(1, 5),
75                round(random.uniform(5, 1500), 2),
76                random.randint(1, 50000)])
77        elif i < 6000:
78            writer.writerow([i+1, "31/13/2024", producto,
79                categorias[producto], ciudad,
80                random.randint(1, 5),
81                round(random.uniform(5, 1500), 2),
82                random.randint(1, 50000)])
83        else:
84            writer.writerow([i+1, fecha.strftime("%Y-%m-%d"),
85                producto, categorias[producto], ciudad,
86                random.randint(1, 5),
87                round(random.uniform(5, 1500), 2),
88                random.randint(1, 50000)])
89
90print("✓ Dataset ventas.csv generado: 1,000,000 filas")
91print("  Con pesos: Cable USB ~27%, Mouse ~20%, Laptop ~2%")
92print("  Con suciedad: ~17,000 filas (1.7%) para limpiar en la L07")

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"

## ejercicios

[01]

Verificar que tu cluster funciona

Escribe un script en el notebook de Jupyter (http://localhost:8888) que conecte al cluster Spark, imprima la versión, cree un DataFrame con 5 filas de datos inventados, muestre el esquema y cuente las filas. Si todo funciona, imprime "✓ Cluster operativo". Este código se conecta a spark-master por la red interna de Docker — no funciona fuera del notebook.

💡 Resultado esperado

Spark version: 3.5.0
root
 |-- nombre: string (nullable = true)
 |-- departamento: string (nullable = true)

Cargando editor...

[02]

Leer un CSV y explorar los datos

En el notebook de Jupyter (http://localhost:8888), lee el archivo ventas.csv con Spark, muestra las primeras 10 filas, imprime el esquema, y calcula: número total de filas, número de ciudades únicas y número de productos únicos. Este código usa spark-master y /home/jovyan/data/ — solo funciona dentro del notebook.

💡 Resultado esperado

Total filas: 1,000,000
Ciudades únicas: 10
Productos únicos: 10

Cargando editor...

[03]

Escalar el cluster y observar el cambio

Escribe los comandos de Docker Compose para: 1) escalar a 3 workers, 2) verificar que los 3 están corriendo, 3) comprobar en la Spark UI que aparecen 3 workers registrados.

💡 Resultado esperado

Despues de escalar, docker compose ps lista 5 contenedores:
  spark-master, pyspark-jupyter y 3 workers.
docker compose logs spark-master | grep "Registered worker" saca 3 lineas.
En http://localhost:8080 la tabla Workers tiene 3 filas

Cargando editor...

Regístrate para guardar tu progreso.

## comentarios

Reporta erratas, ayuda a otros o comparte tu opinión. Sé constructivo.

Inicia sesión para comentar y responder.

cargando comentarios...