Was sind benutzerdefinierte Funktionen (UDFs, User Defined Functions)?

Mit benutzerdefinierten Funktionen (USER-Defined Functions, UDFs) können Sie Code wiederverwenden und freigeben, der integrierte Funktionen für Azure Databricks erweitert. Verwenden Sie UDFs, um bestimmte Aufgaben wie komplexe Berechnungen, Transformationen oder benutzerdefinierte Datenmanipulationen auszuführen.

Wann soll eine UDF- und Apache Spark-Funktion verwendet werden?

Verwenden Sie UDFs für Logik, die mit integrierten Apache Spark-Funktionen schwer ausgedrückt werden kann. Integrierte Apache Spark-Funktionen sind für die verteilte Verarbeitung optimiert und gewährleisten eine bessere Leistung im großen Maßstab. Weitere Informationen finden Sie unter Funktionen.

Databricks empfiehlt UDFs für Ad-hoc-Abfragen, manuelle Datenreinigung, explorative Datenanalyse und Vorgänge für kleine bis mittlere Datasets. Häufige Anwendungsfälle für UDFs sind Datenverschlüsselung, Entschlüsselung, Hashing, JSON-Analyse und Validierung.

Verwenden Sie Apache Spark-Methoden für Vorgänge in großen Datasets und alle Workloads, die regelmäßig oder kontinuierlich ausgeführt werden, einschließlich ETL-Aufträgen und Streamingvorgängen.

Grundlegendes zu UDF-Typen

Wählen Sie einen UDF-Typ auf den folgenden Registerkarten aus, um eine Beschreibung, ein Beispiel, und einen Link anzuzeigen, um mehr zu erfahren.

Skalare UDF

Skalare UDFs werden für eine einzelne Zeile ausgeführt und geben einen einzelnen Ergebniswert für jede Zeile zurück. Sie können von Unity Catalog verwaltet werden oder sitzungsbezogen sein.

Im folgenden Beispiel wird eine skalare UDF verwendet, um die Länge der einzelnen Namen in einer name Spalte zu berechnen und den Wert in einer neuen Spalte name_lengthhinzuzufügen.

+-------+-------+
| name  | score |
+-------+-------+
| alice |  10.0 |
| bob   |  20.0 |
| carol |  30.0 |
| dave  |  40.0 |
| eve   |  50.0 |
+-------+-------+

-- Create a SQL UDF for name length
CREATE OR REPLACE FUNCTION main.test.get_name_length(name STRING)
RETURNS INT
RETURN LENGTH(name);

-- Use the UDF in a SQL query
SELECT name, main.test.get_name_length(name) AS name_length
FROM your_table;

+-------+-------+-------------+
| name  | score | name_length |
+-------+-------+-------------+
| alice |  10.0 |      5      |
|  bob  |  20.0 |      3      |
| carol |  30.0 |      5      |
| dave  |  40.0 |      4      |
|  eve  |  50.0 |      3      |
+-------+-------+-------------+

So implementieren Sie dies in einem Azure Databricks-Notizbuch mithilfe von PySpark:

from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType

@udf(returnType=IntegerType())
def get_name_length(name):
  return len(name)

df = df.withColumn("name_length", get_name_length(df.name))

# Show the result
display(df)

Siehe SQL- und Python-benutzerdefinierte Funktionen (UDFs) in Unity Catalog und skalare Python-benutzerdefinierte Funktionen (UDFs).

Batchskalare UDFs

Verarbeiten von Daten in Batches bei gleichzeitiger Beibehaltung der 1:1-Eingabe-/Ausgabezeilenparität. Dadurch wird der Aufwand für Zeilen-nach-Zeilen-Vorgänge für die Verarbeitung großer Datenmengen reduziert. Batch-UDFs behalten auch den Zustand zwischen Batches bei, um effizienter ausgeführt zu werden, Ressourcen wiederzuverwenden und komplexe Berechnungen zu verarbeiten, die Kontext über Datenblöcke hinweg benötigen.

Sie können von Unity Catalog verwaltet werden oder sitzungsbezogen sein.

Der folgende Batch Unity Catalog Python UDF berechnet den BMI während der Verarbeitung von Reihenstapeln.

+-------------+-------------+
| weight_kg   | height_m    |
+-------------+-------------+
|      90     |     1.8     |
|      77     |     1.6     |
|      50     |     1.5     |
+-------------+-------------+

