Unity 目錄中的 Scala 與 Java 使用者定義函式(UDF)

本頁說明如何建立 Scala 與 Java 使用者定義函式(UDF)、在 Unity 目錄中註冊,以及在運算環境中共享。 Unity Catalog 的 UDF 允許你重用現有的 JVM 邏輯,並具備 Unity Catalog 治理與存取控制。

與僅限於單一筆記本或叢集的 會話範圍 Scala UDF 不同,Unity 目錄中註冊的 UDF 包括:

  • 管理方式:採用 Unity 目錄權限與存取控制。
  • 可重複使用:跨團隊、筆記本、工作,以及專業與無伺服器 SQL 倉庫共享。
  • 可發現:可在目錄檔案總管及系統資料表中顯示。
  • 隔離:於沙盒中執行,每個工作階段只需承擔一次冷啟動成本。 後續呼叫的速度很快。

Requirements

您的工作區必須啟用 Unity Catalog。 以下額外要求適用。

運算:支援的運算包括無伺服器筆記本與工作、專業及無伺服器 SQL 倉庫,以及 Lakeflow 上的 Spark 宣告式管線。 經典運算需要 Databricks Runtime 18.2 或以上版本。 在無伺服器運算,以及 Pro 和無伺服器 SQL 倉儲上,UDF 定義必須在 environment_version 欄位中指定環境版本 4 或以上。 此要求適用於 UDF 定義,不適用於呼叫筆記本或工作。 請參閱 環境版本。

存取 Unity Catalog 機密的 Scala UDF 必須明確將 environment_version 設定為 '6' 或更高。 專業 SQL 倉庫不支援這些工具。 請參閱 UDF 要求與權限 ,了解計算支援與限制。

發展:

  • Scala:2.13.16。 Scala 2.12 不受支援。
  • JDK:17歲。
  • 封裝:包含 UDF 使用的所有第三方相依性的 fat JAR 檔。

權限:

  • 建立 UDF:在結構描述上使用 USAGE 和 CREATE FUNCTION,並在目錄上使用 USAGE。
  • 執行 UDF:在函式上按一下 EXECUTE,並在綱要和目錄上按一下 USAGE。
  • 存取 JAR 檔案:函式建立者需要 READ VOLUME 在存放 JAR 的磁碟區。 呼叫者的需求取決於環境版本。 請參閱 Unity 目錄卷中相依關係的權限。
  • 宣告祕密:請參閱 UDF 的要求與權限。

欲了解更多 Unity 目錄權限資訊,請參閱 Unity 目錄中的權限 管理。

建置您的 UDF JAR

先把編譯好的程式碼打包成 JAR,然後上傳到 Unity 目錄卷,再註冊 UDF。 選擇建造方法:

本地建設

請依照下列步驟,使用本機開發環境建置 fat JAR。

設定您的環境

在你本地的機器上安裝所需的工具。 以下指令是針對 macOS。 其他平台的話,可以用你平台的套件管理器安裝 JDK 17 和 sbt(Scala)或 Maven(Java)。

Scala

安裝 JDK 17 和 sbt:

brew install openjdk@17
brew install sbt

確認您的安裝:

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

JAVA

安裝 JDK 17 和 Maven:

brew install openjdk@17
brew install maven

確認您的安裝:

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

建立您的專案

用 Scala 或 Java 建立專案。

Scala

請使用以下方式 sbt建立新的 Scala 專案:

sbt new scala/scala-seed.g8

當提示時,輸入專案名稱(例如, my-udf-project)。

設定 build.sbt

請將您的 build.sbt 檔案內容替換為以下配置:

scalaVersion := "2.13.16"

ThisBuild / organization := "com.example"

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

啟用 sbt-assembly 插件

建立或編輯 project/assembly.sbt 並新增:

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

這個外掛會建立一個包含所有相依性的 fat JAR。

JAVA

使用快速入門原型建立一個新的 Maven 專案:

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

此指令建立標準的 Maven 專案結構,包含 src/main/java 和 src/test/java 目錄。

設定 pom.xml

在產生的 pom.xml 檔案中,於 <project></project> 標籤內加入一個含有下列設定的 <properties> 區塊:

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

