Skalare benutzerdefinierte Python-Funktionen (UDFs)

Python-skalare UDFs ermöglichen es, benutzerdefinierte Python-Logik innerhalb von SQL-Abfragen auf Azure Databricks auszuführen. Diese Seite zeigt, wie man sie registriert und aufruft, Dienstanmeldeinformationen und Geheimnisse verwendet und mit Einschränkungen bei der Auswertungsreihenfolge von Unterausdrücken in Spark SQL umgeht.

Anforderungen

  • In Databricks Runtime 12.2 LTS und früher werden benutzerdefinierte Funktionen von Python und Pandas (UDFs) nicht in Unity Catalog auf Computeressourcen unterstützt, die den standardmäßigen Zugriffsmodus verwenden.

  • Skalare Python UDFs und Pandas UDFs werden in Databricks Runtime 13.3 LTS und höher für alle Zugriffsmodi unterstützt.

  • Arm-Instanzunterstützung für Python UDFs für Unity-Katalog-fähige Cluster erfordert Databricks Runtime 15.2 oder höher.

In Databricks Runtime 14.0 und darunter werden Python UDFs und Pandas UDFs nicht für Unity-Katalogcluster unterstützt, die den Standardzugriffsmodus verwenden. Skalare Python UDFs und Pandas UDFs werden für alle Zugriffsmodi in Databricks Runtime 14.1 und höher unterstützt.

In Databricks Runtime 14.1 und höher können Sie skalare Python-UDFs mithilfe der SQL-Syntax im Unity-Katalog registrieren. Siehe SQL und Python benutzerdefinierte Funktionen (UDFs) im Unity-Katalog.

Registrieren einer Funktion als UDF

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

Optional können Sie den Rückgabetyp Ihrer UDF festlegen. Der Standardrückgabetyp ist StringType.

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

Aufrufen der UDF in Spark SQL

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

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

Verwenden von UDF mit 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")))

Alternativ können Sie dieselbe UDF mithilfe der Anmerkungssyntax deklarieren:

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

Varianten mit UDF

Der PySpark-Typ für die Variante ist VariantType und die Werte sind vom Typ VariantVal. Informationen zu Varianten finden Sie unter Abfragevariantendaten.

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}|
+---------------+

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}}|
+---------------------+

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}]|
+--------------------+

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}}|
+-----------------------------+

Dateien mit UDF

Important

Dieses Feature befindet sich in der Betaversion.

Der PySpark-Typ für eine Datei ist FileType. Verwenden Sie es als Parameter oder Rückgabetyp in einem UDF, entweder als Typ oberster Ebene oder verschachtelt. Für den Typ, seine Verschachtelungsregeln und die FileRef API siehe FileType.

Um den Inhalt einer Datei in einem UDF zu lesen, rufen Sie file.as_local_file() auf, um einen lokalen Pfad zu erhalten, den Sie öffnen können, oder rufen Sie file.open() auf, um die Bytes der Datei als Stream zu lesen. Beispiele in Python, Scala und SQL, einschließlich Bildverarbeitung, Dateityperkennung und Videoframe-Extraktion, siehe Prozessdateien mit UDFs. Für die Typreferenz siehe FILE Typ.

Ein in Unity Catalog (CREATE FUNCTION) registriertes UDF kann die Metadaten eines FILECodes lesen, aber nicht deren Inhalte, und es kann keine Dateien erstellen. Verwenden Sie eine session-scoped UDF, um den Inhalt einer Datei (open, as_local_file) zu lesen oder eine Datei zu erstellen (from_bytes, from_local_file).

Im UDF-Körper ist jeder FILE Wert ein FileRef Objekt:

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

Auswertungsreihenfolge und NULL-Überprüfung

Spark SQL (einschließlich SQL und der Datenrahmen- und Dataset-APIs) garantiert nicht die Reihenfolge der Auswertung von Teilausdrücken. Insbesondere werden die Eingaben eines Operators oder einer Funktion nicht zwangsläufig von links nach rechts oder in einer anderen festen Reihenfolge ausgewertet. Beispielsweise gilt für logische AND- und OR-Ausdrücke keine „Kurzschluss“-Semantik von links nach rechts.

