Batch Python användardefinierade funktioner (UDF:er) i Unity Catalog

Batch Unity Catalog Python UDFs utökar funktionaliteten hos Unity Catalog UDFs genom att låta dig skriva Python-kod för att bearbeta stora mängder data, vilket avsevärt förbättrar effektiviteten genom att minska den belastning som är förknippad med rad-för-rad UDFs. Dessa optimeringar gör att Unity Catalog batch Python UDFs blir idealiska för storskalig databehandling.

Krav

Batch Unity Catalog Python UDF:er kräver Databricks Runtime versioner 16.3 och senare.

Skapa Batch Unity Catalog Python UDF

Att skapa en Batch Unity Catalog Python UDF liknar att skapa en vanlig UDF för Unity Catalog med följande tillägg:

• PARAMETER STYLE PANDAS: Detta anger att UDF behandlar data i grupper med hjälp av Pandas-iteratorer.
• HANDLER 'handler_function': Detta anger den hanterarfunktion som anropas för att bearbeta satserna.

Följande exempel visar hur du skapar en Batch Unity Catalog Python UDF:

%sql
CREATE OR REPLACE TEMPORARY FUNCTION calculate_bmi_pandas(weight_kg DOUBLE, height_m DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
DETERMINISTIC
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)
$$;

Efter att ha registrerat UDF:en kan du anropa den med SQL eller Python.

SELECT person_id, calculate_bmi_pandas(weight_kg, height_m) AS bmi
FROM (
  SELECT 1 AS person_id, CAST(70.0 AS DOUBLE) AS weight_kg, CAST(1.75 AS DOUBLE) AS height_m UNION ALL
  SELECT 2 AS person_id, CAST(80.0 AS DOUBLE) AS weight_kg, CAST(1.80 AS DOUBLE) AS height_m
);

Hanteringsfunktion för batch-UDF

Batch Unity Catalog Python UDF:er kräver en hanterarfunktion som bearbetar batcher och ger resultat. Du måste ange namnet på hanteringsfunktionen när du skapar UDF med hjälp HANDLER av nyckeln.

Hanteringsfunktionen gör följande:

1. Accepterar ett iterator-argument som itererar över en eller flera pandas.Series. Varje pandas.Series innehåller indatamängder för UDF.
2. Itererar över generatorn och bearbetar data.
3. Returnerar en generator-iterator.

Batch Unity Catalog Python UDF:er måste returnera samma antal rader som inmatningen. Hanterarfunktionen säkerställer detta genom att avge en pandas.Series med samma längd som inmatningsserien för varje batch.

Installera anpassade beroenden

Du kan utöka funktionerna i Batch Unity Catalog Python UDF:er utöver Databricks Runtime-miljön genom att definiera anpassade beroenden för externa bibliotek.

Se Utöka UDF:er med hjälp av anpassade beroenden.

Batch-UDF:er kan acceptera enskilda eller flera parametrar

Enkel parameter: När hanteringsfunktionen använder en enda inmatningsparameter, får den en iterator som itererar över en pandas.Series för varje batch.

%sql
CREATE OR REPLACE TEMPORARY FUNCTION one_parameter_udf(value INT)
RETURNS STRING
LANGUAGE PYTHON
DETERMINISTIC
PARAMETER STYLE PANDAS
HANDLER 'handler_func'
AS $$
import pandas as pd
from typing import Iterator
def handler_func(batch_iter: Iterator[pd.Series]) -> Iterator[pd.Series]:
  for value_batch in batch_iter:
    d = {"min": value_batch.min(), "max": value_batch.max()}
    yield pd.Series([str(d)] * len(value_batch))
$$;
SELECT one_parameter_udf(id), count(*) from range(0, 100000, 3, 8) GROUP BY ALL;

Flera parametrar: För flera indata-parametrar, tar funktionens hanterare emot en iterator som itererar över flera pandas.Series. Värdena i serien är i samma ordning som inmatningsparametrarna.

%sql
CREATE OR REPLACE TEMPORARY FUNCTION two_parameter_udf(p1 INT, p2 INT)
RETURNS INT
LANGUAGE PYTHON
DETERMINISTIC
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 p1, p2 in batch_iter: # same order as arguments above
    yield p1 + p2
$$;
SELECT two_parameter_udf(id , id + 1) from range(0, 100000, 3, 8);

Optimera prestandan genom att separera kostsamma operationer

