Unity CatalogにおけるPythonユーザー定義テーブル関数（UDTF）の紹介

Python UDFを使用すると、カスタムロジックの抽象化レイヤーを構築して、クエリーの構築を簡素化できます。しかし、大規模なモデルを実行したり、テーブルの行をまたいで効率的にパターンを検出したりするなど、複雑なロジックを適用したい場合はどうでしょうか？

以前、より強力なカスタムクエリーロジックをサポートするために、セッションスコープのPythonユーザー定義テーブル関数（UDTF）を紹介しました。UDTFを使用すると、テーブル全体に対して堅牢でステートフルなPythonロジックをランできるため、純粋なSQLでは通常解決が困難な問題を簡単に解決できます。

ユーザー定義テーブル関数を使用する理由:

• あらゆるデータセットを柔軟に処理

  宣言的な TABLE() キーワードを使用すると、テーブル、ビュー、さらには動的なサブクエリを UDTF に直接パイプできます。これにより、関数がデータのあらゆるスライスに対応する、強力で再利用可能なビルディングブロックになります。Python関数内で直接、独立した関数呼び出しによって処理される行のサブセットに入力テーブルをパーティション分割するために、PARTITION BY、ORDER BY、およびWITH SINGLE PARTITION を使用することもできます。

• ラン負荷の高い初期化をパーティションごとに一度だけ

  UDTF を使用すると、大規模な機械学習モデルや大きな参照ファイルの読み込みなど、負荷の高いセットアップ コードを、行ごとにではなく、データ パーティションごとに 1 回だけランできます。

• 行をまたいでコンテキストを維持

  UDTFはパーティション内で、行をまたいで状態を維持できます。このユニークな機能により、時系列パターンの検出や複雑な移動計算などの高度な分析が可能になります。

さらに、UDTFをUnity Catalog（UC）で定義すると、適切なアクセス権を持つすべてのユーザーがこれらの関数にアクセスし、検出し、実行できるようになります。つまり、一度書けばどこでも実行できるということです。

Databricks Runtime 17.3 LTS、Databricks SQL、および Serverless ノートブックとジョブにおいて、UC Python UDTF がパブリックプレビューで利用可能になったことをお知らせします。

このブログでは、UC Python UDTFの一般的なユースケースを例とともにいくつか紹介し、データパイプラインでそれらを使用する方法について説明します。

まずは、なぜUCでUDTFを使うのでしょうか？

Unity Catalog Python UDTF の利点

• 純粋なPythonで一度実装すれば、セッションやWorkspaceを問わずどこからでも呼び出せます

  標準的なPythonクラスにロジックを記述し、SQLウェアハウス（Databricks SQL Proおよびサーバレスを使用）、標準および専用のUCクラスター、Lakeflow宣言型パイプラインからPython UDTFを呼び出します。

• システムテーブルまたはCatalog Explorerを使用して検出
• 完全なUnity Catalogガバナンスにより、ユーザー間で共有できます。

• Python UDTF の権限の付与と取り消し
• LakeGuard の分離による安全な実行: Python UDTF は、一時的なディスクおよびネットワーク アクセス権を持つサンドボックスで実行され、他のワークロードからの干渉の可能性を防ぎます。

クイックスタート：簡素化されたIPアドレスのマッチング

まずは、IPアドレスをネットワークCIDRブロックのリストと照合するという、一般的なデータエンジニアリングの問題からスタートしましょう（たとえば、内部ネットワークからのトラフィックを識別する場合など）。標準 SQL には、CIDR ロジックとパッケージ用の組み込み関数がないため、このタスクは扱いにくいです。

UC Python UDTFは、そうした手間を解消します。これらを使用することで、Pythonの豊富なライブラリとアルゴリズムをSQLに直接持ち込むことができます。次のような関数を構築します。

