メインコンテンツまでスキップ
に最終更新

Unity Catalog の Scala および Java のユーザー定義関数 (UDF)

このページでは、Scala および Java のユーザー定義関数 (UDF) を作成し、Unity Catalog に登録し、コンピュート環境全体で共有する方法について説明します。Unity Catalog UDF は、既存の JVM ロジックを Unity Catalog のガバナンスとアクセス制御で再利用できます。

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

  • **ガバナンス対象**：Unity Catalogの権限とアクセス制御によって管理されます。
  • 再利用可能 ：チーム、ノートブック、ジョブ、SQL Warehouse 間で共有できます。
  • 検出可能 : Catalog Explorer およびシステムテーブルに表示されます。
  • 分離 : セッションごとに 1 回のコールドスタート コストが発生するサンドボックスでランします。後続の呼び出しは高速です。

要件​

ワークスペースで Unity Catalog が有効になっている必要があります。次の追加要件が適用されます。

コンピュート ：Serverlessノートブックやジョブ、SQLウェアハウス、Lakeflow上のSpark宣言型パイプラインなど、すべてのコンピュートタイプがサポートされています。Classic コンピュートには Databricks Runtime 18.2 以降が必要です。ServerlessコンピュートとSQL Warehouseでは、UDFの定義はenvironment_versionフィールドで環境バージョン4以上を指定する必要があります。この要件はUDFの定義に適用され、呼び出し元のノートブックやジョブには適用されません。Serverless環境のバージョンを参照してください。

開発 ：

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

**権限**：

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

Unity Catalogの権限に関する詳細情報については、Unity Catalogでの権限の管理を参照してください。

UDF JARを構築する​

UDF を登録する前に、コンパイル済みのコードを JAR としてパッケージ化し、Unity Catalog ボリュームに upload します。ビルド方法を選択してください:

ローカルでビルド​

ローカル開発環境を使用してfat JARをビルドするには、これらのステップに従ってください。

環境をセットアップする​

必要なツールをローカルマシンにインストールしてください。以下のコマンドはmacOS用です。他のプラットフォームの場合、プラットフォームのパッケージマネージャーを使用してJDK 17とsbt (Scala)またはMaven (Java)をインストールしてください。

JDK 17 と sbt をインストールする:

Bash
brew install openjdk@17
brew install sbt

インストールを確認します：

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

プロジェクトを作成​

ScalaまたはJavaでプロジェクトを設定してください。

sbt を使用して新しい Scala プロジェクトを作成します：

Bash
sbt new scala/scala-seed.g8

プロンプトが表示されたら、プロジェクト名を入力します（例: my-udf-project）。

build.sbt を構成します。

お客様のbuild.sbtファイルの内容を次の構成に置き換えてください。

Scala
scalaVersion := "2.13.16"

ThisBuild / organization := "com.example"

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

SBT-assembly プラグインを有効にする

project/assembly.sbt を作成または編集して追加します:

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

このプラグインは、すべての依存関係を含む fat JAR を作成します。

UDF を記述します​

UDFを作成する際は、サポートされているデータ型についてはデータ型を、ScalaおよびJavaの型がSQL型にどのようにマッピングされるかを確認するには言語マッピングを参照してください。

UDF ハンドラーは次の要件を満たす必要があります。

  • **Scala**：ハンドラーを にメソッドとして定義してください（object classではありません）。HANDLERの値は、Scala objectのメソッドに解決されます。
  • Java ：ハンドラーをpublic staticメソッドとして定義します。
  • シグネチャ: メソッドのパラメーター型、順序、および戻り値の型は、引数リストとRETURNS ステートメントのCREATE FUNCTION 型に対応している必要があります。
  • スカラー値のみ ：ハンドラーは単一のスカラー値を返す必要があります。テーブルの戻り値の型はサポートされていません。
  • 自己完結型 : ハンドラーは、その入力引数でのみ動作する必要があります。Spark API を使用したり、Spark コアパッケージに依存したりすることはできません。制限事項を参照してください。
注記

Scala の場合、プリミティブパラメーター型 (たとえば Int) を持つハンドラーはスキップされ、入力引数のいずれかが SQL NULL の場合、NULL を返します。NULL 値を受け取り処理するには、パラメーターを Option で囲みます (例: Option[Int])。

src/main/scala/com/example/MyUDF.scalaでScalaオブジェクトを作成し、UDF関数を定義します。

基本的な例

Scala
package com.example

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

外部依存関係の例

外部ライブラリを使用するには、build.sbt ファイルに追加します。

Scala
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で依存関係を使用します。

Scala
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のローカルテストを参照してください。

注記

UDFは、アクティブなSparkセッションのない分離されたサンドボックスで実行されるため、関数本体内からSpark APIを使用できません。例えば、DataFramesまたはデータセットを作成したり操作したり、spark.sql(...)をランしたり、SparkSessionまたはSparkContextにアクセスすることはできません。UDFは、その入力引数に関して自己完結型のロジックである必要があります。Spark コアパッケージに依存することもできません。

ファット JAR をビルドする​

すべての依存関係を含むfat JARを作成するために、プロジェクトを構築します。

プロジェクトのルートディレクトリから、実行します:

Bash
sbt clean assembly

fat JAR は target/scala-2.13/ に作成され、my-udf-assembly-0.1.0-SNAPSHOT.jar のような名前になります。

upload your JAR to a Unity Catalog volume​

まだ Unity Catalogボリュームがない場合は、作成してください:

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

