Aller au contenu principal
Dernière mise à jour le

Fonctions scalaires définies par l'utilisateur Python (UDF)

Cet article contient des exemples de fonctions Python définies par l'utilisateur (UDF). Il montre comment enregistrer des UDF, comment invoquer des UDF et fournit des mises en garde concernant l'ordre d'évaluation des sous-expressions dans Spark SQL.

Exigences​

  • Dans Databricks Runtime 12.2 LTS et versions antérieures, les UDF Python et les UDF Pandas ne sont pas pris en charge sur les compute Unity Catalog qui utilisent le mode d'accès standard.

  • Les UDF Python scalaires et les UDF Pandas sont prises en charge dans Databricks Runtime 13.3 LTS et versions supérieures pour tous les modes d’accès.

  • La prise en charge des instances Graviton pour les UDF Python sur les clusters compatibles Unity Catalog nécessite Databricks Runtime 15.2 ou version ultérieure.

Dans Databricks Runtime 14.0 et les versions inférieures, les UDF Python et les UDF Pandas ne sont pas prises en charge sur les clusters Unity Catalog qui utilisent le mode d'accès standard. Les UDF Python scalaires et les UDF Pandas sont prises en charge pour tous les modes d'accès dans Databricks Runtime 14.1 et les versions ultérieures.

Dans Databricks Runtime 14.1 ou version ultérieure, vous pouvez enregistrer des UDF Python scalaires dans Unity Catalog à l'aide de la syntaxe SQL. Consultez les fonctions définies par l'utilisateur (UDF) SQL et Python dans Unity Catalog.

Enregistrer une fonction en tant qu'UDF​

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

Vous pouvez éventuellement définir le type de retour de votre UDF. Le type de retour default est StringType.

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

Appelez l'UDF dans Spark SQL​

Python
spark.range(1, 20).createOrReplaceTempView("test")
SQL
%sql select id, squaredWithPython(id) as id_squared from test

Utiliser les UDF avec les DataFrames​

Python
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 UDF en utilisant la syntaxe d'annotation :

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

Variantes avec UDF​

Le type PySpark pour la variante est VariantType et les valeurs sont de type VariantVal. Pour des informations sur les variants, consultez les données de variant de query.

Python
from pyspark.sql.types import VariantType

# Return Variant
@udf(returnType = VariantType())
def toVariant(jsonString):
  return VariantVal.parseJson(jsonString)

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toVariant(col("json"))).display()
+---------------+
|toVariant(json)|
+---------------+
|        {"a":1}|
+---------------+
Python
# Return Struct<Variant>
@udf(returnType = StructType([StructField("v", VariantType(), True)]))
def toStructVariant(jsonString):
  return {"v": VariantVal.parseJson(jsonString)}

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toStructVariant(col("json"))).display()
+---------------------+
|toStructVariant(json)|
+---------------------+
|        {"v":{"a":1}}|
+---------------------+
Python
# Return Array<Variant>
@udf(returnType = ArrayType(VariantType()))
def toArrayVariant(jsonString):
  return [VariantVal.parseJson(jsonString)]

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toArrayVariant(col("json"))).display()
+--------------------+
|toArrayVariant(json)|
+--------------------+
|           [{"a":1}]|
+--------------------+
Python
# Return Map<String, Variant>
@udf(returnType = MapType(StringType(), VariantType(), True))
def toArrayVariant(jsonString):
  return {"v1": VariantVal.parseJson(jsonString), "v2": VariantVal.parseJson("[" + jsonString + "]")}

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toArrayVariant(col("json"))).display()
+-----------------------------+
|         toArrayVariant(json)|
+-----------------------------+
|{"v2":[{"a":1}],"v1":{"a":1}}|
+-----------------------------+

Ordre d'évaluation et vérification des valeurs nulles​

Spark SQL (y compris SQL et l’API DataFrame et Dataset) 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.

