ユーザー定義関数 (UDF) とは何ですか?
ユーザー定義関数（UDF）を使用すると、Databricks の組み込み機能を拡張するコードを再利用および共有できます。UDF を使用して、複雑な計算、変換、カスタムデータ操作などの特定のタスクを実行します。
UDF と Apache Spark 関数をいつ使用するか
UDFs は、組み込みの Apache Spark 関数では表現しにくいロジックに使用します。組み込みの Apache Spark 関数は、分散処理用に最適化されており、大規模でより優れたパフォーマンスを提供します。詳細については、「 関数」を参照してください。
Databricksは、アドホッククエリー、手動データクリーンアップ、探索的データ分析、および中小規模のデータセットでの操作にUDFを推奨します。UDFの一般的なユースケースには、データ暗号化、復号化、ハッシュ化、JSON解析、検証などがあります。
大規模なデータセットでの操作、および ETL ジョブやストリーミング操作を含む、定期的または継続的に実行されるあらゆるワークロードに Apache Spark メソッドを使用します。
UDF タイプを理解する
次のタブから UDF タイプを選択すると、説明、例、および詳細を確認するためのリンクが表示されます。
- Scalar UDF
- Batch Scalar UDFs
- Non-Scalar UDFs
- UDAF
- UDTFs
スカラー UDF は 1 つの行で動作し、各行に対して 1 つの結果値を返します。これらは、Unity Catalog で管理することも、セッション スコープにすることもできます。
次の例では、スカラー UDF を使用して、 name 列の各名前の長さを計算し、新しい列 name_lengthに値を追加します。
+-------+-------+
| name | score |
+-------+-------+
| alice | 10.0 |
| bob | 20.0 |
| carol | 30.0 |
| dave | 40.0 |
| eve | 50.0 |
+-------+-------+
-- Create a SQL UDF for name length
CREATE OR REPLACE FUNCTION main.test.get_name_length(name STRING)
RETURNS INT
RETURN LENGTH(name);
-- Use the UDF in a SQL query
SELECT name, main.test.get_name_length(name) AS name_length
FROM your_table;
+-------+-------+-------------+
| name | score | name_length |
+-------+-------+-------------+
| alice | 10.0 | 5 |
| bob | 20.0 | 3 |
| carol | 30.0 | 5 |
| dave | 40.0 | 4 |
| eve | 50.0 | 3 |
+-------+-------+-------------+
これをDatabricksノートブックでPySparkを使用して実装するには:
from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType
@udf(returnType=IntegerType())
def get_name_length(name):
return len(name)
df = df.withColumn("name_length", get_name_length(df.name))
# Show the result
display(df)
「Unity CatalogのSQLおよびPythonユーザー定義関数 (UDF)」と「Python スカラーユーザー定義関数 (UDF)」を参照してください。
1:1の入力/出力行パリティを維持しながら、データをバッチで処理します。これにより、大規模なデータ処理のための行ごとの操作のオーバーヘッドが削減されます。また、バッチ UDF は、バッチ間の状態を維持して、より効率的に実行し、リソースを再利用し、データ チャンク間のコンテキストを必要とする複雑な計算を処理します。
これらは、Unity Catalog で管理することも、セッション スコープにすることもできます。
次のバッチ Unity Catalog Python UDF は、行のバッチの処理中に BMI を計算します。
+-------------+-------------+
| weight_kg | height_m |
+-------------+-------------+
| 90 | 1.8 |
| 77 | 1.6 |
| 50 | 1.5 |
+-------------+-------------+
%sql
CREATE OR REPLACE FUNCTION main.test.calculate_bmi_pandas(weight_kg DOUBLE, height_m DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
PARAMETER STYLE PANDAS
HANDLER 'handler_function'
AS $$
import pandas as pd
from typing import Iterator, Tuple
def handler_function(batch_iter: Iterator[Tuple[pd.Series, pd.Series]]) -> Iterator[pd.Series]:
for weight_series, height_series in batch_iter:
yield weight_series / (height_series ** 2)
$$;
select main.test.calculate_bmi_pandas(cast(70 as double), cast(1.8 as double));
+--------+
| BMI |
+--------+
| 27.8 |
| 30.1 |
| 22.2 |
+--------+
Unity CatalogのSQLおよびPythonユーザー定義関数 (UDF)およびUnity CatalogのバッチPythonユーザー定義関数 (UDF)を参照してください。
非スカラー UDF は、柔軟な入力/出力比 (1 または多く).
セッション スコープのバッチ Pandas UDF には、次の種類があります。
- シリーズからシリーズ
- シリーズのイテレータからシリーズのイテレータへ
- 複数のシリーズのイテレータからシリーズのイテレータへ
- 直列からスカラーへ
次に、シリーズ間 Pandas UDFの例を示します。
from pyspark.sql.functions import pandas_udf
import pandas as pd
df = spark.createDataFrame([(70, 1.75), (80, 1.80), (60, 1.65)], ["Weight", "Height"])
@pandas_udf("double")
def calculate_bmi_pandas(weight: pd.Series, height: pd.Series) -> pd.Series:
return weight / (height ** 2)
df.withColumn("BMI", calculate_bmi_pandas(df["Weight"], df["Height"])).display()
ユーザー定義関数Pandasを参照してください。
UDAFs 複数の行を操作し、1 つの集計結果を返します。 UDAFs はセッション スコープのみです。
次の UDA の例では、名前の長さでスコアを集計します。
from pyspark.sql.functions import pandas_udf
from pyspark.sql import SparkSession
import pandas as pd
# Define a pandas UDF for aggregating scores
@pandas_udf("int")
def total_score_udf(scores: pd.Series) -> int:
return scores.sum()
# Group by name length and aggregate
result_df = (df.groupBy("name_length")
.agg(total_score_udf(df["score"]).alias("total_score")))
display(result_df)
+-------------+-------------+
| name_length | total_score |
+-------------+-------------+
| 3 | 70.0 |
| 4 | 40.0 |
| 5 | 40.0 |
+-------------+-------------+
「Pythonのpandasユーザー定義関数」および「Scalaユーザー定義集計関数 (UDAFs)」を参照してください。
UDTF は 1 つ以上の入力引数を受け取り、入力行ごとに複数の行 (場合によっては複数の列) を返します。これらは、Unity Catalog で管理されるか、セッション スコープで指定できます。
次の UDTF は、2 つの整数引数の固定リストを使用してテーブルを作成します。
CREATE OR REPLACE FUNCTION get_sum_diff(x INT, y INT)
RETURNS TABLE (sum INT, diff INT)
LANGUAGE PYTHON
HANDLER 'GetSumDiff'
AS $$
class GetSumDiff:
def eval(self, x: int, y: int):
yield x + y, x - y
$$;
SELECT * FROM get_sum_diff(10, 3);
+-----+------+
| sum | diff |
+-----+------+
| 13 | 7 |
+-----+------+
これをDatabricksノートブックでPySparkを使用して実装するには:
from pyspark.sql.functions import lit, udtf
@udtf(returnType="sum: int, diff: int")
class GetSumDiff:
def eval(self, x: int, y: int):
yield x + y, x - y
GetSumDiff(lit(1), lit(2)).show()
「 Unity Catalog UDTF と セッション スコープの UDTF」を参照してください。
Unity Catalogによって管理されるUDFとセッションスコープのUDF
Unity Catalog は、ガバナンス、再利用、および検出可能性を向上させるために、Unity Catalog によって管理される UDF を永続化します。現在の SparkSession にスコープ指定されたセッション スコープの UDF をノートブックまたはジョブで定義します。SQL、Python、または Scala を使用して、セッションスコープの UDF を定義およびアクセスできます。
次の表を使用して 2 つのカテゴリを判断し、続いてそれぞれの UDF タイプについての以下のチートシートを参照してください。
考慮事項 | Unity Catalog で管理される UDF。 | セッションスコープのUDF |
|---|---|---|
どのようなタスクにベストなのか | チーム、ノートブック、ジョブ、SQL Warehouse間で関数を安全に共有する。 | 単一のノートブックまたはジョブ内での、迅速な反復型開発。 |
言語 | SQL、Python、Scala、Java。 | SQL・Python・Scala。 |
ガバナンスと共有 | Unity Catalogの権限によって管理され、カタログエクスプローラーで検出可能です。 | 現在のSparkSessionにスコープされます。ガバナンスまたは共有されていません。 |
永続性 | Unity Catalog に保存され、セッション間で再利用できます。 | 現在のセッションのみに存在します。 |
Unity Catalog で管理された UDF のチート シート
Unity Catalogで管理されるUDFは、コンピューティング環境全体でカスタム関数を定義、使用、安全に共有、および管理することを可能にします。Unity CatalogのSQLおよびPythonユーザー定義関数 (UDF)を参照してください。
UDFタイプ | 対応コンピュート | 説明 |
|---|---|---|
Unity Catalog Python UDF |
| UDFPythonでUnity Catalog を定義し、ガバナンスのために に登録する。 スカラー UDF は 1 つの行で動作し、各行に対して 1 つの結果値を返します。 |
バッチ Unity Catalog Python UDF |
| UDFPythonでUnity Catalog を定義し、ガバナンスのために に登録する。 複数の値に対するバッチ操作を行い、複数の値を返します。大規模なデータ処理のための行ごとの操作のオーバーヘッドを削減します。 |
Unity Catalog Python UDTF |
| PythonでUDTFを定義し、ガバナンスのためにUnity Catalogで登録する。 UDTF は 1 つ以上の入力引数を受け取り、各入力行に対して複数の行 (場合によっては複数の列) を返します。 |
Unity Catalog Scala または Java UDF |
| ScalaまたはJavaでUDFを定義し、ガバナンスのためにUnity Catalogに登録します。 スカラーUDFは、単一の行で動作し、行ごとに単一の結果値を返します。Scala 2.13.16が必要です。JDK 17および環境バージョン4。 |
セッション スコープ UDFs ユーザー分離コンピュートのチート シート
現在のSparkSessionにスコープされたセッションスコープのUDFを、ノートブックまたはジョブで定義します。SQL、Python、または Scala を使用して、セッションスコープの UDF を定義およびアクセスできます。
UDFタイプ | 対応コンピュート | 説明 |
|---|---|---|
Python スカラー |
| スカラー UDF は 1 つの行で動作し、各行に対して 1 つの結果値を返します。 |
Pandas UDF(ベクトル化) |
| Pandas UDF は Apache 矢印を使用してデータを転送し、 Pandas を使用してデータを操作します。 Pandas UDF は、行ごとのスカラー UDF よりもパフォーマンスを大幅に向上させるベクトル化された操作をサポートしています。 |
Python の UDTF |
| UDTF は 1 つ以上の入力引数を受け取り、各入力行に対して複数の行 (場合によっては複数の列) を返します。 |
Scala スカラー UDFs |
| スカラー UDF は 1 つの行で動作し、各行に対して 1 つの結果値を返します。 |
JAR からの Scala または Java UDF |
|
|
Scala UDAFs |
| UDAFs 複数の行を操作し、1 つの集計結果を返します。 |
パフォーマンスに関する考慮事項
-
Scala UDFs は一般的に Python UDFs よりも高速です。
- 分離されていない Scala UDF は Java 仮想マシン (JVM) で実行されるため、JVM との間でデータを移動するオーバーヘッドを回避できます。
- 分離された Scala UDF は JVM との間でデータを移動する必要がありますが、メモリをより効率的に処理するため、Python UDF よりも高速になる可能性があります。
-
Python UDFとpandas UDFは、データをシリアル化してJVMからPythonインタープリターに移動する必要があるため、Scala UDFよりも遅くなる傾向があります。
- Pandas UDFs は、Apache Arrow を使用してシリアル化コストを削減するため、Python UDF よりも最大 100 倍高速です。