Fabric Data Warehouse(プレビュー)でユーザーデータ機能を使う

適用対象:✅ Warehouse in Microsoft Fabric

Fabric Data Warehouseでは、T-SQL開発者はT-SQLクエリからユーザーデータ関数Fabric呼び出すことができます。 ウェアハウス内のT-SQLプロキシ関数は、公開されたユーザーデータ関数の項目で参照されたPython関数を呼び出します。

この統合を用いて、T-SQLを拡張し、再利用可能なPythonロジックを活用し、以下のようなシナリオに対応します。

• 空間、数値、またはデータサイエンスのワークロード向けに、専門的なPyPIパッケージを使用しています。
• Pythonでカスタムビジネスロジックを実装しています。
• 外部APIを呼び出す。
• 倉庫、レイクハウス、SQLデータベース、Cosmos DBなど、他のFabricアイテムやAzureサービスとのやり取り。

この記事では、Fabric Data Warehouseの公開ユーザーデータ関数を参照し、それをT-SQLクエリで利用する方法を紹介します。

前提条件

Fabric Data Warehouseのユーザーデータ機能を使用するには、以下の条件が必要です:

• ファブリック ウェアハウス。

• 呼び出したいPython関数を含む公開ユーザーデータ関数の項目です。

• ユーザーデータ関数項目への読み取りアクセスと、公開された関数を呼び出す権限。

• ターゲットウェアハウスで関数を作成する許可。

• ユーザーデータ関数プログラミングモデルに従うPython関数です。 構文要件、サポートされる入出力タイプ、デコレーターの使用については、Fabricユーザーデータ関数プログラミングモデルの概要をご覧ください。

• 新しいユーザーデータ機能項目を作成する必要がある場合は、サポート領域にFabric容量があり、その容量に割り当てられたFabricワークスペースも必要です。 詳細については、Fabricの「ユーザーデータ機能項目を作成」および「サービスの詳細」および「Fabricユーザーデータ関数の制限」を参照してください。

ユーザーデータ関数の作成と公開

ウェアハウスからユーザーデータ関数を呼び出す前に、まずその関数を作成してユーザーデータ関数の項目で公開してください。 完全な手順とサンプルPython関数については、「Fabricでユーザーデータ関数項目を作成」をご覧ください。

作成可能なPythonユーザーデータ関数は以下の例に示されています。

import fabric.functions as fn

udf = fn.UserDataFunctions()

@udf.function()
def hello_fabric(name: str) -> str:
    # Your function logic here
    return name

ユーザーデータ関数のアイテム名を使って、T-SQLプロキシ関数をPython関数にバインドします。

T-SQLプロキシ関数を作成する

倉庫でT-SQL関数を作成し、AS EXTERNAL FUNCTION節を使って公開されたPython関数にバインドします。

CREATE OR ALTER FUNCTION dbo.hello_fabric
AS EXTERNAL FUNCTION FunctionSetName.hello_fabric;

FunctionSetNameをユーザーデータ関数の項目名に置き換え、hello_fabricを公開した Python 関数名に置き換えます。

デフォルトでは、CREATE FUNCTION文はリモートFabricユーザーデータ関数(UDF)の定義からパラメータリストとリターン型を導出します。 推論された返り値タイプは関数定義で明示的に指定することで上書きできます。 リターン型を提供することは、リモート関数のメタデータから推測されるよりも正確なリターン型を公開したい場合に役立ちます。

CREATE OR ALTER FUNCTION dbo.hello_fabric
RETURNS VARCHAR(100)
AS EXTERNAL FUNCTION FunctionSetName.hello_fabric;

例えば、Python str戻り値は通常VARCHAR(MAX)として推定されますが、関数が常に最大長さが既知の値を返す場合は、より正確なメタデータや型情報を提供するために、リターンタイプをVARCHAR(100)や他の適切な長さとして明示的に定義できます。

T-SQLクエリでプロキシ関数を使う

プロキシ関数を作成したら、他のユーザー定義関数と同様にT-SQLから呼び出します。

以下のようなクエリ式でプロキシ関数を使用してください:

• SELECT リスト。
• WHERE 条項。
• GROUP BY 条項。
• UPDATEやINSERT文の表現。

以下のパターンはクエリからプロキシ関数を呼び出す方法を示しています:

SELECT dbo.<proxy_function_name>(<arguments>) AS function_result;

高度なPythonパターン、対応型、接続、コンテキストオブジェクトについては、Fabricユーザーデータ関数プログラミングモデルの概要をご覧ください。

バッチモード実行の設定

Fabric Data Warehouse外部ユーザーデータ関数をバッチモードで実行できます。

