11  Introducción a PySpark y Fundamentos ETL

11.1 ¿Qué es el ETL y por qué importa?

ETL (Extract, Transform, Load) es el proceso que mueve datos desde los sistemas operacionales (OLTP, APIs, logs, archivos) hacia el Data Warehouse, transformándolos en el camino para que sean consistentes, limpios y aptos para el análisis.

Nota

Distinción con el Preprocesamiento (Parte II) ETL es ingeniería de datos: opera sobre flujos de producción, terabytes, esquemas fijos, y su salida es reproducible y automatizada. El preprocesamiento analítico (Parte II) toma una vista ya limpia del DW y la transforma para un experimento concreto. El DW es el punto de transferencia entre ambos.

11.1.1 ETL clásico vs ELT moderno

ETL clásico ELT moderno
Orden Extraer → Transformar → Cargar Extraer → Cargar → Transformar
Dónde transforma Motor ETL externo Dentro del DW / lakehouse
Herramientas Informatica, SSIS, Talend dbt + BigQuery / Snowflake / DuckDB
Ventaja Datos siempre limpios al llegar Aprovechar la potencia del DW columnares
Desventaja Cuello de botella en el motor El DW necesita ser potente y barato

11.1.2 ¿Por qué PySpark?

Cuando los datos superan la memoria RAM de una sola máquina, pandas deja de ser suficiente. Apache Spark divide el trabajo en particiones que se procesan en paralelo sobre un clúster de nodos, usando un plan de ejecución optimizado (DAG — Directed Acyclic Graph).

PySpark es la API Python de Spark. Con ella puedes escribir transformaciones que escalan de un laptop (modo local) a cientos de nodos (AWS EMR, Databricks, GCP Dataproc) sin cambiar el código.

11.2 Entorno de prácticas: Databricks Community Edition

Todas las prácticas de este bloque se realizan en Databricks Community Edition, que proporciona un clúster Spark gratuito sin necesidad de instalación local.

11.2.1 Configurar la cuenta

  1. Ir a community.cloud.databricks.com y crear una cuenta gratuita.
  2. En el menú lateral: Compute → Create compute → elegir el runtime Databricks Runtime 14.x LTS (Spark 3.5, Python 3.11).
  3. Crear un Notebook (lenguaje: Python).
Importante

En Databricks, spark ya existe como variable global — no crees un SparkSession manual. Simplemente empieza con spark.read...

11.2.2 Cargar datasets a DBFS

# Databricks File System (DBFS) — sistema de archivos del clúster
# Opción A: subir archivo desde la UI (File → Upload data)

# Opción B: descargar directamente desde el notebook con dbutils
dbutils.fs.mkdirs("dbfs:/datasets/taxi/")

# Descargar NYC Taxi (Parquet, ~500 MB por mes)
import urllib.request
url = "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2023-01.parquet"
urllib.request.urlretrieve(url, "/tmp/yellow_2023_01.parquet")
dbutils.fs.mv("file:/tmp/yellow_2023_01.parquet", "dbfs:/datasets/taxi/yellow_2023_01.parquet")

# Verificar
display(dbutils.fs.ls("dbfs:/datasets/taxi/"))
# Leer desde DBFS en Databricks
df = spark.read.parquet("dbfs:/datasets/taxi/yellow_2023_01.parquet")
display(df.limit(5))   # display() es el equivalente de show() en Databricks

11.2.3 Comandos específicos de Databricks

Comando Descripción
display(df) Tabla interactiva con gráficos integrados
dbutils.fs.ls("dbfs:/ruta/") Listar archivos en DBFS
dbutils.fs.mkdirs("dbfs:/ruta/") Crear directorio en DBFS
dbutils.fs.rm("dbfs:/ruta/", True) Eliminar archivo o directorio
%sql SELECT ... Celda SQL sobre tablas registradas
spark.catalog.listTables() Ver tablas disponibles en el metastore
Tip

Delta Lake está preinstalado en Databricks. No necesitas pip install delta-spark — simplemente usa .format("delta") directamente.

11.2.4 Datasets para las prácticas

Dataset Tamaño Dónde obtenerlo en Databricks
NYC Taxi (1 mes) ~500 MB Descargar con urllib (ver §2.2)
NYC Taxi (todo 2023) ~6 GB Bucle sobre los 12 meses
TPC-H sintético Configurable %pip install tpch + python -m tpch generate
Databricks sample data Varios dbfs:/databricks-datasets/ (preinstalado) 
# Explorar los datasets de muestra de Databricks
display(dbutils.fs.ls("dbfs:/databricks-datasets/"))
# Incluye: nyctaxi, airlines, wine-quality, iot-stream, retail-org, etc.

# Leer el dataset de ejemplo de taxis (ya disponible)
df_sample = spark.read.format("delta")\
    .load("dbfs:/databricks-datasets/nyctaxi/tables/nyctaxi_yellow")
display(df_sample.limit(10))

11.3 PySpark: Fundamentos

11.3.1 SparkSession

En Databricks el SparkSession ya existe como spark. En entorno local:

from pyspark.sql import SparkSession

# Solo necesario fuera de Databricks
spark = (SparkSession.builder
    .appName("ETL-Mineria-Datos")
    .master("local[*]")
    .config("spark.sql.shuffle.partitions", "8")
    .getOrCreate())

11.3.2 El DataFrame de Spark: lazy vs eager

# Transformaciones (lazy) — solo construyen el plan, no ejecutan nada
df = spark.read.parquet("dbfs:/datasets/taxi/yellow_2023_01.parquet")
df2 = df.filter(df.trip_distance > 1.0)
df3 = df2.select("tpep_pickup_datetime", "trip_distance", "total_amount")

# Acción — dispara la ejecución del DAG completo
df3.show(5)    # o display(df3) en Databricks
df3.count()
Concepto Significado
Transformación filter, select, groupBy, join — construyen el DAG
Acción show, count, collect, write — ejecutan el DAG
Partición Unidad de paralelismo; un archivo Parquet por partición típicamente
RDD Capa de bajo nivel; usa DataFrame salvo casos especiales

11.3.3 Esquema explícito (mejor práctica)

from pyspark.sql.types import *

schema = StructType([
    StructField("tpep_pickup_datetime",  TimestampType(), True),
    StructField("tpep_dropoff_datetime", TimestampType(), True),
    StructField("passenger_count",       IntegerType(),   True),
    StructField("trip_distance",         DoubleType(),    True),
    StructField("fare_amount",           DoubleType(),    True),
    StructField("tip_amount",            DoubleType(),    True),
    StructField("total_amount",          DoubleType(),    True),
    StructField("payment_type",          IntegerType(),   True),
])

df = spark.read.parquet("dbfs:/datasets/taxi/yellow_2023_01.parquet")
# Parquet guarda el schema; con CSV/JSON conviene declararlo explícito:
# df = spark.read.csv("dbfs:/ruta.csv", header=True, schema=schema)

inferSchema=True hace una pasada completa del archivo; con esquema explícito la lectura es más rápida y sin sorpresas de tipos.

11.4 Referencias de este capítulo