Pythonを利用したユーザー定義拡張関数

1 はじめに
2 ユーザー定義拡張関数(UDx)とは
3 ユーザー定義拡張関数の種類
4 参考情報
5 検証バージョンについて
6 更新履歴

はじめに

本記事では、Verticaの拡張機能であるユーザー定義拡張関数(UDx)を紹介します。

ユーザー定義拡張関数(UDx)とは

ユーザー定義拡張関数とは、外部の言語を用いて書かれた関数です。通常のSQLでの記述が難しい分析処理や、機械学習においてVerticaには実装されていないアルゴリズムを用いたい場合に有効な手法です。
現在Verticaでは、Python, R, C++, Java の4つの言語を利用することができます。今回は、プログラムコード例をPythonで記述しています。

ユーザー定義拡張関数の種類

Pythonで使用できるユーザー定義拡張関数には、大きくわけてスカラー関数とトランスフォーム関数の2種類があります。(この点はRも同様)

スカラー関数

スカラー関数は、読み込んだデータ1行に対して単一の値（列）を返す関数です。この範囲に収まらない場合は、後述するトランスフォーム関数を利用する必要があります。

トランスフォーム関数

トランスフォーム関数は、読み込んだデータ1行に対して複数の行および複数の列を返すことができます。出力が多岐にわたる場合やテーブルの外形を変化させる処理を実施する場合は、こちらの関数を利用します。

スカラー関数の利用例

例1：2倍の値を返す

入力値と出力値が１列対１列となる場合、スカラー関数で実装可能です。

例2：2列の和の値を返す

入力値と出力値が複数列対１列となる場合でも、スカラー関数で実装可能です。

トランスフォーム関数の利用例

例3：2列の和と差の値を2列で返す

入力値と出力値が複数列対複数列となる場合には、トランスフォーム関数を利用する必要があります。

例4：2列の和と差の値を2行で返す

この例は入力値と出力値が複数列対１列となっています。しかし１行の入力値に対して複数行の出力値を返しているため、このようなケースではトランスフォーム関数を利用する必要があります。

スカラー関数のプログラムコード例

2列の和の値を返すスカラー関数をPythonで実装します。
Pythonコードの例(ファイル名：col_sum_func_sc.py)

#vetica_sdkをインポートする
import vertica_sdk

#メインのクラスcol_sum_func_sc, 処理を記述する
class col_sum_func_sc(vertica_sdk.ScalarFunction):

    def processBlock(self, server_interface, in_data, out_data):
        col0 = []
        col1 = []
        while True:
            col0.append(in_data.getInt(0))
            col1.append(in_data.getInt(1))

            if not in_data.next():
                break

        for k in range(len(col0)):
            out_data.setInt(col0[k] + col1[k])
            out_data.next()

#ファンクションcol_sum_func_scの定義や入出力の型を決定するファンクションcol_sum_func_sc_Factory
class col_sum_func_sc_Factory(vertica_sdk.ScalarFunctionFactory):

    def createScalarFunction(self, srv):
        return col_sum_func_sc()

    def getPrototype(self, srv_interface, arg_types, return_type):
        arg_types.addInt()
        arg_types.addInt()
        return_type.addInt()

    def getReturnType(self, srv_interface, arg_types, return_type):
        return_type.addInt()

#vetica_sdkをインポートする

import vertica_sdk

#メインのクラスcol_sum_func_sc, 処理を記述する

class col_sum_func_sc(vertica_sdk.ScalarFunction):

def processBlock(self, server_interface, in_data, out_data):

col0 = []

col1 = []

while True:

col0.append(in_data.getInt(0))

col1.append(in_data.getInt(1))

if not in_data.next():

break

for k in range(len(col0)):

out_data.setInt(col0[k] + col1[k])

out_data.next()

#ファンクションcol_sum_func_scの定義や入出力の型を決定するファンクションcol_sum_func_sc_Factory

class col_sum_func_sc_Factory(vertica_sdk.ScalarFunctionFactory):

def createScalarFunction(self, srv):

return col_sum_func_sc()