バッチモードで実行すると、複数の関数呼び出しを1回のリクエストでリモートのFabricユーザーデータ機能(UDF)サービスに送信し、ネットワークのオーバーヘッドとリクエスト遅延を削減します。 バッチは最大900回の関数呼び出しを含めます。

バッチモードは通常、同じ関数を1つ以上のテーブルカラムの値に適用する場合に用いられ、クエリプロセッサが各行ごとに個別の要求を出すのではなく、単一のリモート呼び出しで多数の行を評価できるようにします。 この方法は、特に大規模なデータセットや高遅延の外部関数呼び出しにおいてクエリ性能を大幅に向上させることができます。

例えば、クエリがカラム内のすべての値にユーザーデータ関数を適用した場合、Fabric Data Warehouse入力値をバッチにまとめてリモート処理のためにまとめて提出できます。 リモートサービスは対応する結果のバッチを返し、それらはクエリ実行パイプラインに統合されます。

ユーザーデータ機能のバッチモードを有効にするには:

1. ユーザーデータ機能項目を開きます。
2. ライブラリ管理>設定に行ってください。
3. プレビュー機能をオンにしてください。

@udf.function()デコレーターでmax_batch_size=900を設定し、倉庫が各呼び出しで最大900行の入力行を送信できるようにします:

import fabric.functions as fn

udf = fn.UserDataFunctions()

@udf.function(max_batch_size=900)
def normalizeText(value: str) -> str:
    return value.strip().lower()

バッチ対応ユーザーデータ関数(UDF)を参照するT-SQLプロキシ関数を作成すると、Fabric Data Warehouseリモート関数で定義された最大バッチサイズを含むバッチ構成を自動的に検出します。 クエリパターンがバッチ処理をサポートしている場合、クエリプロセッサは複数の関数呼び出しをバッチにまとめ、各行ごとに個別のリクエストを発行するのではなく、リモートサービスに送信して実行します。

ウェアハウス内のユーザーデータ関数をリストアップする

AS EXTERNAL FUNCTIONで作成したユーザー データ関数は、オブジェクト タイプ sys.objects として XF に表示されます。

ウェアハウス内のユーザーデータ機能を一覧表示するには、type = 'XF'でsys.objectsをフィルタリングします。

SELECT SCHEMA_NAME(schema_id) AS schema_name, name, object_id, type, type_desc
FROM sys.objects
WHERE type = 'XF';

通常のスカラー関数(FN)とユーザーデータ関数(XF)をパラメータおよびリターン型署名とともに一覧表示するには、 sys.parameters と sys.typesを使用します。

WITH metadata AS
(
    SELECT o.schema_id, o.object_id, o.name, p.parameter_id, p.name AS parameter_name, t.name AS type_name
    FROM sys.objects AS o
    INNER JOIN sys.parameters AS p
        ON p.object_id = o.object_id
    INNER JOIN sys.types AS t
        ON t.user_type_id = p.user_type_id
    WHERE o.type IN ('FN', 'XF')
),
signature AS
(
    SELECT SCHEMA_NAME(schema_id) AS schema_name,
           name,
           ANY_VALUE(
               CASE WHEN parameter_id = 0 THEN type_name END
           ) AS return_type,
           CONCAT(
               SCHEMA_NAME(schema_id),
               '.',
               name,
               '(',
               STRING_AGG(
                   CASE
                       WHEN parameter_id > 0
                       THEN CONCAT(parameter_name, ' ', type_name)
                   END,
                   ', '
               ) WITHIN GROUP (ORDER BY parameter_id),
               ') -> ',
               MAX(CASE WHEN parameter_id = 0 THEN type_name END)
           ) AS signature
    FROM metadata
    GROUP BY schema_id, object_id, name
)
SELECT *
FROM signature;

注釈

• T-SQLプロキシ関数は、公開されたユーザーデータ関数の項目と関数名に依存します。 Python関数の名前を変更したり削除したりする場合は、それに応じてウェアハウスプロキシ関数を更新してください。
• ユーザーデータ関数プログラミングモデルは、サポートされる入力型と出力型を定義します。 Pythonパラメータと返す注釈が、T-SQLクエリが通す値をサポートしているか確認してください。
• ユーザーデータ関数には、リクエストペイロードサイズ、実行タイムアウト、応答サイズ、ライブラリサイズ、ログ保持のサービス制限があります。 現在の制限については、サービスの詳細およびFabricユーザーデータ機能の制限を参照してください。
• パラメータ名は camelCase を使用し、型注釈を含める必要があります。 @udf.function()で装飾された関数はリターン型も指定しなければなりません。 完全な構文ルールについては、Fabricユーザーデータ関数プログラミングモデルの概要をご覧ください。
• カスタムPythonコードを必要としない組み込みのテキスト変換関数については、「Use AI functions(プレビュー)」をご覧ください。
• ユーザーデータ関数には、リクエストペイロードサイズ、実行タイムアウト、応答サイズ、ライブラリサイズ、ログ保持のサービス制限があります。 現在の制限については、サービスの詳細およびFabricユーザーデータ機能の制限を参照してください。
• パラメータ名は camelCase を使用し、型注釈を含める必要があります。 @udf.function()で装飾された関数はリターン型も指定しなければなりません。 完全な構文ルールについては、Fabricユーザーデータ関数プログラミングモデルの概要をご覧ください。
• カスタムPythonコードを必要としない組み込みのテキスト変換関数については、「Use AI functions(プレビュー)」をご覧ください。

