En mis dos primeros artículos de esta serie, ya has cubierto mucho terreno.
Sabe cómo crear una SparkSession, cargar datos en un DataFrame, definir esquemas, limpiar columnas desordenadas, unir conjuntos de datos, escribir archivos Parquet y crear flujos de trabajo simples de PySpark. Pondré enlaces a esos artículos de esta serie al final de ésta.
Eso ya es suficiente para hacer un trabajo útil. Pero una vez que sus datos crecen o sus transformaciones se vuelven más complejas, comienza a aparecer una nueva serie de preguntas.
¿Por qué este trabajo de repente se volvió lento? ¿Por qué un simple groupBy() tarda más de lo esperado? ¿Por qué PySpark escribe tantos archivos de salida? ¿Cuándo debería almacenar en caché un DataFrame? ¿Y qué diablos es un barajado?
Este artículo trata sobre esos próximos pasos.
Seguimos manteniendo las cosas amigables. No profundizaremos en el ajuste del clúster, los componentes internos de la memoria, Kubernetes, YARN o configuraciones oscuras de PySpark. En lugar de ello, nos centraremos en ideas prácticas que le ayudarán a comprender qué hace PySpark y cómo escribir código PySpark que se comporte de forma más predecible.
El tema general es este:
Alcanzar el nivel intermedio de PySpark se trata principalmente de comprender el movimiento de datos.
Una vez que esa idea hace clic, PySpark comienza a sentirse mucho más predecible. Generalmente hay una buena razón por la cual sus trabajos de PySpark tienen un rendimiento deficiente; sólo necesitas saber dónde buscar.
Cuando PySpark comienza a sentirse lento
Cuando aprende PySpark por primera vez, es natural centrarse en el código.
Escribes cosas como:
df.filter(df.city == "Londres") df.groupBy("customer_id").sum("cantidad") df.join(clientes, on="customer_id")
Pero a medida que avanza hacia PySpark más avanzado, debe comenzar a pensar en lo que PySpark tiene que hacer detrás de escena.
Algunas operaciones son bastante baratas. Filtrar filas es un buen ejemplo. Spark puede mirar cada fila, tomar una decisión de sí o no y seguir adelante.
df_london = df.filter(df.city == "Londres")
Otras operaciones son más caras porque PySpark tiene que mover datos. Considere esto. Al igual que el filtro, es solo una línea de código, pero PySpark debe asegurarse de que todas las filas del mismo cliente terminen juntas.
df_totals = df.groupBy("customer_id").sum("cantidad")
Se trata de una operación mucho más complicada y es posible que necesite datos para moverse entre diferentes partes del trabajo. Ese movimiento es a menudo donde está el costo.
Entonces el hábito de nivel intermedio es simple:
Antes de acceder a la configuración de ajuste, haga una pregunta más simple: ¿esta línea de código obliga a Spark a mover datos?
Esto no significa que las uniones, agregaciones o la clasificación de datos sean malas. Al contrario, son imprescindibles, pero son el tipo de operaciones que debes tratar con un poco más de cuidado.
Cómo PySpark divide el trabajo
Un PySpark DataFrame no se trata como un único bloque de datos contiguo. PySpark lo divide en particiones. Puede pensar en las particiones como fragmentos de datos que PySpark puede procesar por separado y en paralelo. Esa es una de las principales razones por las que PySpark es eficaz y útil para procesar datos a gran escala.
Puedes comprobar cuántas particiones tiene un DataFrame de esta manera:
df.rdd.getNumPartitions()
Como resultado, obtendrá un número entero simple, por ejemplo:
8
Eso significa que PySpark ha dividido los datos en ocho fragmentos, y esto es importante porque la partición afecta la cantidad de trabajo paralelo que puede realizar PySpark. Si tiene muy pocas particiones, es posible que PySpark no pueda utilizar todos los núcleos de CPU disponibles. Si tiene demasiadas particiones, PySpark puede dedicar demasiado tiempo a administrar muchas tareas y conjuntos de datos pequeños. Ninguno de los extremos es ideal.
Si surge alguna de estas situaciones, PySpark ofrece dos funciones útiles para ayudar.
Puedes cambiar el número de particiones con repartition() df_more = df.repartition(16) Y puedes reducir el número de particiones con coalesce(): df_fewer = df.coalesce(4)
repartición() puede aumentar o disminuir la cantidad de particiones, pero hace que PySpark redistribuya los datos, lo que generalmente afectará el rendimiento.
coalesce() se utiliza para reducir la cantidad de particiones y necesita menos movimiento de datos.
Una regla práctica es:
Utilice repartición() cuando necesite distribuir los datos de manera más uniforme. Utilice coalesce() cuando simplemente desee menos archivos de salida.
Por ejemplo, supongamos que tiene un conjunto de datos que produce naturalmente 40 archivos Parquet pequeños cuando se escribe en el disco. Podrías hacer esto en su lugar:
df.coalesce(4).write.mode("sobrescribir").parquet("salida/ventas")
Eso le dice a PySpark que escriba el resultado con cuatro particiones, lo que normalmente produce cuatro archivos de salida.
No es necesario que se obsesione con el recuento exacto de particiones en esta etapa. El punto importante es que las particiones controlan cómo PySpark divide su trabajo. Una vez que comprenda estas ideas, será mucho más fácil razonar sobre PySpark. Los trabajos que se ralentizan dejan de parecer eventos aleatorios y empiezas a ver las causas probables: una mezcla aleatoria aquí, una combinación amplia allí, demasiados archivos pequeños en otros lugares.
Cuando los datos tienen que moverse
Algunas operaciones de PySpark pueden ocurrir dentro de cada partición. Otros requieren que se reúnan datos de diferentes particiones.
Ese movimiento se llama barajar.
Se produce una mezcla cuando PySpark tiene que redistribuir datos entre particiones. Esta es una de las primeras ideas de Spark que realmente cambia la forma de escribir PySpark. Las mezclas no son raras y no son automáticamente incorrectas, pero suelen ser computacionalmente costosas y deben evitarse en la medida de lo posible.
Las operaciones comunes de PySpark que pueden causar mezclas incluyen:
grupoPor()
unirse()
distinto()
ordenar por()
repartición()
Tome este ejemplo:
df.groupBy("customer_id").sum("cantidad")
Spark no puede calcular el total final para cada cliente a menos que se agreguen todas las filas de ese cliente. Normalmente eso significa mover datos entre particiones.
La clasificación es otro ejemplo común:
df.orderBy("fecha_transacción")
Ordenar un conjunto de datos completo requiere que PySpark compare y organice los datos en todo el conjunto de datos, no solo dentro de cada partición. Esa es otra operación aleatoria. La lección principal es:
Las mezclas son caras.
Los necesitará porque el trabajo real de procesamiento de datos depende de agrupar, unir y ordenar. Pero si un trabajo de PySpark parece lento, las mezclas son una de las primeras cosas a investigar.
Un hábito útil para reducir las mezclas es filtrar y seleccionar datos antes de operaciones costosas que podrían desencadenarlas.
En lugar de esto:
df_joined = df_sales.join(df_customers, "customer_id") df_london = df_joined.filter(df_joined.city == "Londres")
Prefiere esto, cuando sea posible:
df_london_customers = df_customers.filter(df_customers.city == "Londres") df_joined = df_sales.join(df_london_customers, "customer_id")
Eliminar datos innecesarios lo antes posible en sus canalizaciones significa que PySpark tiene menos datos para mover (barajar) durante la unión.
Por qué las uniones merecen un cuidado especial
Las uniones son donde muchos trabajos de PySpark comienzan a tener problemas de rendimiento. No porque las uniones sean malas, sino porque hacer coincidir filas entre conjuntos de datos a menudo significa mover datos.
Una unión simple se ve así:
df_joined = df_sales.join(df_customers, on="customer_id", how="izquierda")
Spark necesita que coincidan los ID de los clientes de los DataFrames de ventas y de clientes para encontrarse en el mismo lugar. Dependiendo del tamaño y diseño de los datos, eso puede implicar una mezcla. Hay tres cosas prácticas que puedes hacer para que las uniones sean menos dolorosas.
Primero, intente asegurarse de que sus claves de unión sean del mismo tipo de datos. Si eso no es posible, asegúrese de convertir los tipos de datos de forma adecuada. Si una tabla almacena los ID de los clientes como números enteros y otra los almacena como cadenas, PySpark puede producir errores o resultados inesperados.
desde pyspark.sql las funciones de importación son F df_sales = df_sales.withColumn( "customer_id", F.col("customer_id").cast("string") ) df_customers = df_customers.withColumn( "customer_id",F.col("customer_id").cast("string") )
En segundo lugar, conserve únicamente las columnas que necesita reduciendo los datos antes de unirse.
df_customers_small = df_customers.select("customer_id","ciudad","loyalty_level")
Y filtre temprano si es posible:
df_sales_recent = df_sales.filter(F.col("año") == 2026)
Entonces únete:
df_joined = df_sales_recent.join( df_customers_small,on="customer_id",how="izquierda")
En tercer lugar, considere transmitir pequeñas tablas de búsqueda. Si un DataFrame es muy pequeño, PySpark puede copiarlo a cada trabajador en lugar de mezclar ambos conjuntos de datos.
desde pyspark.sql.functions importar transmisión df_joined = df_sales.join(broadcast(df_customers_small),on="customer_id,"how="left" )
Esto se denomina unión de transmisión y es especialmente útil para tablas de búsqueda, como detalles de clientes, códigos de países, categorías de productos o tipos de cambio.
Una regla simple para recordar es esta:
Si un lado de la unión es lo suficientemente pequeño como para caber cómodamente en la memoria, una unión de difusión puede ser más rápida.
No fuerces las uniones de transmisión en todas partes. Son útiles cuando la mesa más pequeña es realmente pequeña. Intentar transmitir una tabla grande puede crear más problemas de los que resuelve.
Mirando lo que PySpark está planeando
Debido a que PySpark utiliza una evaluación diferida, no ejecuta cada transformación tan pronto como la lee. En cambio, elabora un plan sobre cómo se debe realizar el trabajo.
Puedes echarle un vistazo a ese plan con:
df.explain("formateado")
La primera vez que ve un plan de ejecución, el resultado puede resultar abrumador. Es posible que vea términos como:
Filtro de escaneo Intercambio de proyectos SortMergeJoin BroadcastHashJoin HashAggregate
No es necesario que entiendas cada línea. En esta etapa, simplemente comience a reconocer algunas pistas útiles.
Escanear significa que PySpark está leyendo datos. Filtro significa que PySpark está aplicando un filtro. Proyecto generalmente significa que PySpark está seleccionando columnas o creando columnas derivadas. El intercambio generalmente significa una mezcla. SortMergeJoin significa que PySpark une datos ordenando y haciendo coincidir claves. HashAggregate significa que PySpark agrupa filas y calcula valores agregados, como recuentos, sumas, promedios, mínimos o máximos. BroadcastHashJoin significa que PySpark está utilizando una unión de transmisión.
Al que yo le prestaría atención primero es:
Si ve esto, significa que PySpark está mezclando datos entre particiones. Nuevamente, eso no es automáticamente malo. Pero le indica dónde pueden estar algunos de los costos. Por ejemplo:
df_totals = df.groupBy("customer_id").sum("cantidad") df_totals.explain("formateado")
Es probable que vea un Exchange en el plan porque PySpark necesita agrupar los ID de clientes coincidentes.
También puede utilizar la interfaz de usuario de PySpark para investigar problemas, de los que hablé brevemente en mi artículo anterior de esta serie. Cuando PySpark se ejecuta localmente, normalmente puede abrir la interfaz de usuario de PySpark en esta URL:
http://localhost:4040
Le muestra qué trabajos se han ejecutado, cuánto tiempo llevó cada etapa y si una parte de su trabajo está tardando mucho más que el resto.
No es necesario que comprenda cada TAB en la salida de la interfaz de usuario de Spark en este momento. Simplemente acostúmbrate a abrirlo de vez en cuando para comprobar que tus trabajos avanzan como deberían.
Usar explica() y la interfaz de usuario de PySpark es uno de los mejores hábitos que puedes desarrollar. No es necesario convertirse en un experto interno de PySpark. Simplemente comience a buscar pistas obvias. Después de un tiempo, dejas de intentar leer cada línea y comienzas a buscar las pocas pistas que importan.
Almacenamiento en caché de datos
El almacenamiento en caché es tentador porque se siente como un botón mágico para "hacer esto más rápido". A veces lo hace. Otras veces, simplemente consume memoria sin ayudar mucho.
Puede marcar un DataFrame para almacenamiento en caché de esta manera:
df_clean.cache()
Pero recuerde: PySpark es vago. Esta línea no almacena en caché los datos inmediatamente. Solo le dice a PySpark:
Cuando se calcule este DataFrame, manténgalo en la memoria si es posible.
Para materializar realmente el caché, necesita una acción como:
df_clean.count()
Ahora PySpark calcula df_clean y lo almacena. El almacenamiento en caché generalmente solo es útil cuando se reutiliza el mismo DataFrame varias veces en la misma sesión.
Por ejemplo:
df_clean.cache() df_clean.count() df_by_city = df_clean.groupBy("ciudad").sum("cantidad") df_by_level = df_clean.groupBy("lealtad_level").sum("cantidad") df_sample = df_clean.filter(F.col("cantidad") > 1000)
En este caso, el almacenamiento en caché puede ayudar porque el mismo conjunto de datos limpio se reutiliza en varias operaciones posteriores, pero el almacenamiento en caché no es útil si solo usa el DataFrame una vez.
Por ejemplo, esta llamada de caché no tiene sentido:
df_clean.cache() df_final = df_clean.groupBy("ciudad").sum("cantidad")
Si df_clean no se reutiliza, el almacenamiento en caché solo agrega gastos generales.
Puede eliminar un DataFrame del caché con:
df_clean.unpersist()
Una buena regla general es:
Almacene en caché solo cuando un DataFrame es costoso de crear y se reutiliza más de una vez. No guardes todo en caché. La memoria PySpark es valiosa.
Escribir datos que PySpark pueda leer de manera eficiente
En mi artículo anterior sobre PySpark, presenté Parquet como el formato de archivo preferido para leer y escribir datos. Ahora podemos dar un paso más.
Puede escribir datos de Parquet particionados por una o más columnas:
df.write.mode("sobrescribir") .partitionBy("año", "mes") .parquet("producción/ventas")
Dependiendo de los datos en el DataFrame, esto crearía un diseño de carpeta de salida similar a este:
producción/ventas/ -> año=2025/ -> mes=1/ -> mes=5/ -> mes=12/ -> año=2026/ -> mes=1/ -> mes=2/
La parte útil es que al volver a leer datos desde una estructura de carpetas como esta, Spark no siempre tiene que leer todo el conjunto de datos.
Si luego lees solo los datos de 2026:
df_2026 = spark.read.parquet("producción/ventas").filter(F.col("año") == 2026)
Spark a menudo puede evitar por completo leer la carpeta 2025. Esto se llama poda de partición y puede hacer que las lecturas sean mucho más rápidas cuando los filtros coinciden con las columnas de la partición.
Las buenas columnas de partición suelen ser campos por los que se filtra habitualmente, como por ejemplo:
año mes semana región país departamento
Pero ten cuidado. No particione por una columna con demasiados valores únicos.
Por ejemplo, esto suele ser una mala idea:
.partitionBy("id_transacción")
Si cada ID de transacción es única, PySpark puede crear miles o millones de pequeñas carpetas y archivos. Eso hace que el conjunto de datos sea más difícil de gestionar y más lento de leer.
Una buena regla para recordar es:
Divida por columnas que filtra con frecuencia, pero evite columnas con demasiados valores distintos. La partición basada en fechas suele ser un buen punto de partida.
Dejando que PySpark haga el trabajo
En PySpark, una UDF significa una función definida por el usuario. Le permite escribir su propia función de Python y aplicarla a una columna de PySpark en un Dataframe.
Esto es realmente útil en algunos casos. Pero las UDF de Python pueden ser más lentas que las funciones integradas de PySpark.
A continuación se muestra un ejemplo sencillo de versión integrada en el que queremos convertir parte del texto a mayúsculas:
desde pyspark.sql funciones de importación como F df = df.withColumn("customer_name_upper",F.upper("customer_name"))
Spark comprende F.upper() y puede optimizarlo como parte del plan de ejecución.
Ahora compárelo con una UDF de Python:
de pyspark.sql.functions importar udf de pyspark.sql.types importar StringType def make_upper(valor): si el valor es Ninguno: devolver Ninguno devolver value.upper() make_upper_udf = udf(make_upper, StringType()) df = df.withColumn("customer_name_upper", make_upper_udf("customer_name"))
Esto funciona, pero es más detallado y más difícil de entender y, lo que es más importante, PySpark tiene menos capacidad para optimizarlo. Es posible que los datos deban moverse entre el motor de PySpark y Python, lo que agrega gastos generales.
Entonces el consejo práctico de nivel intermedio es:
Elija primero las funciones integradas de PySpark. Utilice las UDF con moderación y sólo cuando no haya una función integrada que haga lo que necesita.
Trampas comunes a medida que crecen sus datos
Una vez que comienza a trabajar con conjuntos de datos más grandes, algún código PySpark de apariencia inocente puede causar problemas. Una trampa común es utilizar Collect() en datos de gran tamaño.
filas = df.collect()
Esto trae todos los datos de PySpark a Python en su máquina controladora.
Para datos pequeños, eso está bien. Para conjuntos de datos grandes, esto puede abrumar al controlador porque todas las filas se cargan en la memoria local de Python.
En su lugar, utilice:
df.mostrar()
Para ver cómo se ven sus datos antes de decidir qué hacer con ellos. Ver el conjunto de datos puede ayudarle a formular un plan para filtrarlo con mayor precisión o procesarlo de manera más óptima.
Otra trampa es llamar a count() con demasiada frecuencia.
df.count()
Al igual que show(), count() es una acción. Hace que PySpark escanee todo el conjunto de datos. Puede resultar útil, pero no lo utilices en todas partes sin pensarlo.
Clasificar conjuntos de datos enormes también puede resultar caro.
df.orderBy("cantidad")
La clasificación global a menudo requiere una mezcla. Si solo necesita un vistazo rápido a las filas más grandes, puede usar esto en su lugar:
df.orderBy(F.col("cantidad").desc()).limit(10)
Spark a menudo puede optimizar esto como una consulta top-k. Aún necesita escanear los datos, pero puede evitar ordenar y devolver todo el conjunto de datos.
Otro problema común es unir datos antes de filtrarlos. Siempre que sea posible, filtre cada conjunto de datos primero, de modo que Spark tenga menos filas para mover y comparar durante la unión.
Escribir cientos o miles de archivos pequeños también puede ralentizar lecturas futuras. Cuando la salida es lo suficientemente pequeña, usar coalesce() puede reducir la cantidad de archivos que escribe Spark:
df.coalesce(8).write.mode("sobrescribir").parquet("salida/resultados")
Y finalmente, evite, evite, evite el uso de bucles de Python donde una operación de DataFrame funcionaría mejor. Este patrón puede desencadenar muchos trabajos de PySpark separados:
para ciudad en ciudades: df.filter(F.col("ciudad") == ciudad).count()
Esto es mejor:
df.groupBy("ciudad").count()
Spark se siente más feliz cuando describe el resultado que desea y le deja planificar la ruta.
Juntando las piezas
Reunamos las ideas que hemos discutido en un flujo de trabajo simple. Imaginemos que tenemos datos de ventas almacenados como Parquet particionado, además de una pequeña tabla de búsqueda de clientes.
Queremos:
leer las ventas recientes mantener solo las columnas útiles limpiar las claves de unión unirse a los datos del cliente calcular los totales escribir la salida dividida inspeccionar el plan
Así es como podría verse en el código PySpark:
de pyspark.sql importar funciones como F de pyspark.sql.functions importar transmisiones de ventas = spark.read.parquet("data/sales") sales_recent = (sales .filter(F.col("año") == 2026) .select( "transaction_id", "customer_id", "amount", "year", "month", ).dropna(subset=["customer_id", "amount"]) .withColumn("customer_id", F.col("customer_id").cast("string")) ) clientes = ( spark.read.parquet("data/customers") .select("customer_id", "city", "loyalty_level") .withColumn("customer_id", F.col("customer_id").cast("string")) ) enriched = sales_recent.join( broadcast(clientes), on="id_cliente", cómo="izquierda", ) totales_mensuales = ( enriquecido .groupBy("año", "mes", "ciudad") .agg( F.sum("cantidad").alias("cantidad_total"), F.count("*").alias("cuenta_transacciones"), ) ) totales_mensuales.explain("formateado") (totales_mensuales.write .mode("sobrescribir") .partitionBy("año", "mes") .parquet("salida/totales_mensuales") )
No hay nada llamativo aquí. Pero este es un flujo de trabajo de PySpark mucho más maduro que simplemente cargar un CSV y ejecutar una unión.
¿Cómo es el plan de ejecución?
Dado que este flujo de trabajo utiliza una unión de transmisión y una agregación, también es un buen momento para observar el plan que Spark podría crear. El plan exacto depende de su versión de PySpark, el tamaño de los datos, el diseño del archivo y si PySpark decide respetar la sugerencia de transmisión. Pero para este ejemplo, la versión simplificada del plan podría verse así:
== Plan físico == AdaptiveSparkPlan +- WriteFiles +- Ordenar +- HashAggregate +- Exchange +- HashAggregate +- Proyecto +- BroadcastHashJoin LeftOuter :- Proyecto : +- Filtro : +- Escanear datos/ventas de parquet +- BroadcastExchange +- Proyecto +- Escanear datos/clientes de parquet
Los aspectos clave a tener en cuenta son:
BroadcastHashJoin: esto le indica que PySpark está utilizando la unión de transmisión, lo cual normalmente es algo bueno.
Exchange: esto le indica que PySpark está realizando una mezcla. En este flujo de trabajo, se espera la reproducción aleatoria debido a:
.groupBy("año", "mes", "ciudad")
Filtrar: y si sus datos de ventas están divididos por año, también puede ver archivos de poda de PySpark cuando filtra:
.filtro(F.col("año") == 2026)
Eso significa que PySpark puede omitir carpetas irrelevantes en lugar de leer todo el conjunto de datos.
Resumen
Pasar a un nivel intermedio en PySpark se trata menos de memorizar configuraciones y más de comprender las ideas que explican cómo se comporta Spark:
Los datos se dividen en particiones. Algunas operaciones se ejecutan de forma independiente, mientras que otras requieren mezclas. Las combinaciones aleatorias y las uniones pueden resultar costosas. El almacenamiento en caché sólo ayuda cuando se reutilizan los datos. El diseño del archivo Parquet afecta el rendimiento de lectura y escritura. Las funciones integradas suelen ser más rápidas que las UDF de Python. Los planes de ejecución muestran cómo Spark pretende procesar sus datos.
Una vez que comprenda estas ideas, los trabajos lentos serán más fáciles de diagnosticar. Ese es el verdadero paso adelante.
Su primer hito en PySpark fue aprender a ejecutar código. Lo siguiente fue aprender a crear flujos de trabajo limpios. Este tercer hito es aprender cómo hacer que esos flujos de trabajo escale.
No es necesario convertirse en ingeniero de rendimiento de PySpark de la noche a la mañana. Pero si puedes reconocer particiones, mezclas, uniones, almacenamiento en caché, diseño de archivos y planes de ejecución, ya estás escribiendo PySpark con una mentalidad mucho más profesional.
Como referencia, aquí están los enlaces a los dos primeros artículos de esta serie.