Formación corporativa en PySpark

Configurar correctamente el entorno de trabajo es el primer paso para desarrollar con PySpark. La instalación más sencilla se realiza con pip install pyspark, que incluye una distribución de Apache Spark y gestiona automáticamente las dependencias de Java necesarias para ejecutar la JVM.

La SparkSession es el punto de entrada principal de cualquier aplicación PySpark. Se crea mediante el patrón builder y permite configurar parámetros como el nombre de la aplicación, el modo de ejecución (local[*] para desarrollo, yarn o k8s para producción) y opciones avanzadas de memoria y paralelismo.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MiAplicacion") \
    .master("local[*]") \
    .config("spark.executor.memory", "2g") \
    .config("spark.driver.memory", "1g") \
    .getOrCreate()

print(f"Versión de Spark: {spark.version}")

La Spark UI (accesible en http://localhost:4040 durante la ejecución) proporciona información detallada sobre jobs, stages, tasks, uso de almacenamiento y variables de entorno. Es la herramienta principal para diagnosticar problemas de rendimiento y entender cómo Spark ejecuta las operaciones.

Para flujos de trabajo interactivos, PySpark se integra perfectamente con Jupyter Notebooks, Databricks y Google Colab, que proporcionan entornos preconfigurados con soporte para visualización y exploración de datos distribuidos.

Los RDDs (Resilient Distributed Datasets) son la abstracción de datos fundamental de Apache Spark. Representan colecciones inmutables y tolerantes a fallos de elementos distribuidos entre los nodos del clúster. Aunque en la práctica moderna se prefieren los DataFrames, los RDDs siguen siendo el sustrato subyacente y son esenciales para comprender el funcionamiento interno de Spark.

Se pueden crear RDDs de dos formas principales: mediante parallelize() a partir de colecciones Python en memoria, o mediante textFile() para leer datos desde el sistema de ficheros o HDFS:

sc = spark.sparkContext

# Crear RDD desde una colección Python
rdd_numeros = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], numSlices=4)

# Crear RDD desde un fichero de texto
rdd_texto = sc.textFile("hdfs://ruta/al/archivo.txt")

Las transformaciones son operaciones perezosas que producen un nuevo RDD: map, filter, flatMap, distinct, union, reduceByKey, groupByKey, sortByKey, join y cogroup. Las acciones desencadenan la ejecución del DAG y devuelven un resultado: collect, count, first, take, reduce, saveAsTextFile y foreach.

La persistencia permite mantener RDDs en memoria o disco para reutilizarlos eficientemente en múltiples acciones. Los niveles MEMORY_ONLY, MEMORY_AND_DISK y DISK_ONLY controlan dónde se almacenan los datos. unpersist() libera el espacio cuando el RDD ya no es necesario.

Los DataFrames son la API principal de PySpark para trabajar con datos estructurados. A diferencia de los RDDs, los DataFrames tienen un esquema definido y se benefician del optimizador Catalyst, que genera planes de ejecución más eficientes de forma automática.

Se pueden crear DataFrames desde múltiples fuentes: listas de tuplas Python, diccionarios, RDDs existentes o DataFrames de pandas. La definición explícita del esquema con StructType y StructField es recomendable en producción para evitar inferencias costosas y garantizar la corrección de tipos:

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType

schema = StructType([
    StructField("nombre", StringType(), nullable=False),
    StructField("edad", IntegerType(), nullable=True),
    StructField("salario", DoubleType(), nullable=True)
])

df = spark.createDataFrame([("Ana", 30, 45000.0), ("Luis", 25, 38000.0)], schema)
df.printSchema()
df.show()

Las operaciones básicas incluyen select para proyección de columnas, filter/where para filtrado, withColumn para añadir o transformar columnas, alias para renombrar, drop para eliminar y withColumnRenamed para cambios de nombre. Para agrupaciones se usa groupBy con funciones de agregación (count, sum, avg, min, max). La ordenación se realiza con orderBy/sort y la deduplicación con distinct y dropDuplicates.

Spark SQL permite ejecutar consultas SQL estándar directamente sobre DataFrames, combinando la expresividad del lenguaje SQL con la potencia del procesamiento distribuido de Spark. Esta integración facilita la adopción de PySpark por equipos con experiencia en SQL y permite reutilizar consultas existentes.

Para ejecutar SQL, primero se registra el DataFrame como vista temporal con createOrReplaceTempView() y después se lanza la consulta con spark.sql():

df.createOrReplaceTempView("empleados")

resultado = spark.sql("""
    SELECT departamento, AVG(salario) AS salario_medio, COUNT(*) AS total
    FROM empleados
    WHERE activo = true
    GROUP BY departamento
    ORDER BY salario_medio DESC
""")
resultado.show()

Los joins combinan DataFrames por una condición: inner, left, right, full, cross, left_semi y left_anti. Las window functions (row_number, rank, dense_rank, lag, lead, sum acumulado) permiten cálculos sobre ventanas de filas definidas con partitionBy y orderBy.

Las UDFs extienden el motor SQL con lógica Python personalizada. Las pandas_udf (vectorizadas con Apache Arrow) ofrecen un rendimiento significativamente mayor que las UDFs clásicas de Python al procesar datos en lotes en lugar de fila a fila. El optimizador Catalyst analiza y reescribe el plan lógico para generar el plan físico más eficiente posible.

Las transformaciones avanzadas de PySpark cubren las necesidades más complejas de manipulación de datos estructurados y semiestructurados que van más allá de las operaciones básicas de selección y filtrado.