Daher ist es gefährlich, sich auf die Nebeneffekte oder die Reihenfolge der Auswertung von booleschen Ausdrücken und die Reihenfolge von WHERE- und HAVING-Klauseln zu verlassen, da solche Ausdrücke und Klauseln während der Abfrageoptimierung und -planung neu angeordnet werden können. Insbesondere wenn eine UDF für die NULL-Überprüfung auf die Kurzschluss-Semantik in SQL angewiesen ist, gibt es keine Garantie, dass die NULL-Überprüfung vor dem Aufrufen der UDF erfolgt. Zum Beispiel

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

Diese WHERE-Klausel garantiert nicht, dass die UDF strlen nach dem Herausfiltern von NULL-Werten aufgerufen wird.

Es wird empfohlen, eine der folgenden Aktionen auszuführen, um eine ordnungsgemäße NULL-Überprüfung durchzuführen:

  • Machen Sie die UDF selbst nullfähig, und führen Sie die NULL-Überprüfung innerhalb der UDF selbst durch.
  • Verwenden Sie IF- oder CASE WHEN-Ausdrücke zum Durchführen der NULL-Überprüfung, und rufen Sie die UDF in einer bedingten Verzweigung auf.
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

Zugriff auf Unity-Kataloggeheimnisse

Informationen zum Zugriff auf ein Geheimnis in Unity Catalog über eine sitzungsbezogene Python-UDF finden Sie unter Verwenden eines Geheimnisses in einer sitzungsbezogenen Python-UDF. Informationen zum Zugriff auf deklarierte Geheimnisse in einem skalaren oder Batch-Unity-Catalog-Python-UDF finden Sie unter Geheimnisse in einem Python-UDF verwenden.

Service-Zugangsdaten in Python UDFs

Sitzungsgebundene skalare Python-UDFs und skalare Unity Catalog Python-UDFs können Unity Catalog-Dienstanmeldeinformationen verwenden, um sicher auf externe Cloud-Dienste zuzugreifen. Dies ist nützlich, um Vorgänge wie cloudbasierte Tokenisierung, Verschlüsselung oder geheime Verwaltung direkt in Ihre Datentransformationen zu integrieren.

Die Anforderungen variieren je nach UDF-Typ und Berechnung. Siehe Verwenden einer Dienstanmeldeinformation in einer Python-UDF.

Informationen zum Erstellen von Anmeldeinformationen finden Sie unter Anmeldeinformationen erstellen.

Verwenden Sie eine Dienstanmeldeinformation in einer sitzungsbezogenen skalaren Python-UDF

Um auf die Dienstanmeldeinformationen zuzugreifen, verwenden Sie das databricks.service_credentials.getServiceCredentialsProvider() Dienstprogramm in Ihrer UDF-Logik, um Cloud-SDKs mit den entsprechenden Anmeldeinformationen zu initialisieren. Der gesamte Code muss im UDF-Textkörper gekapselt werden.

@udf
def use_service_credential():
    from azure.mgmt.web import WebSiteManagementClient

    # Assuming there is a service credential named 'testcred' set up in Unity Catalog
    web_client = WebSiteManagementClient(subscription_id, credential = getServiceCredentialsProvider('testcred'))
    # Use web_client to perform operations

Berechtigungen für Dienstanmeldeinformationen

Sitzungsbezogene UDFs verwenden die Berechtigungen des Aufrufers. Siehe Verwenden von Dienstanmeldeinformationen in einer Python-UDF für die erforderlichen Berechtigungen.

Standardzugangsdaten auf Rechenebene für sitzungsbezogene UDFs

Wenn Databricks in skalaren Python UDFs verwendet wird, verwendet es automatisch die Standard-Service-Zugangsberechtigung aus der Rechenumgebungsvariable. Mit diesem Verhalten können Sie sicher auf externe Dienste verweisen, ohne Anmeldeinformationen-Aliase im UDF-Code explizit verwalten zu müssen. Siehe Angeben einer Standarddienstanmeldeinformation für eine Rechenressource

