セッションスコープの Scala および Java UDF

備考

Scala および Java UDF は、ガバナンス、再利用、および検索機能のために Unity Catalog に登録できます。「Unity Catalog の Scala および Java ユーザー定義関数 (UDF)」を参照してください。

このページでは、DatabricksでセッションスコープのScalaおよびJava UDFを作成する方法について説明します。セッションスコープのUDFは、ノートブックまたはジョブで定義され、現在のSparkSessionにのみ適用されます。SQL言語リファレンスについては、外部ユーザー定義スカラー関数 (UDF) を参照してください。

アプローチを選択してください

ScalaまたはJava UDFは、次の方法で定義できます。言語、ガバナンス、コンピュート間でUDFのすべてのタイプを比較するには、Unity Catalog のガバナンス対象 UDF とセッションスコープの UDF を参照してください。

アプローチ	説明
インライン Scala UDF	Scala関数またはラムダを使用して、ノートブックでUDFを定義します。セッションスコープです。Serverless コンピュートではサポートされていません。
JARからのJava UDF	`spark.udf.registerJavaFunction` を使用して、JAR から事前コンパイルされた UDF クラスを登録します。セッションスコープです。Serverless コンピュートでサポートされています。
Unity Catalog で統制された Scala または Java UDF	ガバナンス、再利用、検出可能性のために、UDF を Unity Catalog に登録します。Serverless コンピュートでサポートされています。

アプローチ	説明
インライン Scala UDF	Scala関数またはラムダを使用して、ノートブックでUDFを定義します。セッションスコープです。Serverless コンピュートではサポートされていません。
JARからのJava UDF	`spark.udf.registerJavaFunction` を使用して、JAR から事前コンパイルされた UDF クラスを登録します。セッションスコープです。Serverless コンピュートでサポートされています。
Unity Catalog で統制された Scala または Java UDF	ガバナンス、再利用、検出可能性のために、UDF を Unity Catalog に登録します。Serverless コンピュートでサポートされています。

要件

標準アクセスモードの Unity Catalog 有効化コンピュートで Scala UDF を使用するには、Databricks Runtime 14.2 以降が必要です。
Unity Catalog 対応クラスター上の Scala UDF の ARM インスタンスサポートには、Databricks Runtime 15.2 以上が必要です。
spark.udf.registerJavaFunction を使用して JAR から Java UDF を登録するには、Databricks Runtime 18.3 以降が必要です。JAR から Java UDF を登録するを参照してください。

重要

JARを、それを実行するコンピュートと同じScalaおよびApache Sparkバージョンに対してビルドしてください。不一致により、UDFは登録時または呼び出し時に失敗する可能性があります。

**クラシックコンピュート**：Databricks Runtime バージョンの Scala および Spark のバージョンを合わせます。ご使用のバージョンについては、Databricks Runtime リリースノートのバージョンと互換性の「**システム環境**」セクションを参照してください。たとえば、Databricks Runtime 18.3 は Scala 2.13.16 および Apache Spark 4.0 を使用します。
Serverless コンピュート : 環境バージョンの Scala バージョンと一致させてください。Serverless 環境バージョンを参照してください。

Apache Spark の依存関係を provided とマークし、JAR にバンドルされないようにしてください。UDF が使用するサードパーティの依存関係のみを含めてください。

関数を UDF として登録する

spark.udf.registerを使用してScala関数をUDFとして登録します：

Scala
val squared = (s: Long) => {
  s * s
}
spark.udf.register("square", squared)

Spark SQL で UDF を呼び出す

一時ビューを作成し、その後、SQLクエリーでUDFを呼び出します。

Scala
spark.range(1, 20).createOrReplaceTempView("test")

SQL
%sql select id, square(id) as id_squared from test

UDF とデータフレームの併用

DataFrame API を使用して UDF を呼び出すこともできます。

Scala
import org.apache.spark.sql.functions.{col, udf}
val squared = udf((s: Long) => s * s)
display(spark.range(1, 20).select(squared(col("id")) as "id_squared"))

JAR から Java UDF を登録する

UDF を JAR としてパッケージ化し、spark.addArtifact を使用してセッションに追加し、spark.udf.registerJavaFunction を使用して UDF クラスを登録します。

注記

Databricks Runtime 18.3以降の標準アクセスモードおよびServerlessコンピュートでサポートされています。登録された関数はセッションスコープであり、Unity Catalogには登録されていません。

以下のステップでは、プロジェクトの作成、UDFクラスの作成、ファットJARの構築、およびその登録について説明します。

