外部関数とUDFs／UDTFs

使ったことがない機能のドキュメントを読んで詳しくなるシリーズ。
今回は外部関数(External Function)。

【目次】

∨ユーザ定義関数
∨スカラー関数(UDFs)
∨表関数(UDTFs)
∨外部関数
∨リモートサービスの入出力定義
∨リモートサービスの応答定義
∨まとめ

ユーザ定義関数

まずユーザ定義関数(User Defined Function)について。
UDFは、組み込み関数と同様にSQLから呼び出すことができる。組み込み関数にない機能を自作できる。
コードを共通化して2個以上の場所で発生するコードクローンを除去できる。

ユーザ定義関数は単一の値を返すパターンと、表を返すパターンの2パターンを定義できる。
単一の値を返すパターンはスカラー関数(UDFs)。
表、つまり0個,1個,2個以上の値を返すパターンは表関数(UDTFs)。
サポートされている言語はJava,JavaScript,Python,SQL。

スカラー関数(UDFs)

スカラー関数は入力行に対して1つの出力行を返す。
公式のサンプルを読むのが早い。

-- UDFの定義
create or replace function addone(i int)
returns int
language python
runtime_version = '3.8'
handler = 'addone_py'
as
$$
def addone_py(i):
  return i+1
$$;

-- UDFの実行
select addone(1);
2

-- UDFの定義

create or replace function addone(i int)

returns int

language python

runtime_version = '3.8'

handler = 'addone_py'

def addone_py(i):

return i+1

$$;

-- UDFの実行

select addone(1);

こんな調子で、create function の中に書き下した処理をインラインハンドラと言う。
処理を修正する際は、alter function を実行して対応する。
確かにお手軽だが、行数が増えてきたり、処理を共通化したりしたい場合は微妙。

あらかじめコードをコンパイルしておき再利用することもできる(ステージングハンドラ)。
ステージングハンドラであれば、より多くの行数に対応できるし、処理の共通化も用意。
ステージングハンドラの実行例は以下の通り。

CREATE FUNCTION my_udf(i NUMBER)
  RETURNS NUMBER
  LANGUAGE JAVA
  IMPORTS = ('@mystage/handlers/my_handler.jar')
  HANDLER = 'MyClass.myFunction'

CREATE FUNCTION my_udf(i NUMBER)

RETURNS NUMBER

LANGUAGE JAVA

IMPORTS = ('@mystage/handlers/my_handler.jar')

HANDLER = 'MyClass.myFunction'

インラインハンドラは全ての言語で対応できるが、
ステージングハンドラは、JavaまたはPythonが必要。

表関数(UDTFs)

表関数は入力行に対して表、つまり0行,1行,2行,…行を返す。
実行例は以下の通り。なるほど。
戻り値が表(table)になるため、SQLだと処理と戻り値のギャップが少ないが、
SQLでは手続き的な処理が書けないため自由度は低い。

-- サンプルテーブル・データの作成
create or replace table orders (
    product_id varchar, 
    quantity_sold numeric(11, 2)
    );

insert into orders (product_id, quantity_sold) values 
    ('compostable bags', 2000),
    ('re-usable cups',  1000);

-- UDTFの定義
create or replace function orders_for_product(PROD_ID varchar)
    returns table (Product_ID varchar, Quantity_Sold numeric(11, 2))
    as
    $$
        select product_ID, quantity_sold 
            from orders 
            where product_ID = PROD_ID
    $$
    ;

-- UDTFの実行
select product_id, quantity_sold
    from table(orders_for_product('compostable bags'))
    order by product_id;
+------------------+---------------+
| PRODUCT_ID       | QUANTITY_SOLD |
|------------------+---------------|
| compostable bags |       2000.00 |
+------------------+---------------+

-- サンプルテーブル・データの作成

create or replace table orders (

product_id varchar,

quantity_sold numeric(11, 2)

);

insert into orders (product_id, quantity_sold) values

('compostable bags', 2000),

('re-usable cups', 1000);

-- UDTFの定義

create or replace function orders_for_product(PROD_ID varchar)

returns table (Product_ID varchar, Quantity_Sold numeric(11, 2))

select product_ID, quantity_sold

from orders

where product_ID = PROD_ID

;

-- UDTFの実行

select product_id, quantity_sold

from table(orders_for_product('compostable bags'))

order by product_id;

+------------------+---------------+

| PRODUCT_ID | QUANTITY_SOLD |

|------------------+---------------|

| compostable bags | 2000.00 |

+------------------+---------------+

