Funzioni scala e Java definite dall'utente (UDF) nel catalogo Unity

Questa pagina descrive come creare funzioni definite dall'utente (UDF) in Scala e Java, registrarle in Unity Catalog e condividerle tra diversi ambienti di calcolo. Le UDF di Unity Catalog consentono di riutilizzare la logica JVM esistente con la governance e i controlli di accesso di Unity Catalog.

A differenza delle UDF Scala con ambito di sessione, che sono limitate a un singolo notebook o cluster, le UDF registrate in Unity Catalog sono:

• Governato: gestito con permessi e controlli di accesso di Unity Catalog.
• Riutilizzabile: condiviso tra team, notebook, processi e warehouse SQL.
• Rilevabile: visibile in Catalog Explorer e nelle tabelle di sistema.
• Isolato: viene eseguito in sandbox con un costo di avvio a freddo una tantum per sessione. Le chiamate successive sono veloci.

Requisiti

L'area di lavoro deve essere abilitata per Unity Catalog. Si applicano i requisiti aggiuntivi seguenti.

Calcolo: tutti i tipi di calcolo sono supportati, inclusi notebook e processi serverless, warehouse SQL e pipeline dichiarative Spark in Lakeflow. Il calcolo classico richiede Databricks Runtime 18.2 o versione successiva. Negli ambienti di calcolo serverless e nei warehouse SQL, la definizione dell'UDF deve specificare Environment Version 4 o versione successiva nel campo environment_version. Questo requisito si applica alla definizione della UDF, non al notebook o al job chiamante. Consultare le versioni dell'ambiente serverless .

Sviluppo:

• Scala: 2.13.16. Scala 2.12 non è supportato.
• JDK: 17.
• Pacchettizzazione: un JAR fat contenente tutte le dipendenze di terze parti utilizzate dalla funzione definita dall'utente (UDF).

Autorizzazioni:

• Crea una UDF: USAGE e CREATE FUNCTION nello schema e USAGE nel catalogo.
• Eseguire una UDF: EXECUTE sulla funzione e USAGE sullo schema e sul catalogo.
• Accedere al file JAR: READ VOLUME nel volume in cui è archiviato il file JAR.

Per altre informazioni sulle autorizzazioni del catalogo unity, vedere Gestire i privilegi nel catalogo Unity .

Crea il file JAR UDF

Crea un pacchetto JAR con il codice compilato e caricalo in un volume di Unity Catalog prima di registrare l'UDF. Scegliere un metodo di compilazione:

• Compilare in locale
• Compilare in un notebook

Compilare localmente

Seguire questa procedura per creare un file JAR fat usando un ambiente di sviluppo locale.

Configurazione dell'ambiente

Installare gli strumenti necessari nel computer locale. I comandi seguenti sono per macOS. Per altre piattaforme, installare JDK 17 e sbt (Scala) o Maven (Java) usando la gestione pacchetti della piattaforma.

Scala

Installare JDK 17 e sbt:

brew install openjdk@17
brew install sbt

Verificare l'installazione:

java -version   # Should show Java 17
sbt --version   # Should show sbt version

Java

Installare JDK 17 e Maven:

brew install openjdk@17
brew install maven

Verificare l'installazione:

java -version   # Should show Java 17
mvn --version   # Should show Maven version

Creare il progetto

Configurare un progetto in Scala o Java.

Scala

Creare un nuovo progetto Scala usando sbt:

sbt new scala/scala-seed.g8

Quando richiesto, immettere un nome di progetto, ad esempio my-udf-project.

Configurare build.sbt

Sostituire il contenuto del build.sbt file con la configurazione seguente:

scalaVersion := "2.13.16"

ThisBuild / organization := "com.example"

lazy val myUDF = (project in file("."))
  .settings(
    name := "my-udf"
  )

Abilitare il plug-in sbt-assembly

Creare o modificare project/assembly.sbt e aggiungere:

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.0.0")

Questo plug-in crea un file JAR fat contenente tutte le dipendenze.

