Python UDF:er, Scala UDF:er och komplexa datatyper i den interna körningsmotorn

Den interna körningsmotorn i Microsoft Fabric stöder nu Python användardefinierade funktioner (UDF), Scala UDF:er och komplexa datatyper (matriser, kartor och structs). Med de här funktionerna kan du skriva uttrycksfulla Spark-program utan att offra prestanda.

stöd för Python UDF

Python är ett av de mest populära språken inom datateknik och datavetenskap. Historiskt sett har Python UDF:er medfört betydande omkostnader i Spark på grund av serialiseringskostnader mellan JVM- och Python-arbetsprocesserna. Den interna körningsmotorn minimerar dessa kostsamma växlingar, vilket möjliggör snabbare exekvering utan kodändringar.

Så här fungerar Python UDF:er i den interna körningsmotorn

I en konventionell körningsmodell för Spark innebär körning av Python-UDF:er:

1. Datakonvertering från Sparks interna format.
2. Serialisering och överföring till Python arbetsprocesser.
3. Python UDF-körning.
4. Serialisering av resultat tillbaka till JVM.
5. Spark återupptar körningen.

Den här förflyttningen mellan körningsmiljöer skapar kostnader för serialisering och deserialisering, ineffektiv CPU-användning och brutna kolumnära exekveringskedjor. Den inbyggda körningsmotorn minskar denna overhead genom att optimera sökvägen för dataöverföring och bibehålla vektoriserad bearbetning där så är möjligt.

Stöd för Python UDF-typer

Den inbyggda körningsmotorn stöder:

• Scalar UDF: Rad för rad Python funktioner som registrerats med udf().
• Vektoriserade (Pandas) UDF:er: Funktioner som är dekorerade med @pandas_udf och arbetar på databatchar med hjälp av Apache Arrow för effektiv dataöverföring.

Vektoriserade UDF:er ger störst prestandavinster eftersom de passar naturligt in i den kolumnbaserade bearbetningsmodellen för den inbyggda exekveringsmotorn.

Exempel: Vektoriserad Python UDF

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

@pandas_udf(DoubleType())
def calculate_discount(price: pd.Series, rate: pd.Series) -> pd.Series:
    return price * (1 - rate)

df = spark.table("sales.transactions")
result = df.withColumn("discounted_price", calculate_discount(df.price, df.discount_rate))
result.show()

Ingen ytterligare konfiguration krävs utöver att aktivera den inbyggda körningsmotorn. Befintliga Python UDF:er gynnas automatiskt.

Stöd för Scala UDF

Den inbyggda körningsmotorn snabbar även upp Scala-UDF:er. Eftersom Scala-UDF:er körs nativt i JVM kan motorn avlasta operationer som stöds till den vektoriserade C++-exekveringsvägen samtidigt som utvärderingen av Scala-UDF:er förblir effektiv i samma körningsmiljö.

Exempel: Scala UDF

import org.apache.spark.sql.functions.udf

val toUpperCase = udf((s: String) => s.toUpperCase)
val df = spark.table("catalog.customers")
val result = df.withColumn("name_upper", toUpperCase(df("name")))
result.show()

Scala-UDF:er som körs på datatyper som stöds påskyndas utan kodändringar när den interna körningsmotorn är aktiverad.

Stöd för komplexa datatyper

Moderna lakehouse-arkitekturer är beroende av halvstrukturerade och kapslade data. Den interna körningsmotorn har nu optimerat stöd för:

Åtgärder som stöds för komplexa typer

Den inbyggda exekveringsmotorn snabbar upp vanliga åtgärder på komplexa datatyper:

• Arrayfunktioner: explode, array_contains, size, flatten, transform
• Kartfunktioner: map_keys, map_values, element_at
• Struct-åtkomst: Åtkomst till fält med punktnotation, getField
• Kapslade kombinationer: Matriser med structs, kartor med matrisvärden

Exempel: Arbeta med matriser och structs

from pyspark.sql.functions import explode, col, size

# Read data with nested schema
df = spark.table("events.telemetry")

# Operations on arrays - accelerated by native engine
result = (df
    .filter(size(col("tags")) > 0)
    .select(
        col("event_id"),
        col("metadata.source"),  # Struct field access
        explode(col("tags")).alias("tag")
    )
)
result.show()

Exempel: Arbeta med kartor

from pyspark.sql.functions import map_keys, map_values, col

df = spark.table("config.settings")

# Map operations - accelerated by native engine
result = (df
    .select(
        col("setting_id"),
        map_keys(col("properties")).alias("keys"),
        map_values(col("properties")).alias("values")
    )
)
result.show()

Prestandaresultat

Intern benchmarking visar betydande förbättringar mellan arbetsbelastningar som använder Python UDF:er och komplexa datatyper:

Dessa vinster beror på minskade serialiseringskostnader, förbättrad vektorisering och kolumnkörning från slutpunkt till slutpunkt.

Fördelar med avancerade lakehouse-mönster

Komplex datatypsacceleration är särskilt viktigt för:

• Z-ORDER-optimering: Kapslade kolumner deltar i optimerad datalayout.
• Liquid Clustering: Kolumner med komplexa typer drar nytta av klustring utan att plattas ut.
• Halvstrukturerad analys: JSON-nyttolaster och händelseströmmar är fortfarande kapslade för naturlig frågekörning.
• Händelsedrivna arkitekturer: Telemetri- och IoT-data behåller sin hierarkiska struktur.

I stället för att platta ut data eller omstrukturera pipelines för prestanda kan du arbeta naturligt med komplexa scheman och samtidigt upprätthålla hög körningseffektivitet.

Aktivera funktionen

Python stöd för UDF, Scala UDF och komplexa datatyper är tillgängliga när den interna körningsmotorn är aktiverad. Ingen ytterligare konfiguration krävs.

Information om hur du aktiverar den inbyggda körmotorn finns i Inbyggd körmotor för Fabric Datateknik.

Förutsättningar

• Runtime 1.3 (Apache Spark 3.5) eller Runtime 2.0 (Apache Spark 4.0).
• Den inbyggda körningsmotorn aktiveras på miljö-, notebook- eller Spark-jobbdefinitionsnivå.

Limitations

• Alla Python bibliotek stöds inte i den vektoriserade sökvägen. Bibliotek som kräver serialisering av godtyckliga Python-objekt kan fortfarande utlösa en reservlösning.
• Djupt kapslade komplexa typer (till exempel arrayer av mappningar av strukturer) kan återgå till JVM-motorn för vissa operationer.
• ANSI-läge stöds inte av den inbyggda körningsmotorn.

Relaterat innehåll

• Intern körningsmotor för Infrastrukturdatateknik
• Apache Spark Runtimes på plattformen
• Vad innebär autotuning av Apache Spark-konfigurationer i Fabric?