Funzioni scalari definite dall'utente - Python

Articolo
10/25/2024

Questo articolo contiene esempi di funzioni definite dall'utente (UDF) python. Illustra come registrare funzioni definite dall'utente, come richiamare funzioni definite dall'utente e fornisce avvertenze sull'ordine di valutazione delle sottoespressioni in Spark SQL.

In Databricks Runtime 14.0 e versioni successive è possibile usare funzioni di tabella definite dall'utente Python per registrare funzioni che restituiscono intere relazioni anziché valori scalari. Vedere Funzioni di tabella definite dall'utente python (UDF) .

Nota

In Databricks Runtime 12.2 LTS e versioni successive le funzioni definite dall'utente Python e Pandas non sono supportate nelle risorse di calcolo di Unity Catalog che usano la modalità di accesso condiviso. Le funzioni definite dall'utente Python scalari e le funzioni definite dall'utente Pandas sono supportate in Databricks Runtime 13.3 LTS e versioni successive per tutte le modalità di accesso.

In Databricks Runtime 13.3 LTS e versioni successive è possibile registrare funzioni definite dall'utente Python scalari nel catalogo unity usando la sintassi SQL. Si veda Funzioni definite dall'utente (UDF) nel catalogo Unity.

Registrare una funzione come UDF

def squared(s):
  return s * s
spark.udf.register("squaredWithPython", squared)

Facoltativamente, è possibile impostare il tipo restituito della funzione definita dall'utente. Il tipo restituito predefinito è StringType.

from pyspark.sql.types import LongType
def squared_typed(s):
  return s * s
spark.udf.register("squaredWithPython", squared_typed, LongType())

Richiamare l’UDF in Spark SQL

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

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

Usare la UDF con DataFrames

from pyspark.sql.functions import udf
from pyspark.sql.types import LongType
squared_udf = udf(squared, LongType())
df = spark.table("test")
display(df.select("id", squared_udf("id").alias("id_squared")))

In alternativa, è possibile dichiarare la stessa funzione definita dall'utente usando la sintassi di annotazione:

from pyspark.sql.functions import udf
@udf("long")
def squared_udf(s):
  return s * s
df = spark.table("test")
display(df.select("id", squared_udf("id").alias("id_squared")))

Ordine di valutazione e controllo null

Spark SQL (incluso SQL e l'API DataFrame e Dataset) non garantisce l'ordine di valutazione delle sottoespressioni. In particolare, gli input di un operatore o di una funzione non vengono necessariamente valutati da sinistra a destra o in qualsiasi altro ordine fisso. Ad esempio, le espressioni logiche AND e OR non hanno semantica "corto circuito" da sinistra a destra.

Pertanto, è pericoloso basarsi sugli effetti collaterali o sull'ordine di valutazione delle espressioni booleane e sull'ordine delle clausole WHERE e HAVING, poiché tali espressioni e clausole possono essere riordinate durante l'ottimizzazione e la pianificazione delle query. In particolare, se una funzione definita dall'utente si basa sulla semantica di corto circuito in SQL per il controllo null, non esiste alcuna garanzia che il controllo null venga eseguito prima di richiamare la funzione definita dall'utente. Ad esempio,

spark.udf.register("strlen", lambda s: len(s), "int")
spark.sql("select s from test1 where s is not null and strlen(s) > 1") # no guarantee

Questa clausola WHERE non garantisce che la funzione definita dall'utente strlen venga richiamata dopo aver filtrato i valori null.

Per eseguire un controllo null appropriato, è consigliabile eseguire una delle operazioni seguenti:

Rendere la funzione definita dall'utente che riconosce i valori null ed eseguire il controllo null all'interno della funzione definita dall'utente stessa
Usare le espressioni IF o CASE WHEN per eseguire il controllo null e richiamare la funzione definita dall'utente in un ramo condizionale

spark.udf.register("strlen_nullsafe", lambda s: len(s) if not s is None else -1, "int")
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

Limiti

Le funzioni definite dall'utente pySpark nei cluster condivisi o nelle risorse di calcolo serverless non possono accedere a cartelle Git, file dell'area di lavoro o volumi UC per importare moduli in Databricks Runtime 14.2 e versioni successive.

Condividi tramite