Puntos clave: conoce a escribir Funciones Definidas por el Usuario (UDFs) en Python para Apache Spark para mejorar el procesamiento de datos. El tutorial explica tres métodos: usar withColumn(), sentencias SQL y transformaciones de mapas RDD.
- Los UDF Permite registrar y aplicar funciones personalizadas a los dataframes de Spark.
- Los ejemplos de código ilustran eficazmente la implementación de cada método.
En este artículo, explicaré cómo escribir funciones definidas por el usuario (UDF) en Python para Apache Spark. El código de este ejemplo está aquí.
¿Por qué te necesitan UDFs?
Spark almacena datos en dataframes o RDDs—conjuntos de datos distribuidos resilientes. Piensa en esto como bases de datos. Como en una base de datos SQL tradicional, por ejemplo mySQL, no te puedes Crea tu propia función personalizada y ejecutarla directamente contra la base de datos. te tienes que registrar primero la función. Eso están, guárdala en la base de datos como si fuera una de las funciones integradas, como sum(), promedio, conteo(), etc.
Ese es el caso de los dataframes de Spark. Con los RDDs de Spark te puedes ejecutar funciones directamente sobre las filas de un RDD.
Tres enfoques para las UDF
Son tres formas de Crear UDFs:
- df = df.conColumna
- df = sqlContext.sql("sentencia SQL de <df>")
- rdd.map(customFunction())
Mostramos los tres enfoques a continuación, empezando por el primero.
Enfoque 1: withColumn()
A continuación, Crea un dataframe simple y RDD. Escribimos una función para convertir el único campo de texto de la estructura de datos a un entero. Eso están algo que te podría hacer si, por ejemplo, te son trabaja con aprendizaje automático, donde todos los datos deben convertirse en números antes de te introducir eso en un algoritmo.
Fíjate en las importaciones a continuación. Consulta las de cada ejemplo para te saber qué objeto importar para cada uno de los tres enfoques.
A continuación están el código completo del Enfoque 1. Primero, analizamos secciones clave. Crea un dataframe usando el enfoque habitual:
df = spark.createDataFrame(data,schema=schema)
Ahora hacemos dos cosas. Primero, Crea una función colsInt y la registramos. Esa función registrada llama a otra función toInt(), que no necesitamos registrar. El primer argumento en udf.register("colsInt", colsInt) están el nombre que utilizamos para referirnos a la función. El segundo están la función que queremos registrar.
colsInt = udf(lambda z: toInt(z), IntegerType())
spark.udf.register("colsInt", colsInt)
def toInt(s):
if isinstance(s, str) == True:
st = [str(ord(i)) for i in s]
return(int(''.join(st)))
else:
return Null
Luego llamamos a la función colinsInt, así. El primer argumento está el nombre de la nueva columna que queremos crear. El segundo están la columna en el dataframe para conectar a la función.
df2 = df.withColumn( 'semployee',colsInt('employee'))
Recuerda que df['employees'] están un objeto columna, no un solo empleado. Eso significa que tenemos que recorrer todas las filas de esa columna, así que utilizamos este bucle lambda (en línea).
colsInt = udf(lambda z: toInt(z), IntegerType())
Aquí está el Enfoque 1 en su conjunto:
import pyspark
from pyspark import SQLContext
from pyspark.sql.types import StructType, StructField, IntegerType, FloatType, StringType
from pyspark.sql.functions import udf
from pyspark.sql import Row
conf = pyspark.SparkConf()
sc = pyspark.SparkContext.getOrCreate(conf=conf)
spark = SQLContext(sc)
schema = StructType([
StructField("sales", FloatType(),True),
StructField("employee", StringType(),True),
StructField("ID", IntegerType(),True)
])
data = [[ 10.2, "Fred",123]]
df = spark.createDataFrame(data,schema=schema)
colsInt = udf(lambda z: toInt(z), IntegerType())
spark.udf.register("colsInt", colsInt)
def toInt(s):
if isinstance(s, str) == True:
st = [str(ord(i)) for i in s]
return(int(''.join(st)))
else:
return Null
df2 = df.withColumn( 'semployee',colsInt('employee'))
Ahora mostramos los resultados. Fíjate que se ha añadido el nuevo semployee de columna. withColumn() crea un nuevo dataframe, así que creamos df2.
df2.show()
+-----+--------+---+----------+
|sales|employee| ID| semployee|
+-----+--------+---+----------+
| 10.2| Fred|123|1394624364|
+-----+--------+---+----------+
Enfoque 2: Uso de SQL
El primer paso aquí está registrar el dataframe como una tabla, para poder ejecutar sentencias SQL sobre él. DF están el dataframe y DFTAB están la tabla temporal que creamos.
spark.registerDataFrameAsTable(df, "dftab")
Ahora Crea un nuevo dataframe df3 a partir del existente en df y aplicamos la función colsInt a la columna employee.
df3 = spark.sql("select sales, employee, ID, colsInt(employee) as iemployee from dftab")
Aquí son los resultados:
df3.show()
+-----+--------+---+----------+
|sales|employee| ID| iemployee|
+-----+--------+---+----------+
| 10.2| Fred|123|1394624364|
+-----+--------+---+----------+
Enfoque 3: Mapa RDD
Un dataframe no tiene una función map(). Si queremos utilizar esa función, debemos convertir el dataframe a un RDD usando dff.rdd.
Aplica la función así:
rdd = df.rdd.map(toIntEmployee)
Esto pasa un objeto fila a la función toIntEmployee. Así que tenemos que devolver un objeto fila. El RDD está inmutable, así que debemos Crea una nueva fila.
A continuación, nos referimos al elemento empleado en la fila por nombre y luego convertimos cada letra de ese campo en un entero y concatenamos esas.
def toIntEmployee(rdd):
s = rdd["employee"]
if isinstance(s, str) == True:
st = [str(ord(i)) for i in s]
e = int(''.join(st))
else:
e = s
return Row(rdd["sales"],rdd["employee"],rdd["ID"],e)
Ahora imprimimos los resultados:
for x in rdd.collect():
print(x)
Guía Apache Spark
- La Canalización de Aprendizaje Automático de Apache Spark: Una introducción
- Ejemplo de regresión lineal SGD con Apache Spark
- Clasificador de árbol de decisión de chispa
- Uso de regresión logística, scala y chispa
- Cómo utilizar cuadernos Jupyter con Apache Spark
- Uso de aprendizaje automático en Python y Spark para hacer clasificación
- Cómo escribir UDFs (funciones definidas por el usuario) de Spark en Python
- Ejemplo de Python Spark ML K-Means
- Cómo aplicar el aprendizaje automático a la ciberseguridad
- Apache Spark: Trabajando con Streams
- K-significa agrupación con Apache Spark
- Usando Chispa con Colmena
- Cómo utilizar Apache Spark para Haz predicciones para el mantenimiento preventivo
Estas publicaciones son mías y no representan necesariamente la posición, estrategias u opinión de BMC.