Java

Creare un nuovo progetto Maven usando l'archetipo di avvio rapido:

mvn archetype:generate \
  -DgroupId=com.example \
  -DartifactId=my-udf \
  -DarchetypeArtifactId=maven-archetype-quickstart \
  -DinteractiveMode=false

Questo comando crea la struttura standard di un progetto Maven con le directory src/main/java e src/test/java.

Configurare pom.xml

Nel file generato pom.xml , all'interno dei <project></project> tag, aggiungere un <properties> blocco con la configurazione seguente:

<properties>
  <maven.compiler.source>17</maven.compiler.source>
  <maven.compiler.target>17</maven.compiler.target>
  <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>

All'interno dei <project></project> tag aggiungere anche un <build> blocco con la configurazione seguente:

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>3.5.0</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

maven-shade-plugin crea un file JAR fat contenente tutte le dipendenze.

Scrivi la tua UDF

Quando scrivi la tua UDF, fai riferimento a Tipi di dati per i tipi di dati supportati e a Corrispondenze tra linguaggi per vedere come i tipi Scala e Java vengono mappati ai tipi SQL.

Il gestore di funzioni definite dall'utente deve soddisfare i requisiti seguenti:

• Scala: definire il gestore come metodo di un object (non di un class). Il valore HANDLER fa riferimento a un metodo su una Scala object.
• Java: definire il gestore come public static metodo.
• Firma: i tipi dei parametri, il loro ordine e il tipo di ritorno del metodo devono corrispondere all'elenco degli argomenti e al tipo RETURNS nell'istruzione CREATE FUNCTION.
• Solo scalare: il gestore deve restituire un singolo valore scalare. I tipi restituiti dalle tabelle non sono supportati.
• Indipendente: il gestore deve operare solo sui relativi argomenti di input. Non può usare le API Spark o dipendere dai pacchetti principali di Spark. Vedere Limitazioni.

Scala

Creare un oggetto Scala in src/main/scala/com/example/MyUDF.scala e definire la funzione UDF.

Esempio di base

package com.example

object MyUDF {
  def addOne(x: Int): Int = x + 1
}

Esempio con dipendenza esterna

Per usare librerie esterne, aggiungerle al build.sbt file:

scalaVersion := "2.13.16"

ThisBuild / organization := "com.example"

lazy val myUDF = (project in file("."))
  .settings(
    name := "currency-udf",
    libraryDependencies ++= Seq(
      "org.apache.commons" % "commons-lang3" % "3.12.0"
    )
  )

Quindi usa la dipendenza nella tua UDF:

package com.example

import org.apache.commons.lang3.StringUtils

object CurrencyUDF {
  private val rates: Map[String, Double] = Map(
    "USD" -> 1.0,
    "EUR" -> 1.1,
    "GBP" -> 1.3,
    "JPY" -> 0.007
  )

  def convertToUSD(price: Double, currency: String): Double = {
    require(currency != null, "Currency must not be null")

    val normalizedCurrency = StringUtils.upperCase(currency)

    rates.get(normalizedCurrency) match {
      case Some(rate) => price * rate
      case None => throw new IllegalArgumentException(s"Unsupported currency: $currency")
    }
  }
}

Testa la tua UDF con test unitari prima della distribuzione. Consulta Come testare localmente le funzioni definite dall'utente.

Java

Creare una classe Java in src/main/java/com/example/MyUDF.java e definire l'UDF come metodo pubblico statico.

Esempio di base

package com.example;

public class MyUDF {
    public static int addOne(int x) {
        return x + 1;
    }
}

Esempio con dipendenza esterna

Per usare librerie esterne, aggiungerle alla <dependencies> sezione del pom.xml file:

<dependencies>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
        <version>3.12.0</version>
    </dependency>
</dependencies>

Quindi usa la dipendenza nella tua UDF:

package com.example;

import org.apache.commons.lang3.StringUtils;
import java.util.Map;
import java.util.HashMap;