Par conséquent, il est dangereux de se fier aux effets secondaires ou à l'ordre d'évaluation des expressions booléennes, ainsi qu'à l'ordre des clauses WHERE et HAVING, car ces expressions et clauses peuvent être réorganisées lors de l'optimisation et de la planification des requêtes. Plus précisément, si une UDF repose sur une sémantique de court-circuit en SQL pour la vérification des valeurs nulles, il n'y a aucune garantie que la vérification des valeurs nulles se produira avant d'appeler l'UDF. Par exemple,

Python
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 UDF strlen soit appelée après avoir filtré les valeurs nulles.

Pour effectuer une vérification appropriée des valeurs nulles, nous vous recommandons d'effectuer l'une des opérations suivantes :

  • Rendez l'UDF elle-même sensible aux valeurs nulles et effectuez la vérification des valeurs nulles à l'intérieur de l'UDF elle-même.
  • Utilisez les expressions IF ou CASE WHEN pour effectuer la vérification de nullité et appeler la UDF dans une Branch conditionnelle.
Python
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

Informations d'identification de service dans les UDF Python scalaires​

Les UDF Python scalaires peuvent utiliser les identifiants de service Unity Catalog pour accéder en toute sécurité aux services cloud externes. Ceci est utile pour intégrer les opérations telles que la tokenisation, le chiffrement ou la gestion des secrets basés sur le cloud directement dans vos transformations de données.

Les identifiants de service pour les fonctions UDF Python scalaires sont uniquement pris en charge sur SQL Warehouse et le compute général.

remarque

Les identifiants de service dans les UDF Scalar Python nécessitent Databricks Runtime 17,1 et versions ultérieures.

Pour créer un identifiant de service, consultez Créer des identifiants de service.

remarque

API spécifique aux UDF pour les identifiants de service :
Dans les UDF, utilisez databricks.service_credentials.getServiceCredentialsProvider() pour accéder aux identifiants de service.

Cela diffère de la fonction dbutils.credentials.getServiceCredentialsProvider() utilisée dans les notebooks, qui n'est pas disponible dans les contextes d'exécution UDF.

Pour accéder à l'identifiant de service, utilisez l'utilitaire databricks.service_credentials.getServiceCredentialsProvider() dans votre logique UDF pour initialiser les SDK cloud avec l'identifiant approprié. Tout le code doit être encapsulé dans le corps de l'UDF.

Python
@udf
def use_service_credential():
    from databricks.service_credentials import getServiceCredentialsProvider
    import boto3

    # Assuming there is a service credential named 'testcred' set up in Unity Catalog
    boto3_session = boto3.Session(botocore_session=getServiceCredentialsProvider('testcred'))
    # Use the S3 session to perform operations

Autorisations des identifiants de service​

Le créateur de l’UDF doit disposer de l’autorisation ACCÈS sur les informations d’identification de service Unity Catalog.

Les UDF qui s'exécutent dans la portée sans PE, également appelés clusters dédiés, nécessitent des autorisations MANAGE sur l'identifiant de service.

Identifiants default​

Lorsqu'il est utilisé dans les UDF Scalar Python, Databricks utilise automatiquement l'identifiant de service default de la variable d'environnement compute. Ce comportement vous permet de référencer en toute sécurité des services externes sans gérer explicitement les alias d'informations d'identification dans votre code UDF. Voir Spécifier un identifiant de service default pour une ressource compute

La prise en charge des identifiants default n'est disponible que dans les clusters avec les modes d'accès Standard et Dédié. Ce n'est pas disponible dans DBSQL.

Python
@udf
def use_service_credential():
    from databricks.service_credentials import getServiceCredentialsProvider
    import boto3

    # The default service credential for the compute is automatically used
    boto3_session = boto3.Session()
    # Use the S3 client to perform operations

Exemple d'identifiant de service - fonction AWS Lambda​

