UDF

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|
+---------+-------+

interesante

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)

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?

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.