public class CurrencyUDF {
    private static final Map<String, Double> rates = new HashMap<>();

    static {
        rates.put("USD", 1.0);
        rates.put("EUR", 1.1);
        rates.put("GBP", 1.3);
        rates.put("JPY", 0.007);
    }

    public static double convertToUSD(double price, String currency) {
        if (currency == null) {
            throw new IllegalArgumentException("Currency must not be null");
        }

        String normalizedCurrency = StringUtils.upperCase(currency);

        if (!rates.containsKey(normalizedCurrency)) {
            throw new IllegalArgumentException("Unsupported currency: " + currency);
        }

        return price * rates.get(normalizedCurrency);
    }
}

Verifica la tua UDF con test unitari prima di distribuirla. Consulta Testare le funzioni definite dall'utente localmente.

Crea il tuo fat JAR

Compilare il progetto per creare un file JAR fat contenente tutte le dipendenze.

Scala

Dalla directory radice del progetto eseguire:

sbt clean assembly

Il file JAR fat viene creato in target/scala-2.13/ con un nome come my-udf-assembly-0.1.0-SNAPSHOT.jar.

Java

Dalla directory radice del progetto eseguire:

mvn clean package

Il file JAR fat viene creato in target/ con un nome come my-udf-1.0-SNAPSHOT.jar.

Caricare il file JAR in un volume di Unity Catalog

Se non si ha già un volume del catalogo Unity, crearne uno:

CREATE VOLUME IF NOT EXISTS my_catalog.my_schema.udf_jars
COMMENT 'Storage for UDF JAR files';

Se altri utenti devono eseguire l'UDF, concedi loro READ VOLUME sul volume:

GRANT READ VOLUME ON VOLUME my_catalog.my_schema.udf_jars TO `user@example.com`;

Caricare il file JAR nel volume usando Esplora cataloghi:

1. Nell'area di lavoro Azure Databricks, fare clic su Catalogo per aprire Esplora Catalogo.
2. Selezionare il catalogo, quindi selezionare lo schema che contiene il volume.
3. Fare clic sul nome del volume.
4. Fare clic su Carica in questo volume e selezionare il file JAR.
5. Fare clic su Carica.
6. Al termine del caricamento, fare clic sul nome del file JAR.
7. Fare clic su Copia percorso per copiare il percorso del volume negli Appunti. Ad esempio, /Volumes/my_catalog/my_schema/udf_jars/my-udf-assembly-0.1.0-SNAPSHOT.jar (Scala) o /Volumes/my_catalog/my_schema/udf_jars/my-udf-1.0-SNAPSHOT.jar (Java). Questo percorso è necessario quando si registra l'UDF.

Eseguire la build nel notebook

È possibile compilare una UDF, pacchettizzarla come file JAR e caricarla in un volume di Unity Catalog direttamente da un notebook di Azure Databricks. Questo approccio funziona per UDF piccole e senza dipendenze. Per le funzioni definite dall'utente con librerie di terze parti, utilizza Compila in locale.

La seguente cella Python scrive una UDF Java che pulisce una stringa (rimuove gli spazi iniziali e finali, comprime gli spazi ripetuti e converte il testo in minuscolo), la compila con JDK 17, la impacchetta in un file JAR e la copia in un volume di Unity Catalog. Aggiorna volume_path in modo che punti a un volume esistente per il quale si dispone dell'autorizzazione WRITE VOLUME.

import os
import subprocess
import shutil

build_dir = "/tmp/udf_build"
package_dir = f"{build_dir}/src/com/databricks/udf"
classes_dir = f"{build_dir}/classes"
os.makedirs(package_dir, exist_ok=True)
os.makedirs(classes_dir, exist_ok=True)

# The UDF handler: a public static method on a plain Java class.
# The doubled backslashes produce a single backslash in the Java source (\\s+).
udf_code = """package com.databricks.udf;
public class StringCleanUDF {
    public static String clean(String input) {
        if (input == null) return null;
        return input.trim().replaceAll("\\\\s+", " ").toLowerCase();
    }
}
"""
with open(f"{package_dir}/StringCleanUDF.java", "w") as f:
    f.write(udf_code)