def getPrototype(self, srv_interface, arg_types, return_type):

arg_types.addInt()

return_type.addInt()

def getReturnType(self, srv_interface, arg_types, return_type):

return_type.addInt()

テーブルTable1を確認します。

 dbadmin=> SELECT * FROM schema.Table1;
 X1 | X2
----+----
  2 |  3
  5 |  7
 11 | 10
(3 rows)

dbadmin=> SELECT * FROM schema.Table1;

X1 | X2

----+----

2 | 3

5 | 7

11 | 10

(3 rows)

Pythonプログラム col_sum_func_sc.py をVerticaのライブラリPylibSCとして登録します。

dbadmin=> CREATE OR REPLACE LIBRARY schema.PylibSC AS '/home/dbadmin/col_sum_func_sc.py' LANGUAGE 'Python';

1	dbadmin=> CREATE OR REPLACE LIBRARY schema.PylibSC AS '/home/dbadmin/col_sum_func_sc.py' LANGUAGE 'Python';

ライブラリPylibに、ファンクションcol_sum_func_scを定義します。
この際、入出力の型を定義するファンクションcol_sum_func_sc_Factoryも同時に定義します。

dbadmin=> CREATE OR REPLACE FUNCTION col_sum_func_sc AS NAME 'col_sum_func_sc_Factory' LIBRARY schema.PylibSC;

1	dbadmin=> CREATE OR REPLACE FUNCTION col_sum_func_sc AS NAME 'col_sum_func_sc_Factory' LIBRARY schema.PylibSC;

ファンクションcol_sum_func_scを呼び出すSQLを実行します。

dbadmin=> SELECT X1
dbadmin->       ,X2
dbadmin->       ,col_sum_func_sc(X1,X2) AS Y
dbadmin-> FROM schema.Table1;
 X1 | X2 | Y
----+----+----
  2 |  3 |  5
  5 |  7 | 12
 11 | 10 | 21
(3 rows)

dbadmin=> SELECT X1

dbadmin-> ,X2

dbadmin-> ,col_sum_func_sc(X1,X2) AS Y

dbadmin-> FROM schema.Table1;

X1 | X2 | Y

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

2 | 3 | 5

5 | 7 | 12

11 | 10 | 21

(3 rows)

トランスフォーム関数のプログラムコード例

2列の和と差（1列目-2列目）の値を2列で返すトランスフォーム関数をPythonで実装します。
Pythonコードの例（ファイル名：col_sum_func_tf.py）

#vertica_sdkをインポートする
import vertica_sdk

#メインのクラスcol_sum_func_tf, 処理を記述する
class col_sum_func_tf(vertica_sdk.TransformFunction):

    def processPartition(self, server_interface, in_data, out_data):
        col0 = []
        col1 = []
        while True:
            col0.append(in_data.getInt(0))
            col1.append(in_data.getInt(1))

            if not in_data.next():
                break

        for k in range(len(col0)):
            out_data.setInt(0,col0[k]+col1[k])
            out_data.setInt(1,col0[k]-col1[k])
            out_data.next()

#ファンクションcol_sum_func_tfの定義や入出力の型を決定するファンクションcol_sum_func_tf_Factory
class col_sum_func_tf_Factory(vertica_sdk.TransformFunctionFactory):

    def getPrototype(self, server_interface, arg_types, return_types):
        arg_types.addInt()
        arg_types.addInt()
        return_types.addInt()
        return_types.addInt()

    def getReturnType(self, server_interface, arg_types, return_type):
        return_type.addColumn(arg_types.getColumnType(0),"sum")
        return_type.addColumn(arg_types.getColumnType(1),"dif")

    def createTransformFunction(cls, server_interface):
        return col_sum_func_tf()

#vertica_sdkをインポートする

import vertica_sdk

#メインのクラスcol_sum_func_tf, 処理を記述する

class col_sum_func_tf(vertica_sdk.TransformFunction):

def processPartition(self, server_interface, in_data, out_data):

col0 = []

col1 = []

while True:

col0.append(in_data.getInt(0))

