Creación y uso de UDFs en PySpark para manejo de datos faltantes

Clase 21 de 25 • Curso de Fundamentos de Spark para Big Data

Resumen

¿Cómo crear una UDF en PySpark?

¡Bienvenido al fascinante mundo de Spark! En esta clase, aprenderemos a crear nuestra primera función definida por el usuario (UDF) en PySpark. Una UDF nos permite manejar datos complejos y realizar transformaciones personalizadas directamente en los data frames. Acompáñame en este viaje para descubrir cómo manejar datos erróneos y transformarlos eficazmente con una UDF.

¿Cuál es el primer paso para crear una UDF?

Antes de empezar, asegúrate de tener tus datos listos. En nuestro caso, estamos trabajando con un archivo CSV llamado deportista_error.csv. Observamos que este archivo tiene valores faltantes, lo que podría causar problemas al procesarlo. Aquí es donde entra en juego la UDF: nos ayudará a gestionar estos datos erróneos.

Carga del RDD: Primero, cargamos el archivo y eliminamos encabezados.

deportista_error = SparkContext.textFile("ruta/deportista_error.csv")
rdd_sin_encabezado = deportista_error.mapPartitionsWithIndex(
    lambda idx, it: iter(list(it)[1:]) if idx == 0 else it
)

Definición del esquema: Especificamos las columnas y sus tipos de datos.

esquema = StructType([
    StructField("id", StringType(), False),
    StructField("nombre", StringType(), False),
    StructField("genero", StringType(), False),
    StructField("edad", StringType(), False),
    StructField("altura", StringType(), True),
    StructField("peso", StringType(), True),
    StructField("equipo_id", StringType(), False),
])

¿Cómo definir una UDF para transformar datos?

Las UDF se definen mediante funciones de Python que luego se registran como funciones en Spark. Nuestra UDF convertirá campos numéricos erróneos en valor nulo (None) si están vacíos.

Importación necesaria: Importamos desde pyspark.sql.functions.

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

Creación de la UDF: Definimos una función de Python y la registramos.

def convertir_a_entero(valor):
    return int(valor) if valor and len(valor) > 0 else None

convertir_a_entero_udf = udf(convertir_a_entero, IntegerType())

Registro de la UDF: Con el SQLContext, registramos la UDF.

sqlContext.udf.register("convertir_a_entero", convertir_a_entero_udf)

¿Cómo aplicar la UDF a un DataFrame?

Una vez creada y registrada nuestra UDF, es hora de aplicarla a datos reales. Este proceso transforma entradas erróneas o vacías al formato deseado.

Creación del DataFrame: Convertimos el RDD al DataFrame usando el esquema.

df_deportista_error = sqlContext.createDataFrame(rdd_sin_encabezado, esquema)

Aplicación de la UDF: Utilizando select(), transformamos datos específicos.

df_transformado = df_deportista_error.select(
    "id", "nombre", "genero", "edad",
    convertir_a_entero_udf("altura").alias("altura"),
    convertir_a_entero_udf("peso").alias("peso"),
    "equipo_id"
)

Visualización de resultados: Esto nos permite ver cómo se eliminaron los errores y se corrigieron los datos.
```
df_transformado.show()
```

¡Y ahí lo tienes! Has aprendido a manejar datos erróneos mediante una UDF personalizada. La correcta comprensión y uso de UDFs te permitirá realizar transformaciones complejas y vitales en tus datasets. Te animamos a seguir explorando Spark y a profundizar en cómo estas funciones pueden optimizar el trabajo con grandes volúmenes de datos. ¡Hasta la próxima!

Jose Daniel Gallegos Padilla

student•

En mi caso como cargue deportistaError usando directamente el csv de la siguiente manera:

DeportistaError_schema = StructType([
    StructField("deportista_id", StringType(), False),
    StructField("nombre", StringType(), False),
    StructField("genero", StringType(), False),
    StructField("edad", StringType(), False),
    StructField("altura", StringType(), False), 
    StructField("peso", StringType(), False),
    StructField("equipo_id", StringType(), False)
])

DeportistaErrorDF = sqlContext.read.format("csv").\
            option("header", True).\
            schema(DeportistaError_schema).\
            load(path+"deportistaError.csv")

Entonces, me salian las alturas faltantes como "null" por lo que modifique la funcion udf para que en vez de poner None, me ponga un 0. Tambien agregue una function que converte el valor a tipo flotante.

def conversionEnteros(valor):
    return int(valor) if valor else 0

def conversionFlotante(valor):
    return float(valor) if valor else 0.0

conversionEnteros_udf = udf(lambda z: conversionEnteros(z), IntegerType())
sqlContext.udf.register("conversionEnteros_udf", conversionEnteros_udf)

conversionFlotante_udf = udf(lambda z: conversionFlotante(z), FloatType())
sqlContext.udf.register("conversionFlotante_udf", conversionFlotante_udf)

DeportistaErrorDF.select(conversionEnteros_udf("altura").alias("alturaUDF"),\
                        conversionFlotante_udf("peso").alias("pesoUDF")).show(10)