# Compile with JDK 17 to match Environment Version 4.
subprocess.run(
    ["javac", "--release", "17", "-d", classes_dir, f"{package_dir}/StringCleanUDF.java"],
    check=True,
)

# Package the compiled class into a JAR.
jar_path = f"{build_dir}/string_clean_udf.jar"
subprocess.run(["jar", "cf", jar_path, "-C", classes_dir, "."], check=True)

# Copy the JAR to a Unity Catalog volume.
volume_path = "/Volumes/my_catalog/my_schema/udf_jars/string_clean_udf.jar"
os.makedirs(os.path.dirname(volume_path), exist_ok=True)
shutil.copy2(jar_path, volume_path)

print(f"JAR uploaded to: {volume_path}")

Dopo che il JAR è nel volume, registrare l'UDF. Utilizzare LANGUAGE JAVA e impostare HANDLER sul metodo con nome completo, ad esempio com.databricks.udf.StringCleanUDF.clean.

Registra la tua UDF in Unity Catalog

Dopo aver compilato e caricato il file JAR, usa l'istruzione CREATE FUNCTION per registrare la UDF in Unity Catalog.

Scala

CREATE OR REPLACE FUNCTION my_catalog.my_schema.add_one(x INT)
RETURNS INT
LANGUAGE SCALA
DETERMINISTIC
ENVIRONMENT (
  java_dependencies = '["/Volumes/my_catalog/my_schema/udf_jars/my-udf-assembly-0.1.0-SNAPSHOT.jar"]',
  environment_version = '4'
)
HANDLER 'com.example.MyUDF.addOne';

Java

CREATE OR REPLACE FUNCTION my_catalog.my_schema.add_one(x INT)
RETURNS INT
LANGUAGE JAVA
DETERMINISTIC
ENVIRONMENT (
  java_dependencies = '["/Volumes/my_catalog/my_schema/udf_jars/my-udf-1.0-SNAPSHOT.jar"]',
  environment_version = '4'
)
HANDLER 'com.example.MyUDF.addOne';

L'istruzione CREATE FUNCTION usa i parametri seguenti:

• LANGUAGE: Il linguaggio della UDF.

• HANDLER: percorso completo del metodo, nel formato 'package.Object.method' (Scala) o 'package.ClassName.method' (Java).

• DETERMINISTIC: dichiara che la funzione restituisce sempre lo stesso output per lo stesso input, abilitando l'ottimizzazione delle query.

  Note

  Rimuovere DETERMINISTIC se la funzione chiama API esterne o ha un altro comportamento non deterministico.

• ENVIRONMENT: definisce l'ambiente di esecuzione per l'UDF.

  • java_dependencies: matrice JSON di percorsi di file JAR nei volumi del catalogo Unity. Questo è il percorso del file copiato nel passaggio precedente. Usare virgolette singole intorno alla matrice e virgolette doppie per i percorsi.
  • environment_version: deve essere '4' o superiore per le UDF Scala e Java. Environment Version 4 specifica Scala 2.13.16 e JDK 17. Consultare le versioni dell'ambiente serverless .

Richiamare la UDF in SQL e nei notebook

Dopo la registrazione, è possibile usare la UDF in query SQL, notebook e viste:

-- Simple select
SELECT my_catalog.my_schema.add_one(5) AS result;

-- With table data
SELECT
  id,
  price,
  currency,
  my_catalog.my_schema.convert_to_usd(price, currency) AS price_usd
FROM my_catalog.my_schema.transactions;

-- Filtering
SELECT *
FROM my_catalog.my_schema.products
WHERE my_catalog.my_schema.convert_to_usd(price, currency) > 100;

-- Aggregation
SELECT
  category,
  SUM(my_catalog.my_schema.convert_to_usd(price, currency)) AS total_usd