Examples

以下のシナリオは、ユーザーデータ機能でFabric Data Warehouseを拡張する一般的な方法を示しています。

A. IPアドレス解析でT-SQLを拡張する

ウェアハウスクエリが組み込みのT-SQL関数で利用できない機能が必要な場合は、Pythonのユーザーデータ関数を使いましょう。 例えば、IPアドレスパッケージを使ってIPアドレスのサブネットを返すことができます。 この機能は、クエリが多数の行からIPアドレスを処理する際にバッチ単位で適用できます。 設定手順については「 バッチモードの設定」を参照してください。

ユーザーデータ関数項目でPython関数を定義し、公開します:

from netaddr import IPNetwork
import fabric.functions as fn

udf = fn.UserDataFunctions()

@udf.function(max_batch_size=900)
def ipSubnet(ipAddress: str, prefixLength: int = 24) -> str:
    return str(IPNetwork(f"{ipAddress}/{prefixLength}").cidr)

公開されたPython関数を参照するT-SQLプロキシ関数をウェアハウス内に作成します。

CREATE OR ALTER FUNCTION dbo.ip_subnet
AS EXTERNAL FUNCTION inet.ipSubnet;

ウェアハウスクエリからプロキシ関数を呼び出します:

SELECT dbo.ip_subnet('192.168.1.25', 24) AS subnet;

予想される結果:192.168.1.0/24

B. 外部公開APIを呼び出してください

ウェアハウスクエリがデータ強化ワークフローの一環として外部APIを呼び出す必要がある場合、ユーザーデータ関数を活用してください。

Fabric、Azure SQL Database、Azure SQL Managed InstanceのSQLデータベースでは、sys.sp_invoke_external_rest_endpointHTTPS RESTエンドポイントを呼び出します。 Fabric Data Warehouse では、ユーザー データ関数を使用して、Python で HTTPS 呼び出しをラップし、それをプロキシ関数を介して T-SQL に公開することで、代替パターンを提供できます。

ユーザーデータ関数項目でPython関数を定義し、公開します。 この例では、パラメータ名が次のように使われています sys.sp_invoke_external_rest_endpoint:

import json
import requests
import fabric.functions as fn

udf = fn.UserDataFunctions()

@udf.function()
def invokeExternalRestEndpoint(
    url: str,
    method: str = "GET",
    payload: str | None = None,
    headers: str | None = None,
    timeout: int = 30
) -> str:
    requestHeaders = json.loads(headers) if headers else None
    requestPayload = payload if payload else None

    response = requests.request(
        method=method,
        url=url,
        data=requestPayload,
        headers=requestHeaders,
        timeout=timeout
    )
    response.raise_for_status()
    return response.text

ウェアハウス内でT-SQLプロキシ関数を作成します:

CREATE OR ALTER FUNCTION dbo.invoke_external_rest_endpoint
AS EXTERNAL FUNCTION api.invokeExternalRestEndpoint;

ウェアハウスクエリからプロキシ関数を呼び出します:

SELECT dbo.invoke_external_rest_endpoint(
    'https://ipapi.co/8.8.8.8/country_name/',
    'GET',
    NULL,
    NULL,
    30
) AS response;

C. 設定可能な保持期間を使いましょう

ウェアハウスクエリがFabric変数ライブラリから中央管理される設定値を必要とする場合は、ユーザーデータ関数を活用してください。 例えば、変数ライブラリに保持期間を保存し、プロキシ関数で取得し、それをDATEADD T-SQL関数の数値引数として使います。

関数を公開する前に、ユーザーデータ関数の項目から変数ライブラリへの接続を追加し、接続エイリアスを記録します。 この例の変数ライブラリには90のようなRETENTION_DAYS値が含まれています。

ユーザーデータ関数項目でPython関数を定義し、公開します:

import fabric.functions as fn

udf = fn.UserDataFunctions()

@udf.connection(argName="varLib", alias="WarehouseConfig")
@udf.function()
def getRetentionDays(varLib: fn.FabricVariablesClient) -> int:
    variables = varLib.getVariables()
    return int(variables["RETENTION_DAYS"])