%sql
CREATE OR REPLACE FUNCTION main.test.calculate_bmi_pandas(weight_kg DOUBLE, height_m DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
PARAMETER STYLE PANDAS
HANDLER 'handler_function'
AS $$
import pandas as pd
from typing import Iterator, Tuple

def handler_function(batch_iter: Iterator[Tuple[pd.Series, pd.Series]]) -> Iterator[pd.Series]:
  for weight_series, height_series in batch_iter:
    yield weight_series / (height_series ** 2)
$$;

select main.test.calculate_bmi_pandas(cast(70 as double), cast(1.8 as double));

+--------+
|  BMI   |
+--------+
|  27.8  |
|  30.1  |
|  22.2  |
+--------+

Siehe SQL und Python benutzerdefinierte Funktionen (UDFs) in Unity Catalog und Batch Python benutzerdefinierten Funktionen (UDFs) im Unity-Katalog.

Nicht skalare UDFs

Nicht-skalare UDFs verarbeiten vollständige Datensätze/Spalten mit flexiblen Eingabe-/Ausgabeverhältnissen (1:N oder viele:viele).

Sitzungsbezogene Batch-Pandas-UDFs können die folgenden Typen haben:

• Datenreihe zu Datenreihe
• Iterator der Datenreihe zu Iterator der Datenreihe
• Iterator von mehreren Datenreihen zu Iterator der Datenreihe
• Datenreihe zu skalar

Im Folgenden sehen Sie ein Beispiel für ein Pandas-UDF des Typs „Datenreihe zu Datenreihe“.

from pyspark.sql.functions import pandas_udf
import pandas as pd

df = spark.createDataFrame([(70, 1.75), (80, 1.80), (60, 1.65)], ["Weight", "Height"])

@pandas_udf("double")
def calculate_bmi_pandas(weight: pd.Series, height: pd.Series) -> pd.Series:
  return weight / (height ** 2)

df.withColumn("BMI", calculate_bmi_pandas(df["Weight"], df["Height"])).display()

Siehe benutzerdefinierte Pandas-Funktionen.

UDAF

UDAFs arbeiten mit mehreren Zeilen und geben ein einzelnes aggregiertes Ergebnis zurück. UDAFs sind nur sitzungsbezogen.

Im folgenden UDAF-Beispiel werden die Bewertungen anhand der Namenslänge aggregiert.

from pyspark.sql.functions import pandas_udf
from pyspark.sql import SparkSession
import pandas as pd

# Define a pandas UDF for aggregating scores
@pandas_udf("int")
def total_score_udf(scores: pd.Series) -> int:
  return scores.sum()

# Group by name length and aggregate
result_df = (df.groupBy("name_length")
  .agg(total_score_udf(df["score"]).alias("total_score")))

display(result_df)

+-------------+-------------+
| name_length | total_score |
+-------------+-------------+
|      3      |     70.0    |
|      4      |     40.0    |
|      5      |     40.0    |
+-------------+-------------+

Siehe pandas-benutzerdefinierte Funktionen für Python und Scala-benutzerdefinierte Aggregatfunktionen (UDAFs).

Benutzerdefinierte Tabellenfunktionen (UDTFs)

Ein UDTF akzeptiert ein oder mehrere Eingabeargumente und gibt für jede Eingabezeile mehrere Zeilen (und möglicherweise mehrere Spalten) zurück. Sie können von Unity Catalog verwaltet werden oder sitzungsbezogen sein.

Mit dem folgenden UDTF wird eine Tabelle mit einer festen Liste von zwei ganzzahligen Argumenten erstellt:

CREATE OR REPLACE FUNCTION get_sum_diff(x INT, y INT)
RETURNS TABLE (sum INT, diff INT)
LANGUAGE PYTHON
HANDLER 'GetSumDiff'
AS $$
class GetSumDiff:
    def eval(self, x: int, y: int):
        yield x + y, x - y
$$;

SELECT * FROM get_sum_diff(10, 3);

+-----+------+
| sum | diff |
+-----+------+
| 13  | 7    |
+-----+------+

So implementieren Sie dies in einem Azure Databricks-Notizbuch mithilfe von PySpark:

from pyspark.sql.functions import lit, udtf

@udtf(returnType="sum: int, diff: int")
class GetSumDiff:
    def eval(self, x: int, y: int):
        yield x + y, x - y

GetSumDiff(lit(1), lit(2)).show()

Siehe Unity Catalog UDTFs und session-scoped UDTFs.

Durch Unity Catalog gesteuerte vs. sitzungsbezogene UDFs

Unity Catalog speichert von Unity Catalog verwaltete UDFs für eine verbesserte Governance, Wiederverwendbarkeit und Auffindbarkeit. Sie definieren UDFs mit Sitzungsumfang in einem Notebook oder Job; ihr Geltungsbereich ist auf die aktuelle SparkSession beschränkt. Sie können UDFs im Sitzungsbereich mithilfe von SQL, Python oder Scala definieren und auf sie zugreifen.

Verwenden Sie die folgende Tabelle, um zwischen den beiden Kategorien zu entscheiden. Sehen Sie sich anschließend die Kurzübersichten zu den jeweiligen UDF-Typen in jeder Kategorie an.

Spickzettel zu von Unity Catalog verwalteten UDFs

Durch den Unity-Katalog geregelte UDFs können benutzerdefinierte Funktionen definiert, verwendet, sicher freigegeben und in computerübergreifenden Umgebungen gesteuert werden. Siehe SQL und Python benutzerdefinierte Funktionen (UDFs) im Unity-Katalog.

Spickzettel zu sitzungsbezogenen UDFs für benutzerisoliertes Compute

Sie definieren sitzungsgebundene UDFs in einem Notebook oder Job, die auf die aktuelle SparkSession beschränkt sind. Sie können UDFs im Sitzungsbereich mithilfe von SQL, Python oder Scala definieren und auf sie zugreifen.

Leistungsüberlegungen

• Integrierte Funktionen und SQL UDFs sind die effizientesten Optionen.

• Scala UDFs sind im Allgemeinen schneller als Python UDFs.

  • Nicht isolierte Scala UDFs werden auf dem virtuellen Java-Computer (JVM) ausgeführt, sodass sie den Aufwand für das Verschieben von Daten in und aus dem JVM vermeiden.
  • Isolierte Skala-UDFs müssen Daten in und aus dem JVM verschieben, aber sie können immer noch schneller sein als Python UDFs, da sie speichereffizienter verarbeiten.

• Python UDFs und Pandas UDFs neigen dazu, langsamer zu sein als Scala UDFs, da sie Daten serialisieren und aus dem JVM in den Python-Dolmetscher verschieben müssen.

  • Pandas UDFs sind bis zu 100x schneller als Python UDFs, da sie Apache Arrow verwenden, um die Serialisierungskosten zu reduzieren.