另外,也請在 <project></project> 標籤內加入一個 <build> 區塊,並使用下列設定:

<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 會建立一個包含你所有相依性的 fat JAR 檔案。

寫你的UDF

撰寫 UDF 時,請參考支援的資料型別與語言映射,以了解 Scala 和 Java 型別如何映射到 SQL 型別。

您的 UDF 處理員必須符合以下要求:

  • Scala:將處理常式定義為 object 上的方法(而非 class 上的方法)。 HANDLER 值會對應到 Scala object 上的一個方法。
  • Java:將處理常式定義為 public static 方法。
  • 簽名:方法的參數類型、順序和回傳類型必須對應於你RETURNS語句中的參數列表和CREATE FUNCTION類型。
  • 僅限純量:處理常式必須傳回單一純量值。 不支援表格回傳類型。
  • 自足:處理常式必須僅根據自身的輸入引數運作。 它無法使用 Spark API,也無法依賴 Spark 核心套件。 請參閱限制。

Note

對於 Scala,具有基本參數型別(例如 Int)的處理常式,當任何輸入引數為 SQL NULL 時,會被略過並回傳 NULL。 要接收並處理 NULL 數值,將參數包裹在 Option中,例如 Option[Int]。

Scala

在 Scala src/main/scala/com/example/MyUDF.scala 建立一個物件並定義你的 UDF 函式。

基本範例

package com.example

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

外部相依的範例

要使用外部函式庫,請將它們加入你的 build.sbt 檔案中:

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

然後在您的 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")
    }
  }
}

部署前先用單元測試測試你的 UDF。 請參閱 本地測試 UDF。

JAVA

在 裡面src/main/java/com/example/MyUDF.java建立一個 Java 類別,並將你的 UDF 定義為公開的靜態方法。

基本範例

package com.example;

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

外部相依的範例

要使用外部函式庫,請將這些函式庫新增至 <dependencies> 檔案的 pom.xml 區段:

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

然後在您的 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);
    }
}

部署前先用單元測試測試你的 UDF。 請參閱 本地測試 UDF。

Note

你的 UDF 運行在一個隔離的沙盒中,沒有活躍的 Spark 會話,因此無法在函式體內使用 Spark API。 例如,你無法建立或操作資料框或資料集,執行 spark.sql(...),或存取 SparkSessionSparkContext。 UDF 必須對其輸入參數具有自包含邏輯。 它也不能依賴 Spark 核心套件。

打造你的胖胖 JAR

建置您的專案,以產生包含所有相依性的 fat JAR。

Scala

從你的專案根目錄執行:

sbt clean assembly

fat JAR 會建立在 target/scala-2.13/ 中,名稱如 my-udf-assembly-0.1.0-SNAPSHOT.jar。

JAVA

從你的專案根目錄執行:

mvn clean package

fat JAR 會建立在 target/ 中,名稱如 my-udf-1.0-SNAPSHOT.jar。

將你的 JAR 上傳至 Unity Catalog 磁碟區

如果您尚未擁有 Unity Catalog 磁碟區,請建立一個磁碟區:

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

如果所選環境版本的 呼叫端權限需求 需要 READ VOLUME,請在該磁碟區上授予它:

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

使用 目錄瀏覽器將您的 JAR 檔案上傳至卷中:

  1. 在您的 Azure Databricks 工作區中,按一下 [資料] 圖示以開啟目錄總管。
  2. 選擇目錄,再選擇包含你卷的結構。
  3. 點擊卷名。
  4. 點擊 「上傳到此卷 」並選擇你的 JAR 檔案。
  5. 按一下 [上傳] 。
  6. 上傳完成後,點擊你的 JAR 檔案名稱。
  7. 按一下 複製路徑,將磁碟區路徑複製到剪貼簿。 例如, /Volumes/my_catalog/my_schema/udf_jars/my-udf-assembly-0.1.0-SNAPSHOT.jar (Scala)或/Volumes/my_catalog/my_schema/udf_jars/my-udf-1.0-SNAPSHOT.jar(Java)。 註冊 UDF 時需要這條路徑。

