ユーザー定義関数 (UDF) とは何ですか?

ユーザー定義関数 (UDF) は、ユーザーによって定義された関数であり、カスタム ロジックをユーザー環境で再利用できます。 Databricks では、拡張可能なロジックを分散できるように、さまざまな種類の UDF がサポートされています。 この記事では、UDF の一般的な長所と制限事項の一部を紹介します。

注

すべての形式の UDF が Databricks のすべての実行環境で使用できるわけではありません。 Unity Catalogを使用している場合は、 Unity Catalogのユーザー定義関数 (UDF) を参照してください。

UDF の詳細については、次の記事を参照してください。

• Unity Catalogのユーザー定義関数 (UDF)
• PANDAS ユーザー定義関数
• ユーザー定義のスカラー関数 - Python
• Pythonユーザー定義テーブル関数とは何ですか?
• ユーザー定義のスカラー関数 - Scala
• ユーザー定義集計関数 - Scala

シリアル化のペナルティのないカスタムロジックの定義

Databricks は、多くの種類の UDF に関する効率の制限を含め、Apache Spark から UDF 動作の多くを継承しています。 「 どの UDF が最も効率的か」を参照してください。

UDF に関連する潜在的な効率のトレードオフを心配することなく、コードを安全にモジュール化できます。 これを行うには、SQL または Spark DataFramesを使用して、ロジックを一連の Spark 組み込みメソッドとして定義する必要があります。 たとえば、次の SQL 関数と Python 関数は、Spark の組み込みメソッドを組み合わせて、単位変換を再利用可能な関数として定義します。

CREATE FUNCTION convert_f_to_c(unit STRING, temp DOUBLE)
RETURNS DOUBLE
RETURN CASE
  WHEN unit = "F" THEN (temp - 32) * (5/9)
  ELSE temp
END;

SELECT convert_f_to_c(unit, temp) AS c_temp
FROM tv_temp;

def convertFtoC(unitCol, tempCol):
  from pyspark.sql.functions import when
  return when(unitCol == "F", (tempCol - 32) * (5/9)).otherwise(tempCol)

from pyspark.sql.functions import col

df_query = df.select(convertFtoC(col("unit"), col("temp"))).toDF("c_temp")
display(df_query)

上記の UDF を実行するには、 サンプル データを作成します。

どの UDF が最も効率的ですか?

UDF は、コード実行に重大な処理のボトルネックをもたらす可能性があります。 Databricks では、含まれている Apache Spark、SQL、および Delta Lake 構文を使用して記述されたコードに対して、さまざまなオプティマイザーが自動的に使用されます。 カスタム ロジックが UDF によって導入されると、これらのオプティマイザには、このカスタム ロジックに関するタスクを効率的に計画する機能がありません。 さらに、JVM の外部で実行されるロジックには、データのシリアル化に関する追加コストがかかります。

注

Databricksは、Photon対応のコンピュートを使用する場合、Photonを使用して多くの機能を最適化します。 DataFrame コマンドのSpark SQLをチェーンする関数のみがPhotonによって最適化できます。

一部の UDF は、他の UDF よりも効率的です。 パフォーマンスの観点から:

• 組み込み関数は、Databricks オプティマイザーにより最速になります。

• JVM (Scala、Java、Hive UDF) で実行されるコードは、Python UDF よりも高速になります。

• Pandas UDF は Arrow を使用して、Python UDF に関連するシリアル化コストを削減します。

• Python UDF は手続き型ロジックでは適切に機能しますが、大規模なデータセットの運用 ETL ワークロードでは避ける必要があります。

注

Databricks Runtime 14.0 以下では、Python スカラー UDF と Pandas UDF は、共有アクセス モードを使用するクラスターの Unity Catalog ではサポートされていません。 これらの UDF は、Databricks Runtime 14.1 以降のすべてのアクセス モードでサポートされています。

Databricks Runtime 14.1 以前では、共有アクセス モードを使用するクラスターの Unity Catalog で Scala スカラー UDF はサポートされていません。 これらの UDF は、Databricks Runtime 14.2 以降のすべてのアクセス モードでサポートされています。

Databricks Runtime 14.1 以降では、SQL 構文を使用してスカラー Python UDF を Unity Catalog に登録できます。 「 Unity Catalog のユーザー定義関数 (UDF)」を参照してください。

タイプ	最適化	実行環境
Hive UDF	いいえ	JVM
Python UDF	いいえ	Python
Pandas UDF	いいえ	Python (矢印)
Scala UDF	いいえ	JVM
Spark SQL	はい	JVM (Photon)
Spark DataFrame	はい	JVM (Photon)

どのような場合に UDF を使用する必要がありますか?

UDF の主な利点は、ユーザーが使い慣れた言語でロジックを表現できるため、コードのリファクタリングに関連する人的コストが削減されることです。 アドホック クエリー、手動データクレンジング、探索的データ分析、および小規模または中規模のデータセットに対するほとんどの操作では、UDF に関連する待機時間のオーバーヘッド コストがコードのリファクタリングに関連するコストを上回ることはほとんどありません。

ETL ジョブ、ストリーミング操作、非常に大きなデータセットに対する操作、または定期的または継続的に実行されるその他のワークロードの場合、ネイティブの Apache Spark メソッドを使用するようにロジックをリファクタリングすると、すぐに効果が得られます。

UDFなどのデータの例

この記事のコード例では、UDF を使用して摂氏と華氏の間の温度を変換します。 これらの関数を実行する場合は、次の Python コードを使用してサンプル データセットを作成できます。

import numpy as np
import pandas as pd

Fdf = pd.DataFrame(np.random.normal(55, 25, 10000000), columns=["temp"])
Fdf["unit"] = "F"

Cdf = pd.DataFrame(np.random.normal(10, 10, 10000000), columns=["temp"])
Cdf["unit"] = "C"

df = spark.createDataFrame(pd.concat([Fdf, Cdf]).sample(frac=1))

df.cache().count()
df.createOrReplaceTempView("tv_temp")