Funciones escalares definidas por el usuario - Scala

Artículo
10/25/2024

Este artículo contiene ejemplos de funciones definidas por el usuario (UDF) de Scala. Muestra cómo registrar UDF, cómo invocar UDF y advertencias con respecto al orden de evaluación de las subexpresiones en Spark SQL. Consulte Funciones escalares (UDF) definidas por el usuario externas para obtener más información.

Nota:

UDF de Scala en recursos de proceso habilitados para Unity Catalog con el modo de acceso compartido requiere Databricks Runtime 14.2 y versiones posteriores.

Registro de una función como UDF

val squared = (s: Long) => {
  s * s
}
spark.udf.register("square", squared)

Llamada a la UDF en Spark SQL

spark.range(1, 20).createOrReplaceTempView("test")

%sql select id, square(id) as id_squared from test

Uso de UDF con dataframes

import org.apache.spark.sql.functions.{col, udf}
val squared = udf((s: Long) => s * s)
display(spark.range(1, 20).select(squared(col("id")) as "id_squared"))

Orden de evaluación y comprobación de valores NULL

Spark SQL (incluidas SQL y las DataFrame API y Dataset API) no garantiza el orden de evaluación de las subexpresiones. En concreto, las entradas de un operador o función no se evalúan necesariamente de izquierda a derecha ni en ningún otro orden fijo. Por ejemplo, las expresiones lógicas AND y OR no tienen semántica de "cortocircuito" de izquierda a derecha.

Por lo tanto, es peligroso basarse en los efectos secundarios o el orden de evaluación de las expresiones booleanas y el orden de las cláusulas WHERE y HAVING, ya que estas expresiones y cláusulas se pueden reordenar durante la optimización y el planeamiento de consultas. En concreto, si una UDF se basa en la semántica de cortocircuito en SQL para la comprobación de valores NULL, no hay ninguna garantía de que se realizará la comprobación de valores NULL antes de invocar la UDF. Por ejemplo,

spark.udf.register("strlen", (s: String) => s.length)
spark.sql("select s from test1 where s is not null and strlen(s) > 1") // no guarantee

Esta cláusula WHERE no garantiza que se invoque la UDF strlen después de filtrar los valores NULL.

Para realizar una comprobación correcta de los valores NULL, se recomienda realizar una de las siguientes acciones:

Hacer que la UDF tenga en cuenta los valores NULL y realizar la comprobación de valores NULL dentro de la propia UDF
Uso de expresiones IF o CASE WHEN para realizar la comprobación de valores NULL e invocar la UDF en una rama condicional

spark.udf.register("strlen_nullsafe", (s: String) => if (s != null) s.length else -1)
spark.sql("select s from test1 where s is not null and strlen_nullsafe(s) > 1") // ok
spark.sql("select s from test1 where if(s is not null, strlen(s), null) > 1")   // ok

API de conjunto de datos con tipo

Nota:

Esta característica se admite en clústeres habilitados para Unity Catalog con el modo de acceso compartido en Databricks Runtime 15.4 y versiones posteriores.

Las API de conjunto de datos con tipo le permiten ejecutar transformaciones como asignaciones, filtros y agregaciones en los conjuntos de datos resultantes con una función definida por el usuario.

Por ejemplo, la siguiente aplicación de Scala usa la API de map() para modificar un número de una columna de resultados en una cadena con prefijo.

spark.range(3).map(f => s"row-$f").show()

Aunque en este ejemplo se usa la API de map(), esto también se aplica a otras API de conjunto de datos con tipo, como filter(), mapPartitions(), foreach(), foreachPartition(), reduce() y flatMap().

Compartir a través de