Cómo utilizar eficazmente las UDFs de PySpark y las UDFs de Pandas

Las Funciones Definidas por el Usuario (UDFs) en PySpark ofrecen a los programadores de Python una forma potente de manejar tareas únicas que las funciones integradas de Spark simplemente no pueden gestionar. Si eres un ingeniero de datos, un analista o un científico que domina Python, comprender los conceptos de las UDF puede permitirte afrontar con eficacia los complejos retos que plantean los datos en el mundo real. 

Este tutorial te guía a través de los conceptos de PySpark UDF, implementaciones prácticas y mejores prácticas de optimización, pruebas, depuración y patrones de uso avanzados. Al final, serás capaz de escribir, optimizar y desplegar con confianza UDFs eficientes a escala.

Si eres nuevo en PySpark, te recomiendo que primero eches un vistazo a nuestro Introducción a PySpark ya que lo que cubrimos aquí son conceptos avanzados de Spark.

¿Qué son las UDF de PySpark?

Las UDF de PySpark son funciones personalizadas de Python integradas en el marco distribuido de Spark para operar con datos almacenados en DataFrames de Spark. A diferencia de las funciones integradas de Spark, las UDF permiten a los programadores aplicar una lógica compleja y personalizada a nivel de fila o columna. 

Nuestro Hoja de trucos de PySpark cubre todo lo que necesitas saber sobre los DataFrames de Spark, facilitándote aún más la comprensión de las UDF de Spark.

¿Cuándo debes utilizar las UDFs de PySpark?

Utiliza UDFs cuando:

• Necesitas una lógica que no puede expresarse utilizando las funciones integradas de Spark.
• Tu transformación implica operaciones complejas nativas de Python (por ejemplo, manipulaciones regex, lógica NLP personalizada).
• Te parece bien cambiar rendimiento por flexibilidad, especialmente durante la creación de prototipos o para conjuntos de datos pequeños o medianos.

Evita las UDFs cuando:

• Existe una funcionalidad equivalente en pyspark.sql.functions, las funciones nativas de Spark son más rápidas, están optimizadas y se pueden transferir al motor de ejecución.
• Trabajas con grandes conjuntos de datos en los que el rendimiento es fundamental. Las UDF introducen sobrecarga de serialización y rompen la capacidad de Spark para optimizar los planes de ejecución.
• Puedes expresar tu lógica utilizando expresiones SQL, Spark SQL built-ins, o Pandas UDFs (para operaciones vectorizadas).

Aplicaciones estratégicas en ingeniería de datos

Éstos son los principales casos de uso de las UDF de PySpark:

• Transformaciones complejas de datos, como el análisis sintáctico avanzado de textos, la extracción de datos o la manipulación de cadenas.
• Integración con bibliotecas Python de terceros, incluidos los populares marcos de machine learning como TensorFlow y XGBoost.
• Uniendo los sistemas heredados y soportando la evolución sin fisuras de los esquemas a medida que cambian las estructuras de datos.

Las UDF simplifican las complicadas tareas de ingeniería de datos del mundo real, capacitando a los equipos para gestionar diversos requisitos de forma flexible y eficaz. 

Descubramos ahora cómo puedes implementar las UDF de PySpark. 

Implementación de las UDF de PySpark

Esta sección describe cómo definir e implementar en la práctica las FDU de PySpark.

Métodos estándar de declaración UDF

Hay tres enfoques comunes para declarar UDFs en PySpark:

1. UDFs basadas en Lambda: Rápido de definir directamente dentro de las consultas del DataFrame; lo mejor para operaciones sencillas.

La UDF basada en Lambda (UDF básica de Python) es la mejor para transformaciones rápidas y sencillas. Evítalos para trabajos a gran escala en los que el rendimiento sea importante.

He aquí un ejemplo: 

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

uppercase = udf(lambda x: x.upper() if x else None, StringType())
df = spark.createDataFrame([("Ada",), (None,)], ["name"])
df.withColumn("upper_name", uppercase("name")).show()

2. Funciones Python decoradas: Anotado explícitamente mediante @pyspark.sql.functions.udf, lo que favorece la reutilización y la legibilidad.

Por ejemplo:

from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType

@F.udf(returnType=IntegerType())
def str_length(s):
    return len(s) if s else 0