WarehouseConfigを変数ライブラリ接続のエイリアスに置き換えます。

ウェアハウス内でT-SQLプロキシ関数を作成します:

CREATE OR ALTER FUNCTION dbo.get_retention_days
AS EXTERNAL FUNCTION config.getRetentionDays;

永続的なT-SQL接続では、設定値を一度取得し、以下の sp_set_session_contextで保存します。

DECLARE @retention_days int = dbo.get_retention_days();

EXECUTE sys.sp_set_session_context
    @key = N'retention_days',
    @value = @retention_days,
    @read_only = 1;

その価値はセッションの存続期間中、引き続き利用可能です。 SESSION_CONTEXTで取得し、sql_variantからintに変換してから、DATEADDとGETDATEで使ってください。 例えば、保持カットオフを計算します:

SELECT DATEADD(
    day,
    -CONVERT(int, SESSION_CONTEXT(N'retention_days')),
    GETDATE()
) AS retention_cutoff;

同じセッション値を別の文で再利用してテーブルをフィルタリングします:

SELECT *
FROM dbo.events
WHERE event_timestamp >= DATEADD(
    day,
    -CONVERT(int, SESSION_CONTEXT(N'retention_days')),
    GETDATE()
);

ユーザーデータ機能のトラブルシューティング

Query Insightsは、Fabric関数のトラブルシューティングに必要な実行およびパフォーマンス情報を提供します。 関数を呼び出したクエリを特定し、その関数がバッチ実行か行実行かを判断し、外部サービス遅延、再試行、失敗行、ペイロードサイズを調査できます。 クエリレベルの概要から各関数の詳細な統計に移行するには、以下の2つのビューを使用します。

• queryinsights.exec_requests_historyを使って、FabricやAI関数を呼び出したクエリを特定しましょう。 このビューにはクエリテキスト、ステータス、提出時間、経過総時間が含まれます。
• クエリで呼び出される各関数の詳細な統計を得るには queryinsights.external_api_call_stats を使います。 ビューには関数型、実行モード、呼び出しおよび再試行数、外部サービス待ち時間、ペイロードサイズ、行結果が含まれます。

ビューは同じクエリ実行を識別するために distributed_statement_id を使用します。

Fabric関数を使用したクエリを探す

queryinsights.exec_requests_historyを使って、FabricやAI関数を呼び出した最近の発言を探してください:

SELECT TOP 100
       h.distributed_statement_id,
       h.submit_time,
       h.status,
       h.total_elapsed_time_ms,
       h.command
FROM queryinsights.exec_requests_history AS h
WHERE h.is_using_external_api = 1
ORDER BY h.submit_time DESC;

調査したい陳述書の distributed_statement_id をコピーしてください。 次のクエリでは詳細な統計情報をFabric関数にフィルタリングします。

各関数の統計量を文で確認できます

<distributed_statement_id>を前のクエリで返された識別子に置き換えます。 以下のクエリは、文によって呼び出された各異なるFabric関数に対して1行を返します。

DECLARE @distributed_statement_id uniqueidentifier =
    '<distributed_statement_id>';

SELECT function_name,
       execution_mode,
       call_count,
       batch_call_count,
       row_call_count,
       call_retry_count,
       external_service_wait_time_ms,
       external_service_wait_time_ms
           / call_count AS average_wait_time_ms_per_call,
       rows_total,
       rows_succeeded,
       rows_failed,
       data_sent_bytes,
       data_received_bytes
FROM queryinsights.external_api_call_stats
WHERE distributed_statement_id = @distributed_statement_id
  AND function_type = 'FABRIC_FUNCTION'
ORDER BY external_service_wait_time_ms DESC;

結果は以下の例に似ています。

トラブルシューティングのヒント

• row_call_countが高い、またはrowのexecution_mode値は、関数がバッチ実行を使っていないことを示しています。 バッチ処理を期待する場合は、バッチモードのプレビューが有効で、関数が max_batch_size 引数を使用しているかを確認してください。
• external_service_wait_time_msとaverage_wait_time_ms_per_callを使って、外部からの反応が遅いと見極めましょう。 これらの値が高いのに data_sent_bytes や data_received_bytes が低い場合、外部サービス応答時間が主な遅延である可能性が高いです。
• 大きな data_sent_bytes や data_received_bytes 値も実行時間を増加させることがあります。 関数がシナリオに必要なデータ量を超えて転送した場合、入力または出力ペイロードを削減します。

関連するコンテンツ

• Fabricでユーザーデータ関数項目を作成する
• Fabricユーザーデータ機能プログラミングモデルの概要
• Fabric ユーザー データ関数のサービスの詳細と制限事項