+---------+-------+
|alturaUDF|pesoUDF|
+---------+-------+
|      180|   80.0|
|      170|   60.0|
|        0|    0.0|
|        0|    0.0|
|      185|   82.0|
|      188|   75.0|
|      183|   72.0|
|      168|    0.0|
|      186|   96.0|
|        0|    0.0|
+---------+-------+

rusbel bermúdez rivera

student•

A mi la libreria py4j me genera un error, me parece mas eficiente la forma en que lo manejaste

rusbel bermúdez rivera

student•

En mi caso muy particular utilizando python 3.7.9, wls2 y siguiendo los pasos para crear el DataFrame como en la clase me tope con un error, este es del tipo TypeError, en mi notebook los datos “vacíos” de altura al pasarlos por la función conversionEnteros() levantaba un error “TypeError: object of type ‘NoneType’ has no len()”, lo solucione manejando la excepción con un bloque try/catch.

def conversionEnteros(valor):
    try:
        return int(valor) if len(valor) > 0 else 0
    except TypeError as e:
        return 0
    
conversionEnteros_udf = udf(lambda z: conversionEnteros(z), IntegerType())
sqlContext.udf.register("conversionEnteros_udf", conversionEnteros_udf)

DeportistaErrorDF.select(conversionEnteros_udf("altura")
                         .alias("alturaUDF")).show()

Para el aporte de Jose Daniel Gallegos (eres un crack!) funciona de maravilla al importarlo directamente desde el csv se hacen las conversiones de tipos de datos de manera automática a diferencia de la versión RDD.

Gustavo Peralta Guzman

student•

Para los que usen colab y no puedan registrar la UDF con el método del curso, les dejo la siguiente manera:

#CREAR UDF PARA QUITAR ESPACIOS Y AGREGAR NULL
from pyspark.sql.functions import udf

def conversionEnteros(valor):
  return int(valor) if len(valor) > 0 else None

#Z es comun al registrar una udf
conversionEnteros_udf = udf(lambda z: conversionEnteros(z), IntegerType())
#Dar de alta UDF

spark.udf.register("conversionEnteros_udf", conversionEnteros_udf)

Carlos Javier Guevara Contreras

student•

interesante

Gerardo Miguel Pérez Solis

student•

Hice un udf para cambiar el genero del atleta de 1 y 2 a MALE y FEMALE.

from pyspark.sql.functions import udf
#CREO MI FUNCIÓN PERSONALIZADA
def select_gen(s):
    return "MALE" if s == 1 else "FEMALE" 

#LA VUELVO UN UDF PARA USARLA CON DATAFRAMES Y LA REGISTRO PARA USARLA COMO SQL
select_gen_udf = udf(select_gen,StringType())
spark.udf.register("select_gen_udf",select_gen_udf)

deportista_error_df.select("nombre",select_gen_udf("genero").alias("Genero")).show(5)

spark.sql("SELECT nombre,select_gen_udf(genero) AS GENERO FROM deportista_error").show(5)

No utilicé la parte de lambda del profesor, ya que basandome en este url ví que no lo usaba. ¿Alguién sabe cuál es la diferencia de hacerlo con el lambda?

Cesar Alejandro Rodriguez Muñoz

student•

bro pero te debe marcar error ya que pusiste == 1 y es == '1' caracter

Edith Montes Polanco

student•

las udf se pueden quedar almacenadas para reutilizar en otras notebooks o en cada una es necesario volverla a generar y guardar?

Diego Buesaquillo

student•

Sólo quedan disponibles en el notebook del proyecto.

David Marín García

student•

Se puede mapear un DataFrame??????

rogelio cortez

student•

Hola comunidad!

Actualmente es mas sencillo cargar el df con error con una spark session ya pone nulos los valores vacíos automáticamente , a continuacion muestro mi codigo:

# Create a SparkSession
spark = SparkSession.builder.appName("DataFrameExample").getOrCreate()
```# Create a SparkSessionspark = SparkSession.builder.appName("DataFrameExample").getOrCreate()



```js
deportistaError = spark.read.csv(path + '/deportistaError.csv',header = True)
```deportistaError = spark.read.csv(path + '/deportistaError.csv',header = True)

Diana Catalina Velásquez Gaviria

student•

SPOILER : Solo a partir del minuto 8 se habla sobre las UDF .

Pablo Rosa

student•

Alguien me puede explicar porque en la función lambda le pasa el parámetro IntegerType, siendo que ya la función conversionEnteros el return es un entero.

NAHUEL GABIOUD FERNANDEZ

student•

Entiendo que convierte el entero de Python a entero de Spark, pero no estoy seguro.

Alfonso Andres Zapata Guzman

student•

al lambda no se le esta pasando ese parametro, si quieres verlo claramente seria asi:

nuestra_funcion_lambda = lambda z : ci(z)

ci_udf = udf(nuestra_funcion_lambda, IntegerType())

IntegerType() se le esta pasando es a la funcion udf(), y esto se denota porque hay una coma "," y ademas estan en un parentesis, IntegerType() es indicativo del tipo de valor que esperamos obtener.