在 Notebook 中建置

你可以編譯 UDF,打包成 JAR,然後直接從 Azure Databricks 筆記本上傳到 Unity 目錄卷。 此方法適用於小型且無依賴性的 UDF。 對於有第三方函式庫的 UDF,請在 本地使用 Build。

以下的 Python 儲存格會寫入一個 Java UDF,清理字串(裁掉空白、壓縮重複空格和小寫),用 JDK 17 編譯,打包成 JAR,並複製到 Unity 目錄卷。 將 volume_path 更新為指向你具有 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}")

將 JAR 放入磁碟區後,註冊 UDF。 使用 LANGUAGE JAVA 並設定 HANDLER 為完全限定的方法,例如 com.databricks.udf.StringCleanUDF.clean。

在 Unity 目錄中註冊你的 UDF

建置並上傳 JAR 後,使用 CREATE FUNCTION 該語句在 Unity 目錄中註冊你的 UDF。

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';

該 CREATE FUNCTION 陳述使用以下參數:

  • LANGUAGE:UDF的語言。

  • HANDLER:完整限定的方法路徑,格式為 'package.Object.method' Scala 或 'package.ClassName.method' (Java)。

  • DETERMINISTIC宣告函式對相同輸入總是回傳相同輸出,從而實現查詢最佳化。

    Note

    如果你的函式呼叫外部 API 或其他非確定性行為,請移除 DETERMINISTIC 。

  • ENVIRONMENT定義 UDF 的執行環境。

    • java_dependencies: 一個 JSON 陣列,包含 Unity 目錄磁碟區中的 JAR 檔案路徑。 這就是你在前一步複製的檔案路徑。 在陣列周圍用單引號,路徑周圍用雙引號。
    • environment_version:Scala 和 Java UDF 必須為 '4' 或以上。 環境版本 4 規定使用 Scala 2.13.16 與 JDK 17。 請參閱 環境版本。

從 Scala UDF 中存取 Unity Catalog 密鑰

純量 Unity Catalog Scala UDF 可以存取在 SECRETS 子句中宣告的祕密。 此功能在專業 SQL 倉庫中不被支援。

在你的 build.sbt 檔案中加入 Databricks Connect 19.1 或更新版本,並將範圍設為 provided,以便處理常式可依據 Secrets API 進行編譯,而無須將 Databricks Connect 打包到 JAR 中:

libraryDependencies += "com.databricks" %% "databricks-connect" % "19.1" % "provided"

以下處理常式會檢查已宣告的密鑰是否已設定,而不會回傳其值:

package com.example

object SecretUDF {
  def isConfigured(input: Int): Boolean = {
    val apiKey = com.databricks.Secrets.get("main", "default", "api_key")
    apiKey != null && apiKey.nonEmpty
  }
}

建立 JAR,上傳到 Unity 目錄卷,然後註冊這個函式。 您必須明確設定 environment_version 為 '6' 或超過:

CREATE OR REPLACE FUNCTION my_catalog.my_schema.secret_is_configured(input INT)
RETURNS BOOLEAN
LANGUAGE SCALA
SECRETS (main.default.api_key)
ENVIRONMENT (
  java_dependencies = '["/Volumes/my_catalog/my_schema/udf_jars/secret-udf.jar"]',
  environment_version = '6'
)
HANDLER 'com.example.SecretUDF.isConfigured';

Warning

請勿從 UDF 回傳機密值。 秘密塗黑有助於減少錯誤與日誌中的意外暴露,但並不能阻止 UDF 程式碼在查詢結果中揭露機密內容。

關於計算與權限需求,請參見 UDF 要求與權限。

在 SQL 和筆記本中呼叫 UDF

註冊後,你可以在 SQL 查詢、筆記本和檢視中呼叫 UDF:

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

治理與分享

使用 Unity 目錄權限來控制誰能執行你的 UDF,並讓它在整個組織中都能被發現。

授與權限

使用目錄檔案管理器或 SQL 授權其他使用者執行你的 UDF 所需的權限。