JavaScriptで書くと以下のようになる。手続き的に処理を書いている。
これであればある程度の自由度を獲得できる。

-- UDTFの定義
-- IP アドレスの「範囲」を IP アドレスの完全なリストに変換
CREATE OR REPLACE FUNCTION range_to_values(PREFIX VARCHAR, RANGE_START FLOAT, RANGE_END FLOAT)
    RETURNS TABLE (IP_ADDRESS VARCHAR)
    LANGUAGE JAVASCRIPT
    AS $$
      {
        processRow: function f(row, rowWriter, context)  {
          var suffix = row.RANGE_START;
          while (suffix <= row.RANGE_END)  {
            rowWriter.writeRow( {IP_ADDRESS: row.PREFIX + "." + suffix} );
            suffix = suffix + 1;
            }
          }
      }
      $$;

-- UDTFの実行
SELECT * FROM TABLE(range_to_values('192.168.1', 42::FLOAT, 45::FLOAT));
+--------------+
| IP_ADDRESS   |
+==============+
| 192.168.1.42 |
+--------------+
| 192.168.1.43 |
+--------------+
| 192.168.1.44 |
+--------------+
| 192.168.1.45 |
+--------------+

-- UDTFの定義

-- IP アドレスの「範囲」を IP アドレスの完全なリストに変換

CREATE OR REPLACE FUNCTION range_to_values(PREFIX VARCHAR, RANGE_START FLOAT, RANGE_END FLOAT)

RETURNS TABLE (IP_ADDRESS VARCHAR)

LANGUAGE JAVASCRIPT

AS $$