Du kan optimera beräkningsintensiva operationer genom att separera dessa operationer från hanterarfunktionen. Detta säkerställer att de bara körs en gång i stället för under varje iteration över batchar med data.

I följande exempel visas hur du ser till att en dyr beräkning endast utförs en gång:

%sql
CREATE OR REPLACE TEMPORARY FUNCTION expensive_computation_udf(value INT)
RETURNS INT
LANGUAGE PYTHON
DETERMINISTIC
PARAMETER STYLE PANDAS
HANDLER 'handler_func'
AS $$
def compute_value():
  # expensive computation...
  return 1

expensive_value = compute_value()
def handler_func(batch_iter):
  for batch in batch_iter:
    yield batch * expensive_value
$$;
SELECT expensive_computation_udf(id), count(*) from range(0, 100000, 3, 8) GROUP BY ALL

Isoleringsgränser och säkerhet

Batch Unity Catalog Python UDF:er med samma ägare kan dela en isoleringsmiljö som standard. Detta kan förbättra prestanda och minska minnesanvändningen genom att minska antalet separata miljöer som måste startas.

Strikt isolering

Lägg till STRICT ISOLATION egenskapssatsen för att säkerställa att en UDF alltid körs i en egen, helt isolerad miljö.

De flesta UDF:er behöver inte strikt isolering. UDF:er för standarddatabearbetning drar nytta av standardmiljön för delad isolering och körs snabbare med lägre minnesförbrukning.

Lägg till egenskapssatsen STRICT ISOLATION i UDF:er som:

• Kör indata som kod med hjälp av eval(), exec()eller liknande funktioner
• Skriva filer till det lokala filsystemet
• Ändra globala variabler eller systemtillstånd
• Få åtkomst till eller ändra miljövariabler

I följande exempel visas en UDF som kör indata som kod och kräver strikt isolering:

CREATE OR REPLACE TEMPORARY FUNCTION eval_string(input STRING)
RETURNS STRING
LANGUAGE PYTHON
PARAMETER STYLE PANDAS
HANDLER 'handler_func'
STRICT ISOLATION
AS $$
import pandas as pd
from typing import Iterator

def handler_func(batch_iter: Iterator[pd.Series]) -> Iterator[pd.Series]:
  for code_series in batch_iter:
    def eval_func(code):
      try:
        return str(eval(code))
      except Exception as e:
        return f"Error: {e}"
    yield code_series.apply(eval_func)
$$;

Tjänstautentiseringsuppgifter i Batch Unity Catalog Python UDF:er

Batch Unity Catalog Python UDF:er kan använda Unity Catalog Service Credentials för att få tillgång till externa molntjänster. Detta är särskilt användbart för att integrera molnfunktioner som säkerhetstokeniserare i arbetsflöden för databehandling.

Information om hur du skapar en tjänstautentiseringsuppgift finns i Skapa autentiseringsuppgifter för tjänsten.

Ange den tjänsteautentisering du vill använda i CREDENTIALS-klausulen i UDF-definitionen.

CREATE OR REPLACE TEMPORARY FUNCTION example_udf(data STRING)
RETURNS STRING
LANGUAGE PYTHON
PARAMETER STYLE PANDAS
HANDLER 'handler_function'
CREDENTIALS (
  `credential-name` DEFAULT,
  `complicated-credential-name` AS short_name,
  `simple-cred`,
  cred_no_quotes
)
AS $$
# Python code here
$$;

Tjänstautentiseringstillstånd

UDF-skaparen måste ha ACCESS behörighet för Unity Catalog-tjänstens autentiseringsuppgifter. Men för UDF-användare är det tillräckligt att ge dem EXECUTE behörighet på UDF. I synnerhet behöver UDF-anropare inte tillgång till den underliggande tjänstebeviset, eftersom UDF:en körs med de behörigheter som UDF-skaparen har.

För tillfälliga funktioner är skaparen alltid den som anropar. UDF:er som körs i No-PE-scope, även kända som dedikerade kluster, använder istället uppringarens behörigheter.

Förinställda uppgifter och aliaser

Du kan inkludera flera referenser i CREDENTIALS-satsen, men endast en kan markeras som DEFAULT. Du kan använda alias för icke-standardbehörigheter med hjälp av AS nyckelordet. Varje referens måste ha ett unikt alias.

Patchede moln-SDK:er hämtar automatiskt standardautentiseringsuppgifter. Standardautentiseringen har företräde framför alla standardvärden som anges i compute-enhetens Spark-konfiguration och kvarstår i Unity Catalog UDF-definitionen.

