Table-Valued Functions をもっと簡単に

Table-Valued Functions（TVF）は、構造化データの処理において長年強力なツールとして活用されてきました。これらの関数は、単一の値ではなく複数の行と列を返すことができます。これまでは、Apache Spark™ でTVFを使用するにはSQLが必要であったため、DataFrame APIを好むユーザーにとっては柔軟性に欠けていました。

今回、Databricks Runtime（DBR）16.1以降で利用可能な「Table-Valued Functionsのための新しいDataFrame API」を発表できることを嬉しく思います。これにより、TVFをDataFrame操作の中で直接呼び出せるようになり、変換処理がよりシンプルで、組み合わせやすくなり、SparkのDataFrameワークフローにも完全に統合されました。

本ブログでは、TVFとは何か、スカラ引数やテーブル引数の両方を使った活用方法について解説します。TVFを利用することで得られる主なメリットは次の3点です。

主なメリット

• DataFrameとのネイティブ統合：SQLを使わずに spark.tvf.<関数名> を使ってTVFを直接呼び出すことが可能。

• チェーン可能で柔軟な組み合わせ：.filter() や .select() など、お気に入りのDataFrame変換操作とTVFを簡単に組み合わせ可能。

• Lateral Joinのサポート（DBR 17.0以降）：各行のデータに基づいてTVFを使って行を動的に生成・展開する結合が可能。

Table-Valued Function DataFrame APIの使い方

まずは、組み込みのTVFを使ったシンプルな例から始めましょう。Sparkには、たとえば variant_explode のように、JSON構造を複数行に展開する便利なTVFが用意されています。

以下は、従来のSQLを使った実装方法です：

そしてこちらが、同じ処理をDataFrame APIで行った例です：

ご覧の通り、TVFはSQLでもDataFrame APIでも簡単に使うことができます。どちらの方法でも、スカラ引数を用いた場合は同じ結果が得られます。

テーブル引数の受け入れ

では、入力としてテーブル（DataFrame）を渡したい場合はどうでしょうか？
これは、データの各行に対して処理を行いたいときに非常に便利です。

ここでは、車と飛行機による移動の所要時間と費用を計算したいケースを例に見てみましょう。

まずは、以下のようなシンプルなDataFrameを想定します：

クラスに、テーブルの行（Row）を引数として受け取れるようにする必要があります。
ここで注目すべき点は、eval メソッドがスカラ引数ではなく、テーブルの Row を受け取るようになっていることです。

このようにテーブルの Row を扱えるように定義することで、任意のDataFrameをテーブル引数として渡し、目的の結果を計算できるようになります。

また、テーブルを作成し、UDTFを登録したうえで、SQLステートメント内で使用することも可能です。

あるいは、スカラ引数を扱う場合には、LATERAL JOIN を使ってTVFを呼び出すことでも同様の結果が得られます（以下の例をご参照ください）。

次のステップ：LATERAL JOIN の活用

TVFをDataFrame全体に対して 1行ずつ 呼び出すには、LATERAL JOIN を使う方法もあります。
LATERAL JOIN とテーブル引数の両方のサポートは、DBR 17.0 以降で利用可能です。

LATERAL JOIN を使うことで、各行の値に基づいて、TVFがその行に対して動的にデータを展開できるようになります。
ここでは、複数行の例をいくつか見ていきましょう。

組み込みTVFを使ったLATERAL JOIN

たとえば、各行に数値の配列を含むDataFrameがあるとします。
この配列を、前述の variant_explode を使って、要素ごとに行へ展開することができます。

以下は、そのためのSQLの書き方です：

そしてこちらが、同じ処理をDataFrame APIで実行する方法です：

Python UDTF を使った LATERAL JOIN

組み込みのTVFだけでは処理が足りないケースもあります。
データを特定の方法で変換したい場合には、カスタムロジックが必要になることもあるでしょう。
そんなときに活躍するのが ユーザー定義テーブル関数（UDTF） です。

Python UDTFs を使えば、自分だけのTVFをPythonで定義でき、行の展開処理を自在にコントロールできます。

以下は、開始値から終了値までの数列を生成し、その値とその2乗を返す、シンプルなPython UDTFの例です：

では、このUDTFをLATERAL JOINで使ってみましょう。
たとえば、start 列と end 列を持つDataFrameがあるとします。
各行ごとに、指定された範囲の数列を生成したい場合に、このUDTFを活用します。

以下は、UDTFを lateralJoin とともに使う別のわかりやすい例です（公式ドキュメントはこちら）。
都市間の距離を含むDataFrameがあるとして、それぞれの都市間を車や飛行機で移動する際の所要時間や航空運賃など、追加の情報を付け加えた新しいテーブルを生成したいとします。

ここでは、先ほど登場した「航空距離のDataFrame」を使ってみましょう。

先ほどの Python UDTF を少し修正し、2都市間の移動時間と費用を計算できるようにします。
具体的には、eval メソッドが スカラ引数 を受け取るように定義します。

そして最後に、この UDTF を lateralJoin で呼び出すことで、目的の出力が得られます。
先ほどの航空距離の例と異なり、今回のUDTFでは eval メソッドがスカラ引数を受け取る点にご注目ください。

まとめ

Table-Valued Functions（TVF）向けの DataFrame API により、Sparkにおけるデータ変換がより統一的で直感的に行えるようになりました。
本記事では、TVFを活用する3つのアプローチ（SQL、DataFrame、Python UDTF）をご紹介しました。

TVFをDataFrame APIと組み合わせることで、複数行のデータ処理や一括変換を簡潔に実現できます。

さらに、テーブル引数を渡したり lateralJoin を使ってPython UDTFを活用することで、個別のビジネスロジックを柔軟に組み込むことも可能です。
この記事では、スカラ引数とテーブル引数の両方を使った2つの具体的な変換・拡張の例を紹介しました。

新しいAPIの活用により、データ変換やワークフローをさらに最適化していくことができるでしょう。
この機能は Apache Spark™ 4.0.0リリース で利用可能で、Databricksをご利用の方は DBR 16.1以降 でお使いいただけます。