Unity カタログの Scala と Java ユーザー定義関数 (UDF)

このページでは、Scala を作成し、ユーザー定義関数 (UDF) をJavaし、それらを Unity カタログに登録し、コンピューティング環境間で共有する方法について説明します。 Unity カタログ UDF を使用すると、Unity カタログのガバナンスとアクセス制御で既存の JVM ロジックを再利用できます。

1 つのノートブックまたはクラスターに制限されている セッション スコープの Scala UDF とは異なり、Unity カタログに登録されている UDF は次のとおりです。

• 管理: Unity カタログのアクセス許可とアクセス制御で管理されます。
• 再利用可能: チーム、ノートブック、ジョブ、SQL ウェアハウス間で共有されます。
• 検出可能: カタログ エクスプローラーとシステム テーブルに表示されます。
• 分離型: セッションごとに 1 回のみコールドスタートのコストが発生するサンドボックス環境で実行します。 後続の呼び出しは高速です。

Requirements

ワークスペースは、Unity Catalog に対して有効にする必要があります。 次の追加要件が適用されます。

コンピューティング: サーバーレス ノートブックとジョブ、 SQL ウェアハウス、 Lakeflow 上の Spark 宣言パイプラインなど、すべてのコンピューティングの種類がサポートされています。 クラシック コンピューティングには、Databricks Runtime 18.2 以降が必要です。 サーバーレス コンピューティングおよび SQL ウェアハウスでは、UDF 定義で environment_version フィールドに環境バージョン 4 以降を指定する必要があります。 この要件は、呼び出し元のノートブックまたはジョブではなく、UDF 定義に適用されます。 「サーバーレス環境のバージョン」を参照してください。

開発:

• Scala: 2.13.16。 Scala 2.12 はサポートされていません。
• JDK: 17。
• パッケージ化: UDF が使用するすべてのサードパーティ依存関係を含む Fat JAR。

アクセス許可:

• UDF を作成します。スキーマに USAGE して CREATE FUNCTION し、カタログに USAGE します。
• UDF を実行する: 関数に EXECUTE し、スキーマとカタログに USAGE します。
• JAR ファイルにアクセスします。JAR が格納されているボリュームに READ VOLUME します。

Unity カタログのアクセス許可の詳細については、Unity カタログの権限の管理に関するページを参照してください。

UDF JAR をビルドする

コンパイル済みのコードを JAR としてパッケージ化し、UDF を登録する前に Unity カタログ ボリュームにアップロードします。 ビルド方法を選択します。

• ローカルでビルドする
• ノートブックでビルドする

ローカルでビルドする

ローカル開発環境を使用して 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

このコマンドは、 src/main/java ディレクトリと src/test/java ディレクトリを使用して、標準の Maven プロジェクト構造を作成します。

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型に対応している必要があります。
• スカラーのみ: ハンドラーは 1 つのスカラー値を返す必要があります。 テーブルの戻り値の型はサポートされていません。
• 自己完結型: ハンドラーは、その入力引数でのみ動作する必要があります。 Spark API を使用することも、Spark コア パッケージに依存することもできません。 制限事項を参照してください。

Note

Scala の場合、プリミティブ パラメーター型 (Int など) を持つハンドラーはスキップされ、入力引数が SQL NULLの場合はNULLが返されます。 NULL値を受け取って処理するには、パラメーターを Option などのOption[Int]でラップします。

Scala

src/main/scala/com/example/MyUDF.scalaで 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 を使用することはできません。 たとえば、DataFrames やデータセットの作成や操作、 spark.sql(...)の実行、 SparkSession や SparkContextへのアクセスはできません。 UDF は、入力引数に対する自己完結型のロジックである必要があります。 また、Spark コア パッケージに依存することはできません。

fat 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 カタログ ボリュームにアップロードする

Unity カタログ ボリュームをまだお持ちでない場合は、次のものを作成します。

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

