Mik azok a felhasználó által definiált függvények (UDF-ek)?

A felhasználó által definiált függvények (UDF-ek) lehetővé teszik az Azure Databricks beépített funkcióit kiterjesztő kód újrafelhasználását és megosztását. Az UDF-ek használatával konkrét feladatokat hajthat végre, például összetett számításokat, átalakításokat vagy egyéni adatmanipulációkat.

Mikor érdemes UDF-et használni?

A beépített Apache Spark-függvényekkel nehezen kifejezhető logikai UDF-eket használjon. A beépített Apache Spark-függvények elosztott feldolgozásra vannak optimalizálva, és általában jobb teljesítményt nyújtanak nagy méretekben. További információ: Functions.

A Databricks alkalmi lekérdezésekhez, manuális adattisztításhoz, feltáró adatelemzéshez és kis- és közepes méretű adathalmazokon végzett műveletekhez javasolja az UDF-eket. Az UDF-ek gyakori használati esetei közé tartozik az adattitkosítás és a visszafejtés, a kivonatolás, a JSON-elemzés és az ellenőrzés.

Apache Spark-metódusok használata nagyon nagy adathalmazokon és rendszeresen vagy folyamatosan futó számítási feladatokhoz, beleértve az ETL-feladatokat és a streamelési műveleteket.

Regisztrált és munkamenet-hatókörű UDF-ek

Az SQL használatával létrehozott UDF-ek regisztrálva vannak a Unity Katalógusban, és rendelkeznek kapcsolódó engedélyekkel, míg a jegyzetfüzetben létrehozott UDF-ek munkamenetalapúak, és az aktuális SparkSession-hatókörre vannak korlátozva.

A munkamenet-alapú UDF-eket az Azure Databricks által támogatott bármely nyelv használatával definiálhatja és érheti el. Az UDF-ek lehetnek skalárisak vagy nem skalárisak.

Skaláris UDF-ek

A skaláris UDF-ek egyetlen sorban működnek, és minden sorhoz egyetlen értéket ad vissza. Az alábbi példa egy skaláris UDF használatával kiszámítja az egyes nevek hosszát egy name oszlopban, és hozzáadja az értéket egy új oszlophoz name_length:

+-------+-------+
| 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 get_name_length(name STRING)
RETURNS INT
RETURN LENGTH(name);

-- Use the UDF in a SQL query
SELECT name, 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      |
+-------+-------+-------------+

Ezt egy Databricks-jegyzetfüzetben a PySpark használatával valósíthatja meg:

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)

További információkért lásd: Felhasználó által definiált függvények (UDF-ek) a Unity Catalogban és Felhasználó által definiált skaláris függvények – Python.

Felhasználó által megadott aggregátumfüggvények (UDAF-ek)

A felhasználó által definiált összesítő függvények (UDAF-ek) több sorban működnek, és egyetlen összesített eredményt ad vissza. Az alábbi példában egy UDAF van meghatározva, amely összesíti a pontszámokat.

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

Lásd: pandas felhasználó által definiált függvények Pythonhoz és felhasználó által definiált összesítő függvényekhez – Scala.

Python felhasználó által definiált táblafüggvények (UDTF-ek)

A Python által definiált táblafüggvények (UDTF-ek) több sort és oszlopot is visszaadhatnak minden bemeneti sorhoz. Az alábbi példában a pontszámoszlop minden értéke egy kategórialistának felel meg. Egy UDTF van definiálva, amely a vesszővel tagolt listát több sorra osztja. Lásd: Python felhasználó által definiált táblafüggvények (UDTF-ek)

+-------+-------+-----------------+
| name  | score |   categories    |
+-------+-------+-----------------+
| alice |  10.0 |  math,science   |
|  bob  |  20.0 |  history,math   |
| carol |  30.0 | science,history |
| dave  |  40.0 |    math,art     |
|  eve  |  50.0 |  science,art    |
+-------+-------+-----------------+

from pyspark.sql.functions import udtf

@udtf(returnType="score: int, categories: string, name: string")
class ScoreCategoriesUDTF:
    def eval(self, name: str, score: float, categories: str):
        category_list = categories.split(',')
        for category in category_list:
            yield (name, score, category)

# Apply the UDTF
result_df = df.select(ScoreCategoriesUDTF(df.score, df.categories, df.name))
display(result_df)

+-------+-------+----------+
| name  | score | category |
+-------+-------+----------+
| alice |  10.0 |   math   |
| alice |  10.0 | science  |
|  bob  |  20.0 | history  |
|  bob  |  20.0 |   math   |
| carol |  30.0 | science  |
| carol |  30.0 | history  |
| dave  |  40.0 |   math   |
| dave  |  40.0 |   art    |
|  eve  |  50.0 | science  |
|  eve  |  50.0 |   art    |
+-------+-------+----------+

Teljesítménnyel kapcsolatos szempontok

• A beépített függvények és az SQL UDF-ek a leghatékonyabb lehetőségek.
• A Scala UDF-ek általában gyorsabbak, mivel a Java virtuális gépen (JVM) belül futnak, és elkerülik az adatok JVM-be és kifelé történő áthelyezésének többletterhelését.
• A Python UDF-ek és a Pandas UDF-ek általában lassabbak, mint a Scala UDF-jei, mivel az adatok szerializálását és a JVM-ből a Python-értelmezőbe való áthelyezését igénylik. A Pandas UDF-ek akár 100-szor gyorsabban is használhatók, mint a Python UDF-ek, mivel az Apache Arrow használatával csökkentik a szerializációs költségeket.