FROM my_catalog.my_schema.sales
GROUP BY category;

Governance e condivisione

Usa le autorizzazioni di Unity Catalog per controllare chi può eseguire la tua UDF e fare in modo che sia individuabile in tutta l'organizzazione.

Concedere le autorizzazioni

Usare Catalog Explorer o SQL per concedere le autorizzazioni necessarie ad altri utenti per eseguire le UDF.

Esploratore di cataloghi

1. Nella barra laterale fare clic Catalogo.
2. Selezionare il catalogo, quindi selezionare lo schema che contiene la funzione.
3. Fare clic sul nome della funzione.
4. Nella scheda Autorizzazioni fare clic su Concedi.
5. Selezionare le entità a cui si vuole concedere l'accesso e selezionare l'autorizzazione EXECUTE .
6. Cliccare Conferma.

SQL

Eseguire il comando seguente in un notebook o nell'editor SQL di Databricks per concedere EXECUTE autorizzazioni a un utente o a un gruppo.

-- Grant to a specific user
GRANT EXECUTE ON FUNCTION my_catalog.my_schema.add_one TO `user@example.com`;

-- Grant to a group
GRANT EXECUTE ON FUNCTION my_catalog.my_schema.add_one TO `data-engineers`;

Revocare le autorizzazioni

Usare Esplora cataloghi o SQL per revocare le autorizzazioni ad altri utenti.

Esploratore di cataloghi

1. Nella barra laterale fare clic Catalogo.
2. Selezionare il catalogo, quindi selezionare lo schema che contiene la funzione.
3. Fare clic sul nome della funzione.
4. Nella scheda Autorizzazioni selezionare la casella di controllo accanto all'entità a cui si vuole revocare l'accesso. Fare clic su Revoca.
5. Nella notifica fare clic su Revoca.

SQL

Eseguire il comando seguente in un notebook o nell'editor SQL di Databricks per revocare EXECUTE le autorizzazioni a un utente o a un gruppo.

-- Revoke from specific user
REVOKE EXECUTE ON FUNCTION my_catalog.my_schema.add_one FROM `user@example.com`;

-- Revoke from a group
REVOKE EXECUTE ON FUNCTION my_catalog.my_schema.add_one FROM `data-engineers`;

Scopri le UDF

Per trovare le UDF gestite in Unity Catalog, interrogare la tabella information_schema.routines, sostituendo i valori my_catalog e my_schema:

SELECT
  routine_catalog,
  routine_schema,
  routine_name,
  routine_definition,
  created
FROM system.information_schema.routines
WHERE routine_catalog = 'my_catalog'
  AND routine_schema = 'my_schema';

Aggiorna la tua UDF

Per aggiornare un'UDF esistente di Unity Catalog con nuovo codice:

1. Apportare modifiche al codice in locale.
2. Ricompilare il file JAR con un nuovo numero di versione.
   • Scala: sbt clean assembly (ad esempio, my-udf-assembly-0.2.0-SNAPSHOT.jar)
   • Java: mvn clean package (ad esempio, my-udf-2.0-SNAPSHOT.jar)
3. Caricare il nuovo file JAR nel volume di Unity Catalog.
4. Usa CREATE OR REPLACE FUNCTION con lo stesso nome della funzione per aggiornare la UDF. Verifica che nel tuo java_dependencies sia referenziato l'ultimo file JAR.

Azure Databricks usa il nuovo codice nella chiamata successiva. Non è necessario riavviare il cluster.

Ottimizzazione delle prestazioni

Latenza di avvio a freddo

La prima chiamata UDF in una sessione inizializza la sandbox isolata, il che aggiunge latenza. Le chiamate successive nella stessa sessione sono più veloci. Tenere conto di questo problema quando si esegue il benchmarking o si progettano carichi di lavoro sensibili alla latenza.

Memorizzazione in cache di calcoli onerosi

Se la tua UDF esegue un'inizializzazione o un calcolo costosi, memorizzane nella cache il risultato in modo da eseguirli una sola volta.