L'exemple suivant utilise un identifiant de service pour appeler une fonction AWS Lambda à partir d'une UDF Python scalaire. Il réalise les opérations suivantes :

  1. Récupérez l’identifiant par default à l’aide du fournisseur d’identifiants de service Databricks.
  2. Configure une session boto3.
  3. Appelle une fonction Lambda pour traiter une chaîne d'entrée.
Python
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

@udf(StringType())
def call_lambda_udf(input_str):
    import boto3
    import json
    import base64
    from databricks.service_credentials import getServiceCredentialsProvider
    from pyspark.taskcontext import TaskContext

    # Create a session using the default Unity Catalog service credential
    session = boto3.Session()
    client = session.client("lambda", region_name="us-west-2")

    # Optionally attach Spark TaskContext metadata to the Lambda request
    user_ctx = {"custom": {"user": TaskContext.get().getLocalProperty("user")}}

    # Build the Lambda payload
    payload = json.dumps({
        "values": [input_str],
        "is_debug": False
    })

    # Encode context for Lambda's client context
    encoded_ctx = base64.b64encode(json.dumps(user_ctx).encode("utf-8")).decode("utf-8")

    # Call the Lambda function
    response = client.invoke(
        FunctionName="HashValuesFunction",
        InvocationType="RequestResponse",
        ClientContext=encoded_ctx,
        Payload=payload,
    )

    response_payload = json.loads(response["Payload"].read().decode("utf-8"))

    if "errorMessage" in response_payload:
        raise Exception(response_payload["errorMessage"])

    return response_payload["values"][0]

Obtenir le contexte d'exécution de la tâche​

Utilisez l'API PySpark TaskContext pour obtenir des informations contextuelles telles que l'identité de l'utilisateur, les cluster tags, l'ID du job Spark et plus encore. Consultez Obtenir le contexte de tâche dans une UDF.

Limitations​

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 du Workspace ou aux volumes Unity Catalog.

  • Variables de diffusion : les UDF PySpark sur les clusters en mode d'accès standard et le compute Serverless ne prennent pas en charge les variables de diffusion.

  • Informations d'identification de service : les informations d'identification de service sont disponibles uniquement dans les UDF Python de traitement par batch de Unity Catalog et les UDF Python scalaires. Ils ne sont pas pris en charge dans les UDF Python standard de Unity Catalog.

  • Identifiants de service : les identifiants de service sont disponibles uniquement dans le compute serverless lorsque vous utilisez la version 3 ou supérieure de l’environnement serverless. Consultez Versions d’environnement Serverless.

  • Profils d'instance : les UDF PySpark sur les clusters en mode d'accès standard et le compute serverless ne prennent pas en charge les profils d'instance.

  • Limite de mémoire sur Serverless : Les UDF PySpark sur le compute Serverless ont une limite de mémoire de 1 Go par UDF PySpark. Le dépassement de cette limite entraîne une erreur de type UDF_PYSPARK_USER_CODE_ERROR.MEMORY_LIMIT_SERVERLESS.

  • Limite de mémoire en mode d'accès standard : les UDF PySpark en mode d'accès standard ont une limite de mémoire basée sur la mémoire disponible du type d'instance choisi. Le dépassement de la mémoire disponible entraîne une erreur de type UDF_PYSPARK_USER_CODE_ERROR.MEMORY_LIMIT.

  • Accès réseau dans les SQL Warehouse Serverless : Par default, les UDF Python dans les SQL Warehouse Serverless ne peuvent pas effectuer de requêtes réseau sortantes, et les requêtes qui tentent des appels réseau se bloquent indéfiniment. Pour activer l'accès réseau sortant, activez la fonctionnalité d'aperçu public Activer la mise en réseau pour les charges de travail isolées dans les entrepôts SQL Serverless sur la page Aperçus de votre workspace. Dans le cas contraire, utilisez le compute Serverless ou le compute classique pour les UDF qui nécessitent un accès réseau.

Sur cette page