他のユーザーが UDF を実行する必要がある場合は、ボリューム上の READ VOLUME を付与します。

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

カタログ エクスプローラーを使用して JAR ファイルをボリュームにアップロードします。

1. Azure Databricks ワークスペースで、Catalog をクリックしてカタログ エクスプローラーを開きます。
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 を登録するときは、このパスが必要です。

ノートブックでビルドする

UDF をコンパイルし、JAR としてパッケージ化し、Azure Databricks ノートブックから直接 Unity カタログ ボリュームにアップロードできます。 この方法は、依存関係のない小規模な UDF に対して機能します。 サード パーティ製ライブラリを含む UDF の場合は、 ローカルでビルドを使用します。

次の Python セルは、文字列をクリーンアップするJava UDF を書き込みます (空白を削除し、スペースを繰り返し折りたたみ、小文字を折りたたむ)、JDK 17 でコンパイルし、JAR としてパッケージ化して、Unity Catalog ボリュームにコピーします。 権限のある既存のボリュームを指すように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 ステートメントを使用して UDF を Unity カタログに登録します。

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: Unity カタログ ボリューム内の JAR ファイル パスの JSON 配列。 これは、前の手順でコピーしたファイル パスです。 配列の周囲に単一引用符を使用し、パスを囲む二重引用符を使用します。
  • environment_version: Scala と Java UDF の場合は、'4'以上である必要があります。 環境バージョン 4 では、Scala 2.13.16 と JDK 17 が指定されています。 「サーバーレス環境のバージョン」を参照してください。

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. [Confirm](確認) をクリックします。

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

UDF を見つける

Unity カタログで管理されている 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. UDF を更新するには、同じ関数名の CREATE OR REPLACE FUNCTION を使用します。 java_dependenciesで最新の JAR を参照していることを確認します。

Azure Databricksは、次の呼び出しで新しいコードを使用します。 クラスターを再起動する必要はありません。

パフォーマンスの最適化

コールド スタート待機時間

セッション内の最初の UDF 呼び出しによって分離サンドボックスが初期化され、待機時間が追加されます。 同じセッションでの後続の呼び出しの方が高速です。 待ち時間の影響を受けやすいワークロードをベンチマークまたは設計するときは、この点を考慮してください。

高コストな計算のキャッシュ

UDF が高価な初期化または計算を実行する場合は、結果をキャッシュして 1 回だけ計算します。

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 としてマークします。 これにより、クエリ オプティマイザーは結果をキャッシュし、パフォーマンスを向上させることができます。

Limitations

• スカラー UDF のみがサポートされています。 ユーザー定義集計関数 (UDF) とユーザー定義テーブル関数 (UDF) はサポートされていません。
• UDF は、アクティブな Spark セッションなしで分離サンドボックスで実行されます。 Spark API (SparkSession、 SparkContext、 spark.sql(...)、DataFrame、データセットの操作) は使用できません。
• UDF は Spark コア パッケージに依存できません。
• UDF は、実行時にワークスペース ファイルまたは Unity カタログ ボリュームにアクセスできません。

ベスト プラクティス

Databricks では、次のプラクティスをお勧めします。

• JAR ファイルのバージョンを設定します。 たとえば、my-udf-0.1.0.jar、my-udf-0.2.0.jar のようになります。
• デプロイ前に SQL 型マッピングを検証します。 言語マッピングを参照してください。
• UDF を実行する必要があるユーザーにのみ、 READ VOLUME と EXECUTE のアクセス許可を付与します。 チーム間で共有される 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));
    }
}

<dependencies>の pom.xml セクションに JUnit 依存関係を追加します。

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

テストを実行するには:

mvn test

その他のリソース

• セッション スコープの Scala と Java UDF
• Unity カタログの SQL および Python ユーザー定義関数 (UDF)
• ユーザー定義関数 (UDF) とは
• データ型
• Unity カタログで権限を管理する