Fonctions scalaires définies par l’utilisateur - Python

2025-05-10

Cet article contient des exemples de fonctions définies par l’utilisateur (UDF) Python. Il montre comment inscrire et appeler des fonctions définies par l’utilisateur, et il fournit des avertissements concernant l’ordre d’évaluation des sous-expressions dans Spark SQL.

Dans Databricks Runtime 14.0 et versions ultérieures, vous pouvez utiliser des fonctions de table définies par l’utilisateur (UDTF) Python pour inscrire des fonctions qui retournent des relations entières au lieu de valeurs scalaires. Consultez Fonctions de table définies par l’utilisateur (UDTF) Python.

Remarque

Dans Databricks Runtime 12.2 LTS et versions antérieures, les fonctions définies par l’utilisateur Python et Pandas ne sont pas prises en charge sur le calcul Unity Catalog qui utilise le mode d’accès standard. Les fonctions scalaires définies par l’utilisateur Python et Pandas sont prises en charge dans Databricks Runtime 13.3 LTS et versions ultérieures pour tous les modes d’accès.

Dans Databricks Runtime 13.3 LTS et versions ultérieures, vous pouvez enregistrer des fonctions scalaires Python UDF dans Unity Catalog en utilisant la syntaxe SQL. Consultez les Fonctions définies par l’utilisateur (UDF) dans Unity Catalog.

Inscrire une fonction en tant que fonction définie par l’utilisateur

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

Vous pouvez éventuellement définir le type de retour de votre fonction définie par l’utilisateur. Le type de retour par défaut est StringType.

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

Appeler l'UDF dans Spark SQL

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

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

Utiliser UDF avec des 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")))

Vous pouvez également déclarer la même fonction définie par l’utilisateur à l’aide de la syntaxe d’annotation :

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")))

Ordre d’évaluation et vérification de nullité

Spark SQL (incluant SQL et les API TrameDonnées et Jeu de données) ne garantit pas l’ordre d’évaluation des sous-expressions. En particulier, les entrées d’un opérateur ou d’une fonction ne sont pas nécessairement évaluées de gauche à droite ou dans tout autre ordre fixe. Par exemple, les expressions logiques AND et OR n’ont pas de sémantique de « court-circuit » de gauche à droite.

Il est donc dangereux de s’appuyer sur les effets secondaires ou l’ordre d’évaluation des expressions booléennes, ainsi que sur l’ordre des clauses WHERE et HAVING, car ces expressions et clauses peuvent être réordonnées lors de l’optimisation et de la planification des requêtes. Plus précisément, si une fonction UDF s’appuie sur la sémantique de court-circuitage dans SQL pour la vérification null, il n’existe aucune garantie que la vérification null se produit avant d’appeler la fonction UDF. Par exemple,

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

Cette clause WHERE ne garantit pas que la fonction définie par l’utilisateur strlen soit appelée après le filtrage des valeurs nulles.

Pour effectuer une vérification de valeur null correcte, nous vous recommandons d’effectuer l’une des opérations suivantes :

Rendez la fonction définie par l’utilisateur sensible aux valeurs nulles et effectuez une vérification de valeurs nulles au sein de la fonction définie par l’utilisateur
Utilisez les expressions IF ou CASE WHEN pour effectuer la vérification de valeurs nulles et appelez la fonction définie par l’utilisateur dans une branche conditionnelle

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

Obtenir le contexte d’exécution des tâches

Utilisez l’API TaskContext PySpark pour obtenir des informations contextuelles telles que l’identité de l’utilisateur, les balises de cluster, l’ID de travail Spark et bien plus encore. Consultez Obtenir le contexte de tâche dans une fonction UDF.

Limites

Les limitations suivantes s'appliquent aux UDF PySpark :

Restrictions d’accès aux fichiers : Sur Databricks Runtime 14.2 et versions antérieures, les UDF PySpark sur les clusters partagés ne peuvent pas accéder aux dossiers Git, aux fichiers d’espace de travail ou aux volumes de catalogue Unity.
Variables de diffusion : les fonctions définies par l’utilisateur PySpark sur les clusters en mode d’accès standard et le calcul serverless ne prennent pas en charge les variables de diffusion.

Limite de mémoire: les UDF PySpark sur le calcul sans serveur ont une limite de mémoire de 1 Go par UDF PySpark. Le dépassement de cette limite entraîne l’erreur suivante : [UDF_PYSPARK_ERROR.OOM] Python worker exited unexpectedly (crashed) due to running out of memory.

Partager via