ステップ1：プロジェクトを作成する

Scala または Java でプロジェクトを設定します。

Scala
Java

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

Bash
sbt new scala/scala-seed.g8

お使いのbuild.sbtファイルの内容を以下に置き換えてください。scalaVersionとspark-sqlのバージョンをコンピュートに合わせます。

Scala
scalaVersion := "2.13.16"

ThisBuild / organization := "com.example"

lazy val myUDF = (project in file("."))
  .settings(
    name := "my-udf",
    libraryDependencies += "org.apache.spark" %% "spark-sql" % "4.0.0" % "provided"
  )

fat JAR をビルドするために、sbt-assembly プラグインを有効にします。project/assembly.sbt を作成または編集して追加します：

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

quickstart archetype を使用して新しいMavenプロジェクトを作成します。

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

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

生成された pom.xml で、<project></project> タグ内に、<properties> ブロックを追加し、maven-shade-plugin を構成してファットJARをビルドします：

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

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

ステップ2: UDFクラスを作成する

UDF クラスは、org.apache.spark.sql.api.java.UDF インターフェイスのいずれかを実装する必要があります（UDF1 から UDF22）。この数値は、UDF が取得する入力引数の数を示します。ロジックを使用して call() メソッドを実装します。

ハンドラーはJavaクラスである必要があります。spark.udf.registerJavaFunction はリフレクションによってクラスをロードするため、パブリックな引数なしコンストラクターを持つトップレベル (または static にネストされた) パブリッククラスである必要があります。Scala class または object はこの要件を満たしておらず、呼び出し時に失敗します。JARはsbtでビルドできますが、UDFクラス自体はJavaで記述する必要があります。

src/main/java/com/example/MyIntegerUDF.java を作成します。

Java
package com.example;

import org.apache.spark.sql.api.java.UDF1;

public class MyIntegerUDF implements UDF1<Integer, Integer> {
  @Override
  public Integer call(Integer x) {
    return x + 1;
  }
}

ステップ3：ファットJARをビルドする

コンパイルした UDF を fat JAR にパッケージ化します。

Scala
Java

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

Bash
sbt clean assembly

ファット JAR は target/scala-2.13/ に my-udf-assembly-0.1.0-SNAPSHOT.jar のような名前で作成されます。

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

Bash
mvn clean package

ファット JAR は target/ に my-udf-1.0-SNAPSHOT.jar のような名前で作成されます。

ステップ 4：JAR を Unity Catalog ボリュームに upload します

コンピュートがアクセスできるように、JAR を Unity Catalog ボリュームに upload してください。ボリュームをお持ちでない場合は、作成してください。

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

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

Databricks ワークスペースで、[カタログ]をクリックして カタログエクスプローラー を開きます。
カタログを選択し、次にボリュームを含むスキーマを選択してください。
ボリューム名をクリックします。
このボリュームにupload をクリックし、JARファイルを選択します。
Click upload .
アップロードが完了したら、JARファイルの名前をクリックし、 パスをコピー をクリックしてボリュームパスをコピーします。たとえば、/Volumes/my_catalog/my_schema/udf_jars/my-udf-assembly-0.1.0-SNAPSHOT.jar。次のステップでこのパスが必要になります。

ステップ5: UDFを登録して呼び出す

ボリュームパスを使用して JAR をセッションに追加し、UDF クラスを登録し、Spark SQL から呼び出します。

Python
# Add the JAR containing your UDF class to the session
spark.addArtifact("/Volumes/my_catalog/my_schema/udf_jars/my-udf-assembly-0.1.0-SNAPSHOT.jar")

# Register the UDF class, providing the SQL function name,
# the fully qualified class name, and the return type
from pyspark.sql.types import IntegerType

spark.udf.registerJavaFunction(
    "my_udf",
    "com.example.MyIntegerUDF",
    IntegerType(),
)

# Call the UDF from Spark SQL
spark.sql("SELECT my_udf(21)").show()

Serverless および標準アクセスモードのコンピュートでは、明示的な戻り値の型を渡す必要があります。戻り値の型を省略すると、UC_COMMAND_NOT_SUPPORTED_IN_SHARED_ACCESS_MODE で失敗します。ユーザー定義集計関数（UDAF）は registerJavaFunction ではサポートされていません。

クエリーは UDF 出力を返し、関数が登録され、呼び出し可能であることを確認します。

Output
+----------+
| my_udf(21)|
+----------+
|        22|
+----------+

