Aller au contenu principal
Dernière mise à jour le

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

Les UDF scalaires Python vous permettent d’exécuter une logique Python personnalisée dans les requêtes SQL sur Databricks. Cette page indique comment les enregistrer et les appeler, utiliser les identifiants et les secrets de service, et gérer les mises en garde relatives à 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.functions import col, lit, udf
from pyspark.sql.types import VariantType, VariantVal

# 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
from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import StructField, StructType, VariantType, VariantVal

# 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
from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import ArrayType, VariantType, VariantVal

# 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
from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import MapType, StringType, VariantType, VariantVal

# Return Map<String, Variant>
@udf(returnType = MapType(StringType(), VariantType(), True))
def toMapVariant(jsonString):
  return {"v1": VariantVal.parseJson(jsonString), "v2": VariantVal.parseJson("[" + jsonString + "]")}

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

Fichiers avec UDF​

info

Bêta

Cette fonctionnalité est en bêta. Les administrateurs de Workspace peuvent contrôler l'accès à cette fonctionnalité à partir de la page Aperçus . Voir Gérer les prévisualisations Databricks.

Le type PySpark pour un fichier est FileType. Utilisez-le comme parameter ou type de retour dans une UDF, soit en tant que type de niveau supérieur, soit imbriqué. Pour le type, ses règles d’imbrication et l’API FileRef, consultez FileType.

Pour lire le contenu d’un fichier dans une UDF, appelez file.as_local_file() pour obtenir un chemin local que vous pouvez ouvrir, ou file.open() pour lire ses octets sous forme de stream. Pour des exemples en Python, Scala et SQL, incluant le traitement d’images, la détection de type de fichier et l’extraction de trames vidéo, voir Traiter des fichiers avec des UDF. Pour la référence de type, voir typeFILE.

Une UDF enregistrée dans Unity Catalog (CREATE FUNCTION) peut lire les métadonnées d'un FILE, mais pas son contenu, et elle ne peut pas créer de fichiers. Utilisez une UDF à portée de session pour lire le contenu d'un fichier (open, as_local_file) ou en créer un (from_bytes, from_local_file).

Dans le corps de l’UDF, chaque valeur FILE est un objet FileRef :

Python
from pyspark.sql.functions import col, udf
from pyspark.sql.types import FileRef, StringType

@udf(returnType=StringType())
def file_content_type(file: FileRef) -> str:
  return file.content_type

df = spark.table("documents")
display(df.select(col("file").uri, file_content_type(col("file"))))

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

Accéder aux secrets de Unity Catalog​

Pour accéder à un secret Unity Catalog à partir d’une UDF Python avec scellement de session, consultez Use a secret in a session-scoped Python UDF. Pour accéder aux secrets déclarés à partir d’une UDF Python Unity Catalog scalaire ou par batch, consultez Use secrets in a Python UDF.

Identifiants de service dans les UDF Python​

Les UDF Python scalaires à portée de session et les UDF Python Unity Catalog scalaires peuvent utiliser des informations d’identification de service Unity Catalog pour accéder de manière sécurisée aux services cloud externes. Cette fonctionnalité est utile pour intégrer des 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 exigences varient selon le type d’UDF et le compute. Voir Utiliser des informations d’identification de service dans une UDF Python.

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

Utiliser une information d'identification de service dans une UDF Python scalaire limitée à la session​

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​

Les UDF limitées à la session utilisent les autorisations de l’appelant. Consultez Utiliser une accréditation de service dans une UDF Python pour connaître les privilèges requis.

Identifiants par default au niveau du compute pour les UDF à portée de session​

Lorsqu'elle est utilisée dans des UDF Python scalaires, Databricks utilise automatiquement l'information d'identification de service par default issue de la variable d'environnement du 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 une information d'identification de service par default pour une ressource compute

La prise en charge des informations d’identification default est uniquement disponible dans les clusters en mode d’accès Standard et Dédié. Cette fonctionnalité n’est pas disponible sur les SQL Warehouse.

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

Session-scoped Python UDF AWS Lambda function example​

L'exemple suivant utilise des informations d'identification de service pour appeler une fonction AWS Lambda à partir d'une UDF Python. L'opération réalise les actions 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]

Utilisez un identifiant de service dans une UDF Python Unity Catalog scalaire​

Spécifiez l'identifiant de service dans la clause CREDENTIALS de la définition de l'UDF. Vous pouvez marquer un identifiant comme DEFAULT afin que les SDK cloud corrigés l'utilisent automatiquement. Sur le compute classique, cette fonction nécessite Databricks Runtime 18.1 ou une version ultérieure. Sur le compute serverless ainsi que sur les SQL warehouses pro et serverless, définissez explicitement environment_version de l'UDF sur 6 ou une version supérieure. Pour connaître l'ensemble des exigences en matière de compute, de réseau et d'autorisations, consultez Utiliser un identifiant de service dans une UDF Python.

Exemple de justificatif d'identité de service : fonction AWS Lambda​

L’exemple suivant utilise des informations d’identification de service pour appeler une fonction AWS Lambda à partir d’une UDF Python Unity Catalog scalaire. L’exemple utilise la version 6 de l’environnement serverless.

SQL
CREATE OR REPLACE FUNCTION main.test.call_lambda_func(data STRING, debug BOOLEAN)
RETURNS STRING
LANGUAGE PYTHON
CREDENTIALS (
  `scalar-uc-udf-service-creds-example-cred` DEFAULT
)
ENVIRONMENT (
  dependencies = '["boto3"]',
  environment_version = '6'
)
AS $$
import base64
import boto3
import json
from pyspark.taskcontext import TaskContext

session = boto3.Session()
client = session.client("lambda", region_name="us-west-2")

user_context = {"custom": {"user": TaskContext.get().getLocalProperty("user")}}
payload = json.dumps({"values": [data], "is_debug": debug})

response = client.invoke(
  FunctionName="HashValuesFunction",
  InvocationType="RequestResponse",
  ClientContext=base64.b64encode(json.dumps(user_context).encode("utf-8")).decode(
    "utf-8"
  ),
  Payload=payload,
)

response_payload = json.loads(response["Payload"].read().decode("utf-8"))
if "errorMessage" in response_payload:
  raise Exception(str(response_payload))

return response_payload["values"][0]
$$;

Appelez l’UDF une fois qu’elle est enregistrée :

SQL
SELECT main.test.call_lambda_func(data, false)
FROM VALUES
('abc'),
('def')
AS t(data)

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.

  • 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