df.withColumn("name_length", str_length("name")).show()

3. UDFs registradas en SQL: Se registran directamente en contextos SQL de Spark, lo que permite su uso dentro de consultas SQL.

from pyspark.sql.types import StringType

def reverse_string(s):
    return s[::-1] if s else ""

spark.udf.register("reverse_udf", reverse_string, StringType())

df.createOrReplaceTempView("people")
spark.sql("""SELECT name, reverse_udf(name) AS reversed FROM people""").show()

Cada método tiene sus ventajas y sus inconvenientes: las UDF lambda son concisas pero limitadas, mientras que las anotaciones de funciones favorecen la legibilidad, la mantenibilidad y las mejores prácticas.

Las UDFs de Pandas permiten realizar operaciones vectorizadas en lotes de Flechas. Suelen ser más rápidas que las UDF normales y se integran mejor con el motor de ejecución de Spark.

Scalar Pandas UDF (por elementos, como map)

Son los más adecuados para las transformaciones rápidas por filas en grandes conjuntos de datos. Por ejemplo: 

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import IntegerType
import pandas as pd

@pandas_udf(IntegerType())
def pandas_strlen(s: pd.Series) -> pd.Series:
    return s.str.len()

df.withColumn("name_len", pandas_strlen("name")).show()

Mapa agrupado Pandas UDF

Son mejores para la lógica personalizada por grupo, similar a groupby().apply() en Pandas.

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType
import pandas as pd

schema = StructType([
    StructField("group", StringType()),
    StructField("avg_val", DoubleType())
])

@pandas_udf(schema, F.PandasUDFType.GROUPED_MAP)
def group_avg(pdf: pd.DataFrame) -> pd.DataFrame:
    return pd.DataFrame({
        "group": [pdf["group"].iloc[0]],
        "avg_val": [pdf["value"].mean()]
    })

df.groupBy("group").apply(group_avg).show()

UDF Agregado Pandas

Éste realiza agregaciones sobre grupos, más rápido que el mapa agrupado. Por ejemplo: 

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType
import pandas as pd

@pandas_udf(DoubleType(), functionType="grouped_agg")
def mean_udf(v: pd.Series) -> float:
    return v.mean()

df.groupBy("category").agg(mean_udf("value").alias("mean_value")).show()

Iterador UDF de Pandas

La UDF Iterador de Pandas es la mejor para grandes conjuntos de datos que requieren un procesamiento de baja memoria (por lotes). Por ejemplo: 

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import IntegerType
from typing import Iterator
import pandas as pd

@pandas_udf(IntegerType(), functionType="iterator")
def batch_sum(it: Iterator[pd.Series]) -> Iterator[pd.Series]:
    for batch in it:
        yield batch + 1

df.withColumn("incremented", batch_sum("id")).show()

Manipulación de tipos y seguridad nula

Los tipos y los valores nulos plantean retos frecuentes para las UDF de PySpark. PySpark aplica una comprobación de tipos estricta, lo que a menudo provoca conversiones de tipos implícitas o problemas en tiempo de ejecución. Además, Spark pasa valores nulos directamente a las UDF, lo que puede provocar fallos si no se gestionan explícitamente.

Garantiza la solidez de los FDU con estas estrategias:

• Especifica explícitamente los tipos de retorno.
• Incorpora comprobaciones de nulos (por ejemplo, sentencias condicionales) dentro de tus funciones Python.
• Adopta prácticas de codificación defensivas; simples comprobaciones de nulos evitan frustrantes excepciones en tiempo de ejecución.

Optimizar el rendimiento de la UDF

El rendimiento suele ser el talón de Aquiles de las UDF estándar, debido a su modelo de ejecución fila a fila. Aprovechar las UDF vectorizadas y las herramientas de optimización de Spark mejorará significativamente los tiempos de ejecución.

UDFs vectorizados con integración de Pandas

Las UDFs de Pandas introducen un enfoque vectorizado de las UDFs en PySpark, pasando lotes de datos como Series de Pandas a funciones de Python. Este diseño mejora significativamente el rendimiento al reducir la sobrecarga de serialización en comparación con las UDF estándar basadas en filas.