Scala

Usare un val campo nell'oggetto Scala per memorizzare nella cache il risultato:

package example

object CachedUDF {
  // Computed once and cached
  val expensiveData: Map[String, Double] = {
    // Load data from somewhere expensive
    Map("key1" -> 1.0, "key2" -> 2.0)
  }

  def lookup(key: String): Double = {
    expensiveData.getOrElse(key, 0.0)
  }
}

Java

Usare un static campo con un blocco di inizializzatore statico per memorizzare nella cache il risultato:

package example;

import java.util.Map;
import java.util.HashMap;

public class CachedUDF {
    // Computed once and cached
    private static Map<String, Double> expensiveData;

    static {
        // Load data from somewhere expensive
        expensiveData = new HashMap<>();
        expensiveData.put("key1", 1.0);
        expensiveData.put("key2", 2.0);
    }

    public static double lookup(String key) {
        return expensiveData.getOrDefault(key, 0.0);
    }
}

Usa DETERMINISTIC quando opportuno

Contrassegna la tua UDF come DETERMINISTIC se produce sempre lo stesso output per lo stesso input. Ciò consente a Query Optimizer di memorizzare nella cache i risultati e migliorare le prestazioni.

Limitations

• Sono supportate solo le UDF scalari. Le funzioni di aggregazione definite dall'utente (UDAFs) e le funzioni di tabella definite dall'utente non sono supportate.
• Le UDF vengono eseguite in una sandbox isolata in assenza di una sessione Spark attiva. Le API Spark (SparkSession, SparkContext, spark.sql(...), DataFrame e Set di dati) non sono disponibili.
• Le UDF non possono dipendere dai pacchetti core di Spark.
• Le funzioni definite dall'utente non hanno accesso ai file dell'area di lavoro o ai volumi del catalogo Unity in fase di esecuzione.

Procedure consigliate

Databricks consiglia le procedure seguenti:

• Versione dei file JAR. Per esempio, my-udf-0.1.0.jar, my-udf-0.2.0.jar.
• Convalidare i mapping dei tipi SQL prima della distribuzione. Vedi Mappature della lingua.
• Concedere le autorizzazioni READ VOLUME e EXECUTE solo agli utenti che devono eseguire l'UDF. Utilizza la proprietà di gruppo per le UDF condivise tra i team.

Test delle UDF in locale

Testa la tua UDF con test unitari prima della distribuzione in produzione.

Scala

Per testare src/main/scala/example/MyUDF.scala, creare un file di test in src/test/scala/example/MyUDFTest.scala:

package example

import org.scalatest.funsuite.AnyFunSuite

class MyUDFTest extends AnyFunSuite {
  test("addOne should add 1 to input") {
    assert(MyUDF.addOne(5) == 6)
  }

  test("addOne should handle negative numbers") {
    assert(MyUDF.addOne(-1) == 0)
  }
}

Aggiungere la dipendenza di test a build.sbt:

libraryDependencies += "org.scalatest" %% "scalatest" % "3.2.15" % Test

Per eseguire i test:

sbt test

Java

Per testare src/main/java/com/example/MyUDF.java, creare un file di test in src/test/java/com/example/MyUDFTest.java:

package com.example;

import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;

public class MyUDFTest {
    @Test
    public void testAddOne() {
        assertEquals(6, MyUDF.addOne(5));
    }

    @Test
    public void testAddOneWithNegativeNumbers() {
        assertEquals(0, MyUDF.addOne(-1));
    }
}

Aggiungi la dipendenza JUnit nella sezione <dependencies> del tuo pom.xml:

<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.10.0</version>
    <scope>test</scope>
</dependency>

Per eseguire i test:

mvn test

Risorse aggiuntive

• UDF Scala e Java con ambito di sessione
• Funzioni definite dall'utente (UDF) SQL e Python in Unity Catalog
• Che cosa sono le funzioni definite dall'utente ??
• Tipo di dati
• Gestire i privilegi nel Catalogo Unity