Die Unterstützung von Standard-Anmeldedaten ist nur in Standard- und Dedizierten Zugriffsmodus-Clustern verfügbar. Es ist in SQL-Warehouses nicht verfügbar.

Sie müssen das azure-identity Paket installieren, um den DefaultAzureCredential Anbieter zu verwenden. Zur Installation des Pakets siehe Notizbuch-spezifische Python-Bibliotheken oder Compute-spezifische Bibliotheken.

@udf
def use_service_credential():
    from azure.identity import DefaultAzureCredential
    from azure.mgmt.web import WebSiteManagementClient

    # DefaultAzureCredential is automatically using the default service credential for the compute
    web_client_default = WebSiteManagementClient(DefaultAzureCredential(), subscription_id)

    # Use web_client to perform operations

Verwende eine Service-Credential in einem skalaren Unity Catalog Python UDF

Spezifizieren Sie die Service-Zugangsberechtigung in der CREDENTIALS Klausel der UDF-Definition. Man kann eine Zugangsberechtigung so DEFAULT markieren, dass gepatchte Cloud-SDKs sie automatisch verwenden. Bei klassischer Berechnung benötigt diese Funktion Databricks Runtime 18.1 oder höher. Bei serverloser Rechenleistung sowie in Pro- und serverlosen SQL-Warehouses setzen Sie den Wert environment_version der UDF explizit auf 6 oder höher. Für vollständige Rechen-, Netzwerk- und Berechtigungsanforderungen siehe Use a Service credential in a Python UDF.

Aufgabenausführungskontext abrufen

Verwenden Sie die TaskContext PySpark-API, um Kontextinformationen wie Die Identität des Benutzers, Clustertags, Sparkauftrags-ID und vieles mehr abzurufen. Weitere Informationen finden Sie unter Abrufen des Aufgabenkontexts in einer UDF.

Begrenzungen

Die folgenden Einschränkungen gelten für PySpark UDFs:

  • Dateizugriffseinschränkungen: Auf Databricks Runtime 14.2 und darunter können PySpark UDFs auf freigegebenen Clustern nicht auf Git-Ordner, Arbeitsbereichsdateien oder Unity-Katalogvolumes zugreifen.

  • Übertragungsvariablen: PySpark UDFs auf Standardzugriffsmodusclustern und serverlosen Compute unterstützen keine Übertragungsvariablen.

  • Speicherlimit auf serverlosen Servern: PySpark UDFs auf serverlosen Compute haben eine Speichergrenze von 1 GB pro PySpark UDF. Das Überschreiten dieses Grenzwerts führt zu einem Fehler vom Typ UDF_PYSPARK_USER_CODE_ERROR. MEMORY_LIMIT_SERVERLESS.
  • Speicherbeschränkung für den Standardzugriffsmodus: PySpark UDFs im Standardzugriffsmodus haben eine Speicherbeschränkung basierend auf dem verfügbaren Speicher des ausgewählten Instanztyps. Das Überschreiten des verfügbaren Arbeitsspeichers führt zu einem Fehler vom Typ UDF_PYSPARK_USER_CODE_ERROR. MEMORY_LIMIT.
  • Netzwerkzugriff in serverlosen SQL-Warehouses: Standardmäßig können Python-UDFs in serverlosen SQL-Warehouses keine ausgehenden Netzwerkanfragen senden, und Abfragen, die Netzwerkaufrufe auszuführen versuchen, hängen auf unbestimmte Zeit. Um den ausgehenden Netzwerkzugriff zu aktivieren, aktivieren Sie die Previewfunktion Netzwerk für isolierte Workloads in serverlosen SQL-Warehouses aktivieren der Vorschauversionenseite Ihres Arbeitsbereichs. Verwenden Sie andernfalls serverlose oder klassische Rechenleistung für UDFs, die Netzwerkzugriff erfordern.
  • Zuletzt aktualisiert um