RDDs y DataFrames en Spark SQL

En Spark SQL, existen dos formas principales de trabajar con datos estructurados: RDDs y DataFrames. Mientras que los RDDs representan el enfoque tradicional, los DataFrames son la abstracción de datos preferida en versiones más recientes de Spark.

Entender las diferencias entre ambas APIs es fundamental para cualquier desarrollador que trabaje con Apache Spark. Elegir la abstracción correcta impacta directamente en el rendimiento, la legibilidad del código y la capacidad de Spark para optimizar la ejecución de tus operaciones. En este artículo exploraremos ambas APIs en profundidad, con ejemplos prácticos en PySpark, y veremos cómo Spark SQL y el Catalyst Optimizer unifican todo el ecosistema.

RDDs (Resilient Distributed Datasets)

Los RDDs son colecciones de elementos distribuidos en los nodos de un clúster, sobre los cuales se pueden realizar operaciones en paralelo. Se caracterizan por:

• Dependencias: Describen cómo se construye un RDD a partir de sus entradas.
• Particiones: Permiten dividir el trabajo para paralelizar el cálculo.
• Función de cálculo: Genera un iterador para los datos del RDD.

Cada RDD mantiene un linaje (lineage), es decir, un registro de todas las transformaciones aplicadas desde los datos de origen. Este linaje permite a Spark reconstruir cualquier partición perdida en caso de fallo, lo que garantiza la tolerancia a fallos sin necesidad de replicar datos.

Creación de RDDs

Se pueden crear RDDs a partir de archivos de texto, colecciones de Python, etc.

from pyspark import SparkContext

sc = SparkContext("local", "RDD Example")

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

# Crear un RDD desde un archivo de texto
rdd_texto = sc.textFile("datos/ventas.txt")

Transformaciones y acciones

Las operaciones sobre RDDs se dividen en transformaciones y acciones:

• Transformaciones: Crean un nuevo RDD aplicando lógica a cada elemento del RDD existente. Son de evaluación perezosa, es decir, se agregan al DAG de cálculo y se ejecutan solo cuando se necesita el resultado.
  • Ejemplos: map(), flatMap(), filter(), coalesce(), repartition(), reduceByKey().
• Acciones: Activan el cálculo del DAG de transformaciones. Se dividen en conductoras (devuelven datos al controlador) y distribuidas (se ejecutan en los nodos del clúster).
  • Ejemplos: reduce(), count(), collect(), take(), max(), saveAsTextFile().

La evaluación perezosa (lazy evaluation) es un concepto clave: cuando aplicas una transformación como map() o filter(), Spark no ejecuta nada inmediatamente. En su lugar, construye un DAG (Directed Acyclic Graph) con todas las transformaciones encadenadas. La ejecución real solo ocurre cuando invocas una acción.

# Transformaciones (perezosas - no se ejecutan aún)
rdd_pares = rdd.filter(lambda x: x % 2 == 0)
rdd_cuadrados = rdd_pares.map(lambda x: x ** 2)

# Acción (dispara la ejecución del DAG completo)
resultado = rdd_cuadrados.collect()
print(resultado)  # [4, 16, 36, 64, 100]

Un ejemplo más completo con flatMap() y reduceByKey() para contar palabras:

# Conteo de palabras con RDDs
rdd_lineas = sc.textFile("datos/documento.txt")

conteo = (rdd_lineas
    .flatMap(lambda linea: linea.split(" "))
    .map(lambda palabra: (palabra.lower(), 1))
    .reduceByKey(lambda a, b: a + b))

for palabra, cantidad in conteo.collect():
    print(f"{palabra}: {cantidad}")

Caché y persistencia

Aspectos avanzados del trabajo con RDDs incluyen el almacenamiento en caché (persist(), cache(), unpersist()), el particionado (HashPartitioner, RangePartitioner), el shuffling de datos y el uso de variables broadcast y acumuladores.

Cuando un RDD se reutiliza en múltiples operaciones, almacenarlo en caché evita recalcularlo desde cero cada vez:

from pyspark import StorageLevel

# cache() almacena en memoria (equivale a persist(MEMORY_ONLY))
rdd_pares.cache()

# persist() permite elegir el nivel de almacenamiento
rdd_pares.persist(StorageLevel.MEMORY_AND_DISK)

# Liberar el caché cuando ya no se necesita
rdd_pares.unpersist()

DataFrames

Introducidos en Spark 1.6, los DataFrames son la abstracción de datos preferida para la mayoría de los casos de uso. Se basan en el concepto de esquema, que define la estructura de los datos.

Un esquema en Spark se define mediante StructType y StructField, que especifican el nombre, tipo de dato y si la columna admite valores nulos:

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

spark = SparkSession.builder.appName("DataFrame Example").getOrCreate()