評価順序と null チェック

Spark SQL（SQL、DataFrame および Dataset APIs を含む）は、部分式の評価順序を保証しません。Spark は演算子または関数の入力を左から右に評価しません。論理 AND と OR 式には、左から右への短絡セマンティクスはありません。

Boolean式の副作用または評価順序、あるいはWHEREとHAVING句の順序に依存しないでください。クエリーオプティマイザは、これらの式と句を並べ替えることができます。UDFがnullチェックの短絡評価セマンティクスに依存する場合、SparkはnullチェックがUDFの前に実行されることを保証しません。例えば：

Scala
spark.udf.register("strlen", (s: String) => s.length)
spark.sql("select s from test1 where s is not null and strlen(s) > 1") // no guarantee

この WHERE 句は、Spark が null をフィルターで除外した後に strlen UDF を呼び出すことを保証しません。

nullチェックを処理するには、Databricksでは次のいずれかを推奨しています。

UDF自体をNULL対応にし、UDF内でNULLチェックを実行します。
IF式またはCASE WHEN式を使用してヌルチェックを行い、条件分岐でUDFを呼び出します

Scala
spark.udf.register("strlen_nullsafe", (s: String) => if (s != null) s.length else -1)
spark.sql("select s from test1 where s is not null and strlen_nullsafe(s) > 1") // ok
spark.sql("select s from test1 where if(s is not null, strlen(s), null) > 1")   // ok

型付きデータセット API

注記

この機能は、 Databricks Runtime 15.4 以降の標準アクセスモードを使用した Unity Catalog 対応クラスターでサポートされています。

型付きデータセットAPIsを使用して、ユーザー定義関数を持つデータセットに対して、マップ、フィルター、集計などの変換を実行します。

次の例では、map() API を使用して、結果列の数値をプレフィックス付き文字列に変更します：

Scala
spark.range(3).map(f => s"row-$f").show()

この例では map() を使用しますが、filter()、mapPartitions()、foreach()、foreachPartition()、reduce()、および flatMap() などの他の型付きデータセット APIs にも同じパターンが適用されます。

Scala UDF 機能と Databricks Runtime の互換性

以下の機能には、標準 (共有) アクセスモードの Unity Catalog 対応クラスターで Databricks Runtime の最低バージョンが必要です。

機能	最小Databricks Runtimeバージョン
スカラー UDF	Databricks Runtime 14.2
`Dataset.map`、`Dataset.mapPartitions`、`Dataset.filter`、`Dataset.reduce`、 `Dataset.flatMap`	Databricks Runtime 15.4
`KeyValueGroupedDataset.flatMapGroups`, `KeyValueGroupedDataset.mapGroups`	Databricks Runtime 15.4
(ストリーミング) `foreachWriter Sink`	Databricks Runtime 15.4
(ストリーミング) `foreachBatch`	Databricks Runtime 16.1
(ストリーミング) `KeyValueGroupedDataset.flatMapGroupsWithState`	Databricks Runtime 16.2
`spark.udf.registerJavaFunction` （JAR からの Java UDF）	Databricks Runtime 18.3

機能	最小Databricks Runtimeバージョン
スカラー UDF	Databricks Runtime 14.2
`Dataset.map`、`Dataset.mapPartitions`、`Dataset.filter`、`Dataset.reduce`、 `Dataset.flatMap`	Databricks Runtime 15.4
`KeyValueGroupedDataset.flatMapGroups`, `KeyValueGroupedDataset.mapGroups`	Databricks Runtime 15.4
(ストリーミング) `foreachWriter Sink`	Databricks Runtime 15.4
(ストリーミング) `foreachBatch`	Databricks Runtime 16.1
(ストリーミング) `KeyValueGroupedDataset.flatMapGroupsWithState`	Databricks Runtime 16.2
`spark.udf.registerJavaFunction` （JAR からの Java UDF）	Databricks Runtime 18.3

アプローチを選択してください​

要件​

関数を UDF として登録する​

Spark SQL で UDF を呼び出す​

UDF と データフレーム の併用​

JAR から Java UDF を登録する​

ステップ1：プロジェクトを作成する​

ステップ2: UDFクラスを作成する​

ステップ3：ファットJARをビルドする​

ステップ 4：JAR を Unity Catalog ボリュームに upload します​

ステップ5: UDFを登録して呼び出す​

評価順序と null チェック​

型付きデータセット API​

Scala UDF 機能と Databricks Runtime の互換性​