1. IP logs のテーブルを入力として受け取ります。
2. データパーティションごとに1回だけ、既知のネットワークCIDRのリストを効率的に読み込みます。
3. IP アドレスごとに、Python の強力な ipaddress ライブラリを使用して、既知のいずれかのネットワークに属しているかどうかを確認します。
4. 一致するネットワーク情報が付加された元のLogsデータを返します。

まず、IPv4 と IPv6 の両方のアドレスを含むサンプルデータから始めましょう。

次に、UDTF を定義して登録します。Python クラスの構造に注目してください:

• t TABLE パラメーターは任意のスキーマの入力テーブルを受け入れます。UDTFは、提供された列を処理するために自動的に適応します。この柔軟性により、関数シグネチャを変更することなく、異なるテーブル間で同じ関数を使用できますが、行のスキーマを注意深く確認する必要もあります。
• __init__ メソッドは、大規模なネットワーク リストの読み込みなど、負荷の高い一度だけのセットアップに最適です。この処理は、入力テーブルのパーティションごとに1回実行されます。
• eval メソッドは、中核となるマッチングロジックを含み、各行を処理します。このメソッドは、そのパーティションに対応するIpMatcher UDTF クラスのインスタンスによって消費される入力パーティションの各行に対して、ちょうど 1 回実行されます。
• HANDLER 句は、UDTF ロジックを実装する Python クラスの名前を指定します。

我々のip_cidr_matcherがUnity Catalogに登録されたので、TABLE()構文を使用してSQLから直接呼び出すことができます。通常のテーブルへのクエリと同じくらい簡単です。

出力:

バッチ推論による画像キャプションの生成

この例では、DatabricksビジョンモデルサービングEndpointを使用して、バッチ画像キャプション用のUC Python UDTFをセットアップし、使用する方法を紹介します。まず、Wikimedia Commonsの公開画像URLを含むテーブルを作成します。

このテーブルには、自然の中の遊歩道、アリのマクロ写真、猫、銀河の4つのサンプル画像が含まれています。

そして、画像キャプションを生成するために UC Python UDTF を作成します。

1. まず、バッチサイズ、Databricks APIトークン、ビジョンモデルのendpoint、およびworkspace URLなどの構成を使用してUDTFを初期化します。
2. eval メソッドでは、画像 URL をバッファに収集します。バッファがバッチサイズに達すると、バッチ処理をtriggerします。これにより、画像ごとに個別に API を呼び出すのではなく、1 回の API 呼び出しで複数の画像をまとめて処理できます。
3. バッチ処理方式では、バッファリングされたすべての画像をdownloadして base64 としてエンコードし、Databricks VisionModel への単一のAPIリクエストで送信します。モデルはすべての画像を同時に処理し、バッチ全体のキャプションを返します。
4. terminate メソッドは、各パーティションの最後に 1 回だけ実行されます。`terminate`メソッドでは、バッファに残っている画像を処理し、収集したすべてのキャプションを結果として生成します。

<workspace-url> は、実際のDatabricksワークスペースのURL（例: https://your-workspace.cloud.databricks.com）に置き換えてください。

バッチ画像キャプションUDTFを使用するには、サンプル画像テーブルで呼び出すだけです。Databricks APIトークンの実際のsecret scopeとキー名を使用して、your_secret_scopeとapi_tokenを置き換えてください。

出力は次のとおりです：

カテゴリごとに画像のキャプションを生成することもできます:

出力は次のとおりです：

今後の課題

私たちは、以下のような、さらに強力でパフォーマンスの高い機能によってPython UDTFを拡張すべく、積極的に取り組んでいます。

• Unity Catalog のポリモーフィック UDTF は、入力引数に基づいて出力スキーマが動的に分析および解決される関数です。これらはセッション スコープの Python UDTF ですでにサポートされており、Unity Catalog の Python UDTF については現在対応中です。
• Python Arrow UDTF: ネイティブのApache Arrowレコードバッチ（iterator[Arrow.record_batch]）によるデータ処理を可能にする新しいPython UDTF APIです。大規模なデータセットで大幅なパフォーマンス向上を実現します。