{

processRow: function f(row, rowWriter, context) {

var suffix = row.RANGE_START;

while (suffix <= row.RANGE_END) {

rowWriter.writeRow( {IP_ADDRESS: row.PREFIX + "." + suffix} );

suffix = suffix + 1;

}

$$;

-- UDTFの実行

SELECT * FROM TABLE(range_to_values('192.168.1', 42::FLOAT, 45::FLOAT));

+--------------+

| IP_ADDRESS |

+==============+

| 192.168.1.42 |

+--------------+

| 192.168.1.43 |

+--------------+

| 192.168.1.44 |

+--------------+

| 192.168.1.45 |

+--------------+

外部関数

ユーザ定義関数のロジック(ハンドラ)を外部のコンピューティングリソースに配置することができる。
概念図は以下の通り。Snowflake側から見て、外部関数、API統合、プロキシサービス、リモートサービスから成る。

外部関数(External Function)はスカラー関数(UDFs)だが、自前でコードを書かない。
その実体は、プロキシサービスを介してリモートサービスのコードを実行するためのオブジェクトで、
CREATE EXTERNAL FUNCTION により定義する。

リモートサービスは、外部のコンピューティングリソース。
AWS Lambda、Azure Functionsが分かりやすいが、
HTTPSでJSONを受けてJSONを返せれば、一般的なWebサーバでも良い様子。

プロキシサービスは、要はリモートサービスを抽象化するレイヤの機能を持つ。
AWS API Gateway、Azure API Managementサービスと言うと分かりやすい。
外部関数が直接リモートサービスを叩かない理由は、プロキシサービスに認証機能を持たせるため。

API統合は、外部関数をSnowflakeの世界に持ち込むための抽象化レイヤ。
同様の機能として、S3,Blobへのアクセス手段を抽象化するストレージ統合がある。

使い場合はUDFと同様。

select my_external_function_2(column_1, column_2)
    from table_1;

select col1
    from table_1
    where my_external_function_3(col2) < 0;

create view view1 (col1) as
    select my_external_function_5(col1)
        from table9;

select upper(zipcode_to_city_external_function(zipcode))
  from address_table;

select my_external_function_2(column_1, column_2)

from table_1;

select col1

from table_1

where my_external_function_3(col2) < 0;

create view view1 (col1) as

select my_external_function_5(col1)

from table9;

select upper(zipcode_to_city_external_function(zipcode))

from address_table;

リモートサービスの入出力定義

前述のように、リモートサービスはHTTPSでJSONを受けてJSONを返す必要がある。
そのインターフェースは公式に記述がある。
リモートサービスの入力および出力データ形式

基本的に、POSTを受けて同期的に応答するようにする。
ただし、非同期処理もサポートしており、その際はGETでポーリングリクエストを受ける必要がある。

関数 ext_func(n integer) returns varchar のケースにおいて、リクエストヘッダは以下。
いわゆるRPC(Remote Procedure Call)を定義する何かに準じている。

ヘッダ文字列	値
sf-external-function-format	json固定
sf-external-function-format-version	1.0固定
sf-external-function-query-batch-id	リクエスト固有の識別子。この値を使って処理の成否を確認する。
sf-external-function-name	関数名 ext_func
sf-external-function-name-base64	関数名(base64) base64(ext_func)
sf-external-function-signature	関数名署名 (N NUMBER)
sf-external-function-signature-base64	関数名署名(base64) base64(N NUMBER)
sf-external-function-return-type	戻り値型 VARCHAR(16777216)
sf-external-function-return-type-base64	戻り値型(base64) base64(VARCHAR(16777216))

リクエスト本文はJSON。
重要なことは、外部関数はスカラー関数であり、1個の入力に対して1個の応答を返す必要があるが、
一度に複数の入力を受けることができ、それぞれに対して応答を返す必要がある、ということ。
複数の入力のセットを公式ではバッチ、と読んでいる様子。

署名 f(integer, varchar, timestamp) を持つ外部関数にシリアル化したJSONを送る。例は以下。
先頭の数字が1個多い気がするが、これは行番号を表す。

{
    "data": [
                [0, 10, "Alex", "2014-01-01 16:00:00"],
                [1, 20, "Steve", "2015-01-01 16:00:00"],
                [2, 30, "Alice", "2016-01-01 16:00:00"],
                [3, 40, "Adrian", "2017-01-01 16:00:00"]
            ]
}

{

"data": [

[0, 10, "Alex", "2014-01-01 16:00:00"],

[1, 20, "Steve", "2015-01-01 16:00:00"],

[2, 30, "Alice", "2016-01-01 16:00:00"],

[3, 40, "Adrian", "2017-01-01 16:00:00"]

]

}

リモートサービスの応答定義

レスポンス本文はJSON。リクエスト本文と同様にコレクションの先頭は行番号を設定する必要がある。
公式には、都市名を引数として、その緯度軽度を返す関数の例が書いてある。
このように、入力と対応する応答値を、入力と同じ順番にコレクションに詰めてシリアライズして返す。

この外部関数は、VARIANT型を戻している。このように、スカラー関数ではあるが、
1回の処理が(半)構造を持てるので、事実上複数の入力値に対して複数の応答値を返せる。

-- 呼び出し例
select val:city, val:latitude, val:longitude
    from (select ext_func_city_lat_long(city_name) as val from table_of_city_names);

-- レスポンス
{
    "data":
        [
            [ 0, { "City" : "Warsaw",  "latitude" : 52.23, "longitude" :  21.01 } ],
            [ 1, { "City" : "Toronto", "latitude" : 43.65, "longitude" : -79.38 } ]
        ]
}

-- 呼び出し例

select val:city, val:latitude, val:longitude

from (select ext_func_city_lat_long(city_name) as val from table_of_city_names);

-- レスポンス

{

"data":

[

[ 0, { "City" : "Warsaw", "latitude" : 52.23, "longitude" : 21.01 } ],

[ 1, { "City" : "Toronto", "latitude" : 43.65, "longitude" : -79.38 } ]

]

}

StatusCode=200で、同期応答完了。
202を返すと、非同期応答となり処理を継続できる様子だが今回は省略。
その他はエラー応答として扱われる。

まとめ

外部関数に関するドキュメントと、その理解に必要なUDFs、UDTSsのドキュメントを読んでみた。
実体はHTTPS上のWebAPIであり、事実上無限の自由度を獲得できると思う。
(これを持ち出す頃には、本末転倒な何かに陥っていないか注意は必要だと思う)

外部関数とUDFs／UDTFs

ユーザ定義関数

スカラー関数(UDFs)

表関数(UDTFs)

外部関数

リモートサービスの入出力定義

リモートサービスの応答定義

まとめ

AirflowでEnd-To-End Pipeline Testsを行うためにAirflow APIを調べてみた話

CustomOperatorのUnitTestを理解するためGCSToBigQueryOperatorのUnitTestを読んでみた話

GoogleによるAirflow DAG実装のベスプラ集を読んでみた – その1

Snowpark Container Services上でWebアプリ(FastAPI/React/TypeScript)を動かしてみた

Azure Queue StorageとAzure Service Busを比較してみた

Azure Functionsの機能まとめ（座学版）

External Network Accessを使ってSnowflakeとFitbitAPIを繋いでみた話

Deep dive into the internals of Snowflake Virtual Warehousesを読んでみた

GCSとのストレージ統合を設定した話

デプロイメントについて調べてみた話（端折り気味）

AirflowでEnd-To-End P

CustomOperatorのUnitT