# Definir el esquema explícitamente
esquema = StructType([
    StructField("nombre", StringType(), True),
    StructField("departamento", StringType(), True),
    StructField("salario", DoubleType(), True),
    StructField("edad", IntegerType(), True)
])

Creación de DataFrames

Se pueden crear DataFrames a partir de:

• RDDs.
• Fuentes de datos en diversos formatos: CSV, JSON, Parquet, ORC, etc.

# Crear un DataFrame desde una lista de tuplas
datos = [
    ("Ana", "Ingeniería", 75000.0, 30),
    ("Carlos", "Marketing", 62000.0, 28),
    ("Diana", "Ingeniería", 82000.0, 35),
    ("Eduardo", "Marketing", 58000.0, 25),
    ("Fernanda", "Datos", 90000.0, 32)
]

df = spark.createDataFrame(datos, schema=esquema)
df.show()

+--------+-------------+-------+----+
|  nombre|departamento |salario|edad|
+--------+-------------+-------+----+
|     Ana|  Ingeniería |75000.0|  30|
|  Carlos|    Marketing|62000.0|  28|
|   Diana|  Ingeniería |82000.0|  35|
| Eduardo|    Marketing|58000.0|  25|
|Fernanda|        Datos|90000.0|  32|
+--------+-------------+-------+----+

# Crear un DataFrame desde un archivo CSV
df_csv = spark.read.csv("datos/empleados.csv", header=True, inferSchema=True)

# Crear un DataFrame desde un archivo JSON
df_json = spark.read.json("datos/empleados.json")

# Crear un DataFrame desde un archivo Parquet
df_parquet = spark.read.parquet("datos/empleados.parquet")

Transformaciones y acciones sobre DataFrames

Al igual que con los RDDs, las operaciones sobre DataFrames se dividen en transformaciones y acciones. La sintaxis para trabajar con DataFrames es más concisa y similar a SQL.

• Transformaciones: Ejemplos: select(), selectExpr(), filter(), where(), distinct(), dropDuplicates(), sort(), orderBy(), limit(), withColumn(), withColumnRenamed(), drop(), sample(), randomSplit().
• Acciones: Ejemplos: show(), count(), collect(), first(), take(), head().

from pyspark.sql.functions import col, avg, count, desc

# Seleccionar columnas específicas
df.select("nombre", "salario").show()

# Filtrar filas
df.filter(col("salario") > 70000).show()

# Agregar una columna calculada
df_con_bono = df.withColumn("bono", col("salario") * 0.10)
df_con_bono.show()

# Agrupar y agregar
df.groupBy("departamento") \
    .agg(
        avg("salario").alias("salario_promedio"),
        count("nombre").alias("total_empleados")
    ) \
    .orderBy(desc("salario_promedio")) \
    .show()

+-------------+----------------+---------------+
|departamento |salario_promedio|total_empleados|
+-------------+----------------+---------------+
|        Datos|         90000.0|              1|
|  Ingeniería |         78500.0|              2|
|    Marketing|         60000.0|              2|
+-------------+----------------+---------------+

Comparación entre RDDs y DataFrames en Spark

Ventajas de los DataFrames

Los DataFrames ofrecen ventajas significativas sobre los RDDs en términos de rendimiento y facilidad de uso:

• El Catalyst Optimizer optimiza automáticamente las operaciones sobre DataFrames, lo que resulta en una ejecución más rápida.
• La sintaxis similar a SQL hace que los DataFrames sean más intuitivos para trabajar con datos estructurados.

En general, los DataFrames son la opción preferida para trabajar con datos estructurados en Spark SQL. Los RDDs siguen siendo relevantes para casos de uso específicos, como:

• El trabajo con datos no estructurados.
• La manipulación de bajo nivel de particiones.

Spark SQL

Spark SQL es un módulo de Spark para trabajar con datos estructurados. Se basa en el concepto de DataFrames, que son colecciones de datos organizadas en columnas con nombre y tipos de datos definidos. Esto permite a Spark realizar optimizaciones significativas para un mejor rendimiento.

La integración entre DataFrames y Spark SQL es directa: cualquier DataFrame puede registrarse como una vista temporal y consultarse con SQL estándar. Esto es especialmente útil para equipos donde algunos miembros prefieren SQL y otros prefieren la API programática.

# Registrar el DataFrame como una vista temporal
df.createOrReplaceTempView("empleados")

# Ejecutar consultas SQL directamente
resultado = spark.sql("""
    SELECT departamento,
           AVG(salario) AS salario_promedio,
           COUNT(*) AS total
    FROM empleados
    WHERE edad > 25
    GROUP BY departamento
    ORDER BY salario_promedio DESC
""")

resultado.show()