El pivoteo con pivot() y la operación inversa con stack() permiten reestructurar DataFrames para transformar filas en columnas o viceversa, operación habitual en la preparación de datos para análisis y reporting. Las funciones explode y posexplode convierten columnas de arrays o mapas en múltiples filas, facilitando el trabajo con datos anidados de JSON o formatos de eventos:

from pyspark.sql.functions import explode, col

# Explotar una columna de array en filas individuales
df_explodido = df.withColumn("etiqueta", explode(col("etiquetas")))
df_explodido.show()

Las funciones de fecha (year, month, dayofweek, datediff, date_add, to_timestamp, unix_timestamp) son esenciales para series temporales y análisis de logs. Las funciones de cadena (concat, substring, regexp_extract, split, trim, lower, upper) permiten limpiar y transformar texto a escala.

El manejo de nulos con isNull, isNotNull, coalesce, fillna, dropna y la función condicional when/otherwise son herramientas indispensables para la limpieza y calidad de datos en pipelines de producción.

Structured Streaming es el motor de procesamiento en tiempo real de Spark. A diferencia de los enfoques tradicionales de streaming, utiliza la misma API de DataFrames que el procesamiento por lotes, lo que facilita la reutilización de código y la curva de aprendizaje.

El modelo de ejecución es microbatch por defecto: Spark procesa los datos en pequeños lotes a intervalos configurables mediante trigger. También existe el modo continuo para latencias muy bajas. Las fuentes de datos disponibles incluyen archivos en directorios, sockets TCP, rate (para pruebas) y Apache Kafka:

# Leer un stream desde Kafka
stream_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "eventos") \
    .load()

# Procesar: deserializar el valor y agregar por ventana de tiempo
from pyspark.sql.functions import window, col, from_json
from pyspark.sql.types import StructType, StringType, DoubleType

schema = StructType().add("producto", StringType()).add("precio", DoubleType())
df_parsed = stream_df.select(from_json(col("value").cast("string"), schema).alias("data")).select("data.*")

df_ventana = df_parsed.groupBy(window(col("timestamp"), "5 minutes"), "producto").sum("precio")

# Escribir el resultado
query = df_ventana.writeStream.outputMode("update").format("console").start()

Los watermarks gestionan los datos que llegan con retraso, definiendo cuánto tiempo esperar antes de considerar una ventana temporal completa. Los modos de salida append, complete y update controlan qué filas se emiten en cada microbatch. El checkpointing garantiza la recuperación ante fallos con semántica exactly-once.

MLlib es la biblioteca de machine learning de Apache Spark, diseñada para entrenar modelos sobre conjuntos de datos que superan la capacidad de memoria de una sola máquina. Proporciona algoritmos de clasificación, regresión, clustering y reducción de dimensionalidad, todos integrados con la API de DataFrames.

El concepto clave es el Pipeline, que encadena transformaciones (Transformer) y estimadores (Estimator) en un flujo de trabajo reproducible. Los Transformer aplican transformaciones (como VectorAssembler o StandardScaler) y los Estimator aprenden de los datos con fit() para producir un modelo que luego transforma nuevos datos con transform():

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# Construir el pipeline
assembler = VectorAssembler(inputCols=["col1", "col2", "col3"], outputCol="features_raw")
scaler = StandardScaler(inputCol="features_raw", outputCol="features")
rf = RandomForestClassifier(featuresCol="features", labelCol="label", numTrees=100)

pipeline = Pipeline(stages=[assembler, scaler, rf])

# Entrenar y evaluar
train, test = df.randomSplit([0.8, 0.2], seed=42)
modelo = pipeline.fit(train)
predicciones = modelo.transform(test)

evaluador = BinaryClassificationEvaluator(labelCol="label")
print(f"AUC: {evaluador.evaluate(predicciones):.4f}")

El módulo incluye algoritmos de clasificación (LogisticRegressión, RandomForestClassifier, GBTClassifier, NaiveBayes), regresión (LinearRegressión, RandomForestRegressor, GBTRegressor), clustering (KMeans, BisectingKMeans, GaussianMixture) y reducción de dimensionalidad con PCA. La búsqueda de hiperparámetros se realiza con CrossValidator y ParamGridBuilder.

Llevar aplicaciones PySpark a producción implica dominar técnicas de particionamiento, optimización de rendimiento y despliegue en clúster. La eficiencia de una aplicación Spark depende en gran medida de cómo se distribuyen los datos y de las decisiones de configuración tomadas durante el desarrollo.

El particionamiento controla cómo se distribuyen los datos entre los executors. repartition(n) redistribuye todos los datos con un shuffle completo (útil para aumentar el paralelismo), mientras que coalesce(n) reduce particiones sin shuffle (más eficiente para escribir ficheros). El partitionBy en la escritura organiza los datos en directorios por columna, acelerando las lecturas posteriores con predicado pushdown:

# Escribir con particionamiento por año y mes
df.write \
    .mode("overwrite") \
    .partitionBy("anio", "mes") \
    .parquet("s3://mi-bucket/datos/")

Las broadcast variables comparten datos de solo lectura (típicamente tablas de dimensión pequeñas) con todos los executors sin enviarlos en cada tarea. Los accumulators permiten agregar métricas desde los executors al driver de forma segura. La función broadcast() también optimiza los joins cuando una de las tablas es pequeña, evitando un shuffle costoso.

La Adaptive Query Execution (AQE) en Spark 3.x ajusta automáticamente el plan de ejecución en tiempo real según las estadísticas reales de los datos. spark-submit es la herramienta de línea de comandos para enviar aplicaciones a clústeres Standalone, YARN o Kubernetes con control total sobre recursos y configuración.