目錄檢視器

  1. 在提要欄位中,按兩下 [資料] 圖示。目錄。
  2. 選擇目錄,然後選擇包含你功能的結構。
  3. 按一下函式名稱。
  4. 在 權限 標籤中,點選 授權。
  5. 選擇你想授權的主體,並選擇權限。EXECUTE
  6. 按一下 [確認]。

SQL

在筆記本或 Databricks SQL 編輯器中執行以下指令,授予 EXECUTE 使用者或群組權限。

-- 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`;

撤銷權限

使用目錄檔案管理器或 SQL 來撤銷其他使用者的權限。

目錄檢視器

  1. 在提要欄位中,按兩下 [資料] 圖示。目錄。
  2. 選擇目錄,然後選擇包含你功能的結構。
  3. 按一下函式名稱。
  4. 在 權限 標籤中,選擇你想撤銷存取權的主體旁的勾選框。 按兩下 [ 撤銷]。
  5. 在通知中,點擊 撤銷。

SQL

請在筆記本或 Databricks SQL 編輯器中執行以下指令,以撤銷 EXECUTE 使用者或群組的權限。

-- 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`;

探索使用者定義函式

要找到 Unity Catalog 管理的 UDF,請查詢 information_schema.routines 表格,替換 my_catalog 和 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';

更新你的 UDF

要用新程式碼更新現有的 Unity 目錄 UDF:

  1. 在本地修改程式碼。
  2. 用新的版本號重建 JAR。
    • Scala: sbt clean assembly (例如, my-udf-assembly-0.2.0-SNAPSHOT.jar)
    • Java: mvn clean package (例如,my-udf-2.0-SNAPSHOT.jar)
  3. 將新的 JAR 上傳到 Unity 目錄卷。
  4. 用 CREATE OR REPLACE FUNCTION 同一個函式名稱來更新 UDF。 請確認你在 java_dependencies 中參照的是最新的 JAR。

Azure Databricks 在下一個調用時使用新程式碼。 你不需要重新啟動叢集。

效能優化

冷啟動延遲

會話中的第一個 UDF 呼叫會初始化隔離的沙箱,這會增加延遲。 同一會話中後續呼叫會更快。 在進行基準測試或設計延遲敏感的工作負載時,請考慮這一點。

快取高成本運算

如果你的 UDF 會執行高成本的初始化作業或運算,請將結果快取起來,讓其只需計算一次。

Scala

在 Scala 物件中使用 val 欄位來快取結果:

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

使用 static 帶有靜態初始化區塊的欄位來快取結果:

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);
    }
}

適當時使用 DETERMINISTIC

如果您的 UDF 對相同的輸入一律產生相同的輸出,請將其標記為 DETERMINISTIC。 這讓查詢優化器能夠快取結果並提升效能。

限制

  • 僅支援純量 UDF。 不支援使用者定義的聚合函數(UDAFs)及使用者定義的表格函數(UDTF)。
  • UDF 運行在一個孤立的沙盒中,沒有主動的 Spark 會話。 Spark API(SparkSessionSparkContextspark.sql(...)、、DataFrame 及 Dataset 操作)無法使用。
  • UDF 不能依賴 Spark 核心套件。
  • UDF 在執行時無法存取工作區檔案或 Unity 目錄卷。

最佳做法

Databricks 建議以下做法:

  • 修改你的 JAR 檔案版本。 例如,my-udf-0.1.0.jar、my-udf-0.2.0.jar。
  • 部署前驗證 SQL 型別映射。 參見 語言映射。
  • 僅將 EXECUTE 授予需要執行 UDF 的使用者。 使用群組所有權來管理跨團隊共享的 UDF。

在本機測試 UDF

部署到生產環境前,先用單元測試測試你的 UDF。

Scala

要測試 src/main/scala/example/MyUDF.scala,建立一個測試檔案: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)
  }
}

將測試相依性新增至 build.sbt:

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

若要執行測試:

sbt test

JAVA

要測試 src/main/java/com/example/MyUDF.java,建立一個測試檔案: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));
    }
}

將 JUnit 相依性加入你的 <dependencies> 的 pom.xml 區段:

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

若要執行測試:

mvn test

其他資源