+-------------+----------------+-----+
|departamento |salario_promedio|total|
+-------------+----------------+-----+
|        Datos|         90000.0|    1|
|  Ingeniería |         78500.0|    2|
|    Marketing|         62000.0|    1|
+-------------+----------------+-----+

Las vistas temporales existen dentro de la sesión de Spark que las creó. Si necesitas compartir una vista entre sesiones, puedes usar createOrReplaceGlobalTempView(), que registra la vista en la base de datos global global_temp.

Catalyst Optimizer

El Catalyst Optimizer es un componente clave de Spark SQL. Su función es optimizar las operaciones sobre DataFrames para una ejecución más rápida. Lo hace traduciendo la lógica de procesamiento de datos en un plan lógico, optimizándolo y luego convirtiéndolo en un plan físico para la ejecución.

El proceso de optimización sigue cuatro fases:

1. Análisis: Resuelve las referencias a columnas y tablas, validando que el plan lógico sea correcto.
2. Optimización lógica: Aplica reglas como predicate pushdown (mover filtros lo más cerca posible de la fuente de datos), constant folding (evaluar expresiones constantes en tiempo de compilación) y column pruning (eliminar columnas que no se usan en el resultado final).
3. Planificación física: Genera uno o más planes físicos a partir del plan lógico optimizado y selecciona el más eficiente basándose en un modelo de costos.
4. Generación de código: Produce bytecode Java optimizado para la ejecución (code generation con Tungsten).

Puedes inspeccionar el plan de ejecución de cualquier DataFrame usando explain():

# Ver el plan de ejecución completo
df.filter(col("salario") > 70000) \
    .select("nombre", "departamento") \
    .explain(True)

== Parsed Logical Plan ==
Project [nombre, departamento]
+- Filter (salario > 70000.0)
   +- LogicalRDD [nombre, departamento, salario, edad]

== Analyzed Logical Plan ==
nombre: string, departamento: string
Project [nombre, departamento]
+- Filter (salario > 70000.0)
   +- LogicalRDD [nombre, departamento, salario, edad]

== Optimized Logical Plan ==
Project [nombre, departamento]
+- Filter (salario > 70000.0)
   +- LogicalRDD [nombre, departamento, salario, edad]

== Physical Plan ==
*(1) Project [nombre, departamento]
+- *(1) Filter (salario > 70000.0)
   +- *(1) Scan ExistingRDD [nombre, departamento, salario, edad]

Observa cómo Catalyst aplica column pruning automáticamente: aunque el RDD original tiene 4 columnas, el plan físico solo lee nombre, departamento y salario (esta última necesaria para el filtro).

Buenas prácticas: ¿Cuándo usar RDDs vs DataFrames?

Aunque los DataFrames son la opción recomendada en la mayoría de escenarios, hay situaciones donde los RDDs siguen siendo la mejor elección:

• Usa DataFrames cuando:
  • Trabajas con datos estructurados o semi-estructurados (CSV, JSON, Parquet).
  • Necesitas el máximo rendimiento gracias al Catalyst Optimizer.
  • Tu equipo está familiarizado con SQL.
  • Quieres interoperabilidad con otros lenguajes (Scala, Java, R).
• Usa RDDs cuando:
  • Necesitas control de bajo nivel sobre las particiones y el paralelismo.
  • Trabajas con datos no estructurados (texto libre, binarios).
  • Necesitas implementar lógica personalizada que no se puede expresar con las funciones built-in de DataFrames.
  • Mantienes código legacy que aún no se ha migrado.

Una práctica común es combinar ambas APIs: usar DataFrames para las transformaciones principales y convertir a RDD solo cuando necesitas operaciones de bajo nivel específicas.

# Convertir DataFrame a RDD cuando necesitas control fino
rdd_desde_df = df.rdd.map(lambda row: (row.nombre, row.salario * 1.10))

# Convertir RDD de vuelta a DataFrame
from pyspark.sql import Row
rdd_rows = rdd_desde_df.map(lambda x: Row(nombre=x[0], salario_ajustado=x[1]))
df_ajustado = spark.createDataFrame(rdd_rows)
df_ajustado.show()

Conclusión

RDDs y DataFrames son las dos abstracciones fundamentales de Apache Spark para el procesamiento distribuido de datos. Los RDDs ofrecen control de bajo nivel y flexibilidad, mientras que los DataFrames proporcionan una API de alto nivel optimizada por el Catalyst Optimizer que resulta en un mejor rendimiento para la mayoría de los casos de uso.

Spark SQL actúa como el puente que conecta ambos mundos, permitiendo ejecutar consultas SQL sobre DataFrames y aprovechando las optimizaciones de Catalyst de forma transparente. Para proyectos nuevos, la recomendación es clara: comienza con DataFrames y Spark SQL, y recurre a los RDDs solo cuando necesites ese nivel adicional de control sobre la ejecución.