Funções escalares definidas pelo usuário - Scala

Artigo
10/25/2024

Este artigo contém exemplos de função definida pelo usuário (UDF) do Scala. Ele mostra como registrar UDFs, como invocar UDFs e advertências sobre a ordem de avaliação de subexpressões no Spark SQL. Consulte Funções escalares definidas pelo usuário externo (UDFs) para obter mais detalhes.

Nota

UDFs Scala em recursos de computação habilitados para Unity Catalog com modo de acesso compartilhado requer o Databricks Runtime 14.2 e superior.

Registrar uma função como UDF

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

Chamar o UDF no Spark SQL

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

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

Usar UDF com 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"))

Ordem de avaliação e verificação nula

O Spark SQL (incluindo SQL e as APIs DataFrame e Dataset) não garante a ordem de avaliação das subexpressões. Em particular, as entradas de um operador ou função não são necessariamente avaliadas da esquerda para a direita ou em qualquer outra ordem fixa. Por exemplo, a lógica AND e OR as expressões não têm semântica de "curto-circuito" da esquerda para a direita.

Portanto, é perigoso confiar nos efeitos colaterais ou na ordem de avaliação das expressões booleanas, e na ordem das e HAVING cláusulas, uma vez que tais expressões e cláusulas podem ser reordenadas durante a otimização e o planejamento da WHERE consulta. Especificamente, se um UDF depende de semântica de curto-circuito no SQL para verificação nula, não há garantia de que a verificação nula acontecerá antes de invocar o UDF. Por exemplo,

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 WHERE cláusula não garante que o strlen UDF seja invocado após filtrar nulos.

Para executar a verificação nula adequada, recomendamos que você siga um destes procedimentos:

Tornar a própria UDF nula e fazer verificação nula dentro da própria UDF
Use IF ou CASE WHEN expressões para fazer a verificação nula e invocar a UDF em uma ramificação 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

APIs de conjunto de dados tipadas

Nota

Esse recurso é suportado em clusters habilitados para Unity Catalog com modo de acesso compartilhado no Databricks Runtime 15.4 e superior.

As APIs de conjunto de dados tipadas permitem executar transformações como mapa, filtro e agregações em conjuntos de dados resultantes com uma função definida pelo usuário.

Por exemplo, o seguinte aplicativo Scala usa a map() API para modificar um número em uma coluna de resultados para uma cadeia de caracteres prefixada.

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

Embora este exemplo use a map() API, isso também se aplica a outras APIs de conjunto de dados tipadas, como filter(), mapPartitions(), foreach(), foreachPartition(), reduce()e flatMap().

Partilhar via