他のユーザーがUDFをランする必要がある場合は、ボリュームに対してREAD VOLUMEを付与します:

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

カタログエクスプローラーを使用して、JARファイルをボリュームにuploadします:

  1. Databricks ワークスペースで、データアイコン。[カタログ]をクリックして カタログエクスプローラー を開きます。
  2. カタログを選択し、次にボリュームを含むスキーマを選択します。
  3. ボリューム名をクリックします。
  4. このボリュームにupload をクリックし、JARファイルを選択します。
  5. Click upload .
  6. upload が完了したら、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としてパッケージ化して、Databricksノートブックから直接Unity Catalogボリュームにuploadできます。このアプローチは、小規模で依存関係のないUDFに有効です。サードパーティ ライブラリを使用する UDF には、ローカルでビルドを使用します。

以下の Python セルは、文字列をクリーンアップする (空白をトリミングし、繰り返されるスペースをまとめ、小文字にする) Java UDF を記述し、それを JDK 17 でコンパイルし、JAR としてパッケージ化し、Unity Catalog ボリュームにコピーします。WRITE VOLUME 権限を持つ既存のボリュームを指すように、volume_path を更新してください。

Python
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などの完全修飾メソッドに設定します。

UDF を Unity Catalog に登録します​

JAR をビルドしてuploadしたら、CREATE FUNCTION ステートメントを使用して Unity Catalog に UDF を登録します。

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

CREATE FUNCTIONステートメントは次のパラメーターを使用します:

  • LANGUAGE：UDFの言語。

  • HANDLER：メソッドへの完全修飾パスです。形式は 'package.Object.method'（Scala）または 'package.ClassName.method'（Java）です。

  • DETERMINISTIC関数は常に同じ入力に対して同じ出力が返されることを宣言し、クエリー最適化を可能にします。

注記

関数が外部APIsを呼び出すか、その他の非決定的な動作がある場合は、DETERMINISTICを削除します。

  • ENVIRONMENT: UDFの実行環境を定義します。

    • java_dependencies：Unity Catalog ボリューム内の JAR ファイル パスの JSON 配列。これは、前のステップでコピーしたファイル パスです。配列はシングルクォーテーションで囲み、パスはダブルクォーテーションで囲みます。
    • environment_version: Scala および Java UDF の場合、'4' 以上である必要があります。環境バージョン 4 は Scala 2.13.16 および JDK 17 を指定します。Serverless 環境バージョンを参照してください。

SQL とノートブックで UDF を呼び出します​

登録後、SQLクエリー、ノートブック、およびビューでUDFを呼び出すことができます。

SQL
-- 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 Catalog の権限を使用して、UDF をランできるユーザーを制御し、組織全体で検出可能にします。

権限を付与​

Catalog Explorer または SQL を使用して、他のユーザーが UDF をランするために必要な権限を付与します。

  1. サイドバーで [データアイコン。 カタログ ]をクリックします。
  2. カタログを選択し、次に、関数を含むスキーマを選択してください。
  3. 関数名をクリックします。
  4. 「アクセス許可」tabで、「許可」をクリックしてください。
  5. アクセス権を付与するプリンシパルを選択し、EXECUTE 権限を選択します。
  6. 確認 をクリックします。

権限を取り消す​

Catalog Explorer または SQL を使用して、他のユーザーから権限を取り消します。

  1. サイドバーで [データアイコン。 カタログ ]をクリックします。
  2. カタログを選択し、次に、関数を含むスキーマを選択してください。
  3. 関数名をクリックします。
  4. Permissions tabで、アクセスを取り消す対象のプリンシパルの横にあるチェックボックスを選択します。 取り消し をクリックします。
  5. 通知で、 取り消し をクリックします。

UDF​

Unity Catalogで管理されているUDFを見つけるには、information_schema.routinesテーブルをクエリーし、my_catalogとmy_schemaの値を置き換えます。

SQL
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 Catalog 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 Catalog ボリュームに upload します。

  4. CREATE OR REPLACE FUNCTION を使用して、同じ関数名で UDF を更新します。最新の JAR が java_dependencies で参照されていることを確認してください。

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

パフォーマンスの最適化​

コールド**起動**のレイテンシー​

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

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

UDFでコストのかかる初期化や計算が実行される場合は、結果をキャッシュして1回だけコンピュートします。

結果をキャッシュするには、Scalaオブジェクトのvalフィールドを使用します。

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

必要に応じて DETERMINISTIC を使用してください。​

同じ入力に対して常に同じ出力が生成される場合、UDF を DETERMINISTIC としてマークします。これにより、クエリーオプティマイザーは結果をキャッシュし、パフォーマンスを向上させることができます。

制限事項​

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

ベストプラクティス​

Databricksでは、以下のプラクティスを推奨します：

  • JAR ファイルにバージョンを付けます。たとえば、my-udf-0.1.0.jar、my-udf-0.2.0.jarです。
  • デプロイの前にSQL 型マッピングを検証してください。言語マッピングを参照してください。
  • UDF を実行する必要があるユーザーにのみ、READ VOLUME と EXECUTE の権限を付与してください。チーム間で共有される UDF には、グループ所有権を使用します。

UDFをローカルでテストする​

UDFを本番運用にデプロイする前に、ユニットテストでテストします。

src/main/scala/example/MyUDF.scalaをテストするには、src/test/scala/example/MyUDFTest.scalaにテストファイルを作成してください：

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 に追加します:

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

テストを実行するには：

Bash
sbt test

その他のリソース​

このページの見出し