Du måste installera azure-identity paketen för att använda DefaultAzureCredential leverantören. Använd ENVIRONMENT klausulen för att installera externa bibliotek. För att lära dig mer om att installera externa bibliotek, se Utöka UDF:er med anpassade beroenden.

Exempel på tjänstautentiseringsuppgifter – Azure Functions

I följande exempel används en tjänstautentiseringsuppgift för att anropa en Azure-funktion från en Batch Unity Catalog Python UDF:

%sql
CREATE OR REPLACE FUNCTION main.test.call_azure_func(data STRING, debug BOOLEAN) RETURNS STRING LANGUAGE PYTHON
PARAMETER STYLE PANDAS
HANDLER 'batchhandler'
CREDENTIALS (
  `batch-udf-service-creds-example-cred` DEFAULT
)
ENVIRONMENT (
  dependencies = '["azure-identity", "requests"]', environment_version = 'None'
)
AS $$
import json
import pandas as pd
import requests
from azure.identity import DefaultAzureCredential
from pyspark.taskcontext import TaskContext


def batchhandler(it):
  # DefaultAzureCredential automatically uses the DEFAULT credential
  credential = DefaultAzureCredential()
  token = credential.get_token("https://management.azure.com/.default")

  # Azure Function URL
  function_url = "https://your-function-app.azurewebsites.net/api/HashValuesFunction"

  # Propagate TaskContext information:
  user_ctx = {"custom": {"user": TaskContext.get().getLocalProperty("user")}}

  for vals, is_debug in it:
    payload = {
      "values": vals.to_list(),
      "is_debug": bool(is_debug[0]),
      "context": user_ctx
    }

    headers = {
      "Authorization": f"Bearer {token.token}",
      "Content-Type": "application/json"
    }

    response = requests.post(function_url, json=payload, headers=headers)

    if response.status_code != 200:
      raise Exception(f"Function call failed: {response.text}")

    response_data = response.json()
    if "errorMessage" in response_data:
      raise Exception(str(response_data))

    yield pd.Series(response_data["values"])
$$;

Ring UDF efter att den har registrerats:

SELECT main.test.call_azure_func(data, false)
FROM VALUES
('abc'),
('def')
AS t(data)

Hämta uppgiftskörningskontext

Använd TaskContext PySpark API för att få kontextinformation såsom användarens identitet, klustertaggar, spark-jobb-ID och mer. Se Hämta uppgiftskontext i en UDF.

Ange DETERMINISTIC om funktionen ger konsekventa resultat

Lägg till DETERMINISTIC i funktionsdefinitionen om den genererar samma utdata för samma indata. På så sätt kan frågeoptimeringar förbättra prestandan.

Som standardinställning antas Batch Unity Catalog Python-UDTF:er vara icke-deterministiska om de inte explicit deklareras. Exempel på icke-deterministiska funktioner är: generera slumpmässiga värden, komma åt aktuella tider eller datum eller göra externa API-anrop.

Se CREATE FUNCTION (SQL och Python)

Begränsningar

• Python-funktioner måste hantera NULL-värden oberoende, och alla typmappningar måste följa Azure Databricks SQL-språkets mappningar.
• Batch Unity Catalog Python UDF:er körs i en säker, isolerad miljö och har inte tillgång till ett delat filsystem eller interna tjänster.
• Flera UDF-anrop inom ett steg är serialiserade och mellanresultat är materialiserade och kan överföras till disk.
• Tjänstautentiseringsuppgifter är endast tillgängliga i Batch Unity Catalog Python UDF:er och scalar Python UDF:er. De stöds inte i Standard Unity Catalog Python UDF:er.
• På dedikerade kluster och för temporära funktioner måste funktionsanroparen ha ACCESS behörigheter för tjänstens autentiseringsuppgifter. Se Bevilja behörighet för att använda en tjänstebaserad autentiseringsuppgift för att få tillgång till en extern molntjänst.
• Aktivera funktionen Offentlig förhandsversion Aktivera nätverk för UDF:er i serverlösa SQL Warehouses på arbetsytans förhandsgranskningssida för att göra Batch Unity Catalog Python UDF-anrop till externa tjänster på serverlös SQL Warehouse-beräkning.

• Om du vill göra Batch Unity Catalog Python UDF-anrop på en serverlös anteckningsbok eller jobbberäkning måste du konfigurera serverlös utflödeskontroll