col1.append(in_data.getInt(1))

if not in_data.next():

break

for k in range(len(col0)):

out_data.setInt(0,col0[k]+col1[k])

out_data.setInt(1,col0[k]-col1[k])

out_data.next()

#ファンクションcol_sum_func_tfの定義や入出力の型を決定するファンクションcol_sum_func_tf_Factory

class col_sum_func_tf_Factory(vertica_sdk.TransformFunctionFactory):

def getPrototype(self, server_interface, arg_types, return_types):

arg_types.addInt()

return_types.addInt()

def getReturnType(self, server_interface, arg_types, return_type):

return_type.addColumn(arg_types.getColumnType(0),"sum")

return_type.addColumn(arg_types.getColumnType(1),"dif")

def createTransformFunction(cls, server_interface):

return col_sum_func_tf()

Pythonプログラム col_sum_func_tf.py をVerticaのライブラリPylibとして登録します。

dbadmin=> CREATE OR REPLACE LIBRARY schema.PylibTF AS '/home/dbadmin/col_sum_func_tf.py' LANGUAGE 'Python';

1	dbadmin=> CREATE OR REPLACE LIBRARY schema.PylibTF AS '/home/dbadmin/col_sum_func_tf.py' LANGUAGE 'Python';

ライブラリPylibに、ファンクションcol_sum_func_tfを定義します。
この際、入出力の型を定義するファンクションcol_sum_func_tf_Factoryも同時に定義します。

dbadmin=> CREATE OR REPLACE TRANSFORM FUNCTION  col_sum_func_tf AS NAME 'col_sum_func_tf_Factory' LIBRARY schema.PylibTF;

1	dbadmin=> CREATE OR REPLACE TRANSFORM FUNCTION col_sum_func_tf AS NAME 'col_sum_func_tf_Factory' LIBRARY schema.PylibTF;

ファンクションcol_sum_func_tfを呼び出すSQLを実行します。

dbadmin=> SELECT X1
dbadmin->       ,X2
dbadmin->       ,col_sum_func_tf(X1,X2)
dbadmin-> OVER(partition by X1,X2) 
dbadmin-> FROM schema.Table1;
 X1 | X2 | sum | dif
----+----+-----+-----
  2 |  3 |   5 |  -1
  5 |  7 |  12 |  -2
 11 | 10 |  21 |   1
(3 rows)

dbadmin=> SELECT X1

dbadmin-> ,X2

dbadmin-> ,col_sum_func_tf(X1,X2)

dbadmin-> OVER(partition by X1,X2)

dbadmin-> FROM schema.Table1;

X1 | X2 | sum | dif

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

2 | 3 | 5 | -1

5 | 7 | 12 | -2

11 | 10 | 21 | 1

(3 rows)

参考情報

Developing User-Defined Extensions (UDxs)
https://www.vertica.com/docs/9.3.x/HTML/Content/Authoring/ExtendingVertica/UDx/DevelopingUDxs.htm

Scalar Functions (UDSFs)
https://www.vertica.com/docs/9.3.x/HTML/Content/Authoring/ExtendingVertica/UDx/ScalarFunctions/ScalarFunctions.htm

Transform Functions (UDTFs)
https://www.vertica.com/docs/9.3.x/HTML/Content/Authoring/ExtendingVertica/UDx/TransformFunctions/TransformFunctions.htm

検証バージョンについて

この記事の内容はVertica 9.3.1で確認しています。

更新履歴

投稿タグ: 機械学習, machile learning

MENU

Pythonを利用したユーザー定義拡張関数

はじめに

ユーザー定義拡張関数(UDx)とは

ユーザー定義拡張関数の種類

スカラー関数

トランスフォーム関数

スカラー関数の利用例

例1：2倍の値を返す

例2：2列の和の値を返す

トランスフォーム関数の利用例

例3：2列の和と差の値を2列で返す

例4：2列の和と差の値を2行で返す

スカラー関数のプログラムコード例

トランスフォーム関数のプログラムコード例

参考情報

検証バージョンについて

更新履歴

カテゴリー

最近の投稿