Användardefinierade skalärfunktioner – Python

Den här artikeln innehåller python-exempel på användardefinierade funktioner (UDF). Den visar hur du registrerar UDF:er, hur du anropar UDF:er och ger varningar om utvärderingsordning för underuttryck i Spark SQL.

I Databricks Runtime 14.0 och senare kan du använda Användardefinierade python-tabellfunktioner (UDF: er) för att registrera funktioner som returnerar hela relationer i stället för skalära värden. Se Användardefinierade tabellfunktioner i Python (UDF).

Registrera en funktion som en UDF

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

Du kan också ange returtypen för din UDF. Standardreturtypen är StringType.

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

Anropa UDF i Spark SQL

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

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

Använda UDF med 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")))

Du kan också deklarera samma UDF med hjälp av anteckningssyntax:

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

Utvärderingsordning och nullkontroll

Spark SQL (inklusive SQL och DataFrame- och Dataset-API:et) garanterar inte utvärderingsordningen för underuttryck. I synnerhet utvärderas inte indata från en operator eller funktion nödvändigtvis från vänster till höger eller i någon annan fast ordning. Till exempel har logiska AND uttryck OR inte vänster-till-höger-semantik för "kortslutning".

Därför är det farligt att förlita sig på biverkningar eller ordning för utvärdering av booleska uttryck och ordningen WHERE på och HAVING satser, eftersom sådana uttryck och satser kan ordnas om under frågeoptimering och planering. Mer specifikt, om en UDF förlitar sig på kortslutningssemantik i SQL för null-kontroll, finns det ingen garanti för att null-kontrollen sker innan UDF anropas. Ett exempel:

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

Den här WHERE satsen garanterar strlen inte att UDF anropas efter filtrering av null-värden.

För att utföra rätt null-kontroll rekommenderar vi att du gör något av följande:

• Gör UDF själv null-medveten och gör null-kontroll i själva UDF:n
• Använd IF eller CASE WHEN uttryck för att göra null-kontrollen och anropa UDF i en villkorsstyrd gren

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

Begränsningar

• PySpark-UDF:er på delade kluster eller serverlös beräkning kan inte komma åt Git-mappar, arbetsytefiler eller UC-volymer för att importera moduler på Databricks Runtime 14.2 och nedan.