Con el respaldo de Apache Arrow para la transferencia de datos sin copia entre la JVM y los procesos de Python, las UDF de Pandas permiten la ejecución eficiente de operaciones a escala. Son especialmente eficaces para cálculos intensivos y manipulaciones complejas de cadenas en millones de registros. 

Cubrimos más detalles sobre la manipulación de datos con PySpark en nuestro curso Curso de Limpieza de Datos con PySpark.

Además, las UDF de Pandas permiten una integración perfecta con el ecosistema más amplio de ciencia de datos de Python, aprovechando herramientas y flujos de trabajo conocidos.

Técnicas de optimización de flechas

El formato de memoria columnar de Apache Arrow permite una transferencia de datos eficiente y sin copias entre la JVM de Spark y los procesos de Python. Al activar Arrow (spark.sql.execution.arrow.pyspark.enabled=true) en tus configuraciones de Spark, los datos se mueven rápidamente entre los entornos JVM y Python, lo que acelera considerablemente la ejecución de las UDF.

Optimización del plan de ejecución

Optimizar los trabajos de PySpark implica entender cómo influir en el optimizador Catalyst de Spark. Las estrategias avanzadas incluyen técnicas como el pushdown de predicados, la poda de columnas y el uso de sugerencias de unión de difusión para mejorar la planificación de las consultas y la eficacia de la ejecución.

Para maximizar el rendimiento, es importante minimizar el ámbito de ejecución de las UDF y favorecer las funciones SQL integradas en Spark siempre que sea posible. El uso estratégico de la memoria caché y la elaboración cuidadosa del plan pueden mejorar aún más la velocidad de ejecución del trabajo y la utilización de los recursos.

La optimización del rendimiento es una de las principales preguntas que puedes encontrarte en una entrevista de PySpark. Descubre cómo responder a ésta y más preguntas sobre Spark en nuestro Las 36 mejores preguntas y respuestas de PySpark para 2025 entrada del blog.

Patrones y antipatrones avanzados

Comprender los patrones de uso adecuados e inadecuados ayuda a garantizar despliegues UDF estables y eficientes.

Implementaciones UDF con estado

Las UDFs con estado y no deterministas presentan retos únicos en PySpark. Estas funciones producen resultados que dependen de un estado externo o de condiciones cambiantes, como las variables de entorno, el tiempo del sistema o el contexto de la sesión.

Aunque las UDF no deterministas a veces son necesarias -por ejemplo, para generar marcas de tiempo, rastrear sesiones de usuario o introducir aleatoriedad-, pueden complicar la depuración, la reproducibilidad y la optimización.

Implementar UDFs con estado requiere patrones de diseño cuidadosos: documentar el comportamiento con claridad, aislar los efectos secundarios y añadir un registro exhaustivo para ayudar a solucionar los problemas y garantizar la coherencia entre las ejecuciones de los trabajos.

Cuando se utilizan con cuidado, pueden desbloquear potentes capacidades, pero mantener canalizaciones de datos fiables requiere una gestión disciplinada. Nuestro Fundamentos de Big Data con PySpark entra en más detalle sobre cómo manejar big data en PySpark.

Antipatrones comunes

Los anti-patrones comunes en el uso de UDF pueden degradar significativamente el rendimiento de PySpark:

• Procesamiento por filas en lugar de por lotes: Aplicar UDFs a filas individuales en lugar de utilizar enfoques vectorizados como los UDFs de Pandas provoca importantes ralentizaciones en la ejecución.
• Operaciones anidadas de DataFrame dentro de UDFs: Incrustar consultas DataFrame dentro de UDFs provoca un cálculo excesivo y dificulta la capacidad de Spark para optimizar los planes de ejecución.
• Registro repetido de UDF en línea: Definir y registrar las UDF varias veces dentro de las consultas añade una sobrecarga innecesaria; es mejor declarar las UDF una vez y reutilizarlas en todos los trabajos.
• Uso excesivo de lógica Python personalizada para operaciones sencillas: Tareas como el filtrado básico, la aritmética o las transformaciones sencillas deberían preferir las funciones integradas de Spark, altamente optimizadas, a las UDF personalizadas.

Evitar estas trampas garantiza un mejor rendimiento, una optimización más sencilla por parte de Catalyst y un código PySpark más fácil de mantener.

Depurar y probar las UDFs de PySpark

Probar y depurar las UDF garantiza su fiabilidad y solidez en escenarios de producción.

Patrones de manejo de excepciones

Implementar la captura estructurada de errores dentro de las UDFs es esencial para construir pipelines PySpark resistentes y mantenibles. Utiliza bloques try-except dentro de las UDFs para manejar con elegancia las sorpresas en tiempo de ejecución, como valores nulos, desajustes de tipo o errores de división por cero.

Un sólido tratamiento de las excepciones estabiliza los procesos frente a datos inesperados y simplifica la depuración, mostrando mensajes de error claros y procesables. Las excepciones correctamente capturadas y registradas hacen que el comportamiento de la UDF sea más transparente, acelerando la resolución de problemas y mejorando la fiabilidad general de la tubería.

Marcos de pruebas unitarias

Utiliza la clase base de pruebas incorporada en PySpark, pyspark.testing.utils.ReusedPySparkTestCase, junto con frameworks como pytest, para escribir pruebas unitarias fiables para tus UDFs. Estructurar pruebas claras y centradas garantiza la corrección, estabilidad y mantenibilidad de tu lógica UDF a lo largo del tiempo.

Las mejores prácticas para probar las FDU incluyen cubrir tanto los casos típicos como los extremos, validar las salidas con resultados conocidos y aislar el comportamiento de las FDU de las dependencias externas. Las pruebas bien diseñadas no sólo protegen contra las regresiones, sino que también simplifican los futuros esfuerzos de desarrollo y refactorización.

Evolución y orientaciones futuras

El ecosistema PySpark sigue evolucionando rápidamente, introduciendo nuevas capacidades que mejoran aún más las UDF.

Integración del Catálogo Unity

Los desarrollos recientes han integrado el registro UDF en Catálogo Unityagilizando la gestión, el descubrimiento y el gobierno de las UDF a gran escala. Unity Catalog permite un control centralizado de la gestión del ciclo de vida de las UDF, incluidos el registro, el control de versiones y el control de acceso, todos ellos aspectos críticos para los entornos empresariales.

Esta integración mejora la gobernanza, impone políticas de seguridad coherentes y mejora la capacidad de descubrimiento en todos los equipos, haciendo que los UDF sean más fáciles de reutilizar, auditar y gestionar dentro de ecosistemas de datos grandes y complejos.

UDF aceleradas en la GPU

Los marcos como el Acelerador RAPIDS permiten la descarga de la GPU para tareas UDF de cálculo intensivo en PySpark, ofreciendo mejoras de rendimiento transformadoras. Al trasladar las operaciones pesadas, como el análisis numérico, la inferencia de aprendizaje profundo y el modelado de datos a gran escala, a las GPU, RAPIDS puede reducir los tiempos de ejecución de horas a minutos en cargas de trabajo adecuadas.

La aceleración en la GPU es especialmente beneficiosa para escenarios que implican conjuntos de datos masivos, cálculos vectorizados complejos y pipelines de machine learning, ampliando drásticamente el rendimiento y la escalabilidad de PySpark para tareas modernas de ingeniería de datos. Nuestro curso de Machine Learning con PySpark profundiza en estos conceptos.

Conclusión

Las UDFs de PySpark son una potente herramienta para ampliar las capacidades de Spark, permitiendo a los equipos abordar tareas complejas y personalizadas de procesamiento de datos que van más allá de las funciones incorporadas. Cuando se aplican correctamente, desbloquean la flexibilidad y la innovación en las canalizaciones de datos a gran escala. 

Sin embargo, optimizar el rendimiento de las UDF requiere una atención cuidadosa, evitando errores comunes como las operaciones por filas, gestionando las excepciones con elegancia y aprovechando técnicas como la vectorización de UDF de Pandas con integración de Arrow.

Los nuevos avances, como la aceleración de la GPU a través de marcos como RAPIDS, están ampliando aún más las posibilidades de los flujos de trabajo basados en UDF. Tanto si transformas datos desordenados del mundo real como si integras análisis avanzados en sistemas de producción, dominar las mejores prácticas de UDF es esencial para construir canalizaciones de datos rápidas, eficientes y fiables. 

Aprende los detalles a los que los científicos de datos dedican el 70-80% de su tiempo, la gestión de datos y la ingeniería de características, en nuestro curso de Curso de Ingeniería de Características con PySpark.