E-MapReduce:Java UDF

スカラー UDF を開発する

スカラー UDF を使用して、単一のデータ行に対して操作を実行し、単一の結果行を生成できます。スカラー UDF を使用してデータをクエリする場合、各データ行は最終的に結果セットに行ごとに表示されます。典型的なスカラー UDF には、UPPER、LOWER、ROUND、ABS などがあります。

次の例は、JSON データを抽出する方法を示しています。たとえば、JSON データのフィールドの値は、JSON オブジェクトではなく JSON 文字列である場合があります。この場合、JSON 文字列を抽出する場合、SQL 文で関数 GET_JSON_STRING のネスト呼び出し、つまり GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0") が必要になります。

SQL 文を簡略化するために、MY_UDF_JSON_GET('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key.k0") など、JSON 文字列を直接抽出する UDF を開発できます。

package com.starrocks.udf.sample;
import com.alibaba.fastjson.JSONPath;

public class UDFJsonGet {
    public final String evaluate(String jsonObj, String key) {
        if (obj == null || key == null) return null; // オブジェクトまたはキーが null の場合は null を返す
        try {
            // JSONPath ライブラリは、フィールドの値が JSON 文字列であっても完全に展開できます。
            return JSONPath.read(jsonObj, key).toString();
        } catch (Exception e) {
            return null; // 例外が発生した場合は null を返す
        }
    }
}

ユーザー定義クラスは、以下のメソッドを実装する必要があります。

UDAF を開発する

UDAF を使用して、複数のデータ行に対して操作を実行し、単一の結果行を生成できます。典型的な UDAF には、SUM、COUNT、MAX、MIN などがあります。これらの関数は、各 GROUP BY グループの複数のデータ行を集計し、単一の結果行を生成するために使用できます。

次の例では、関数 MY_SUM_INT を使用します。戻り値のデータ型が BIGINT である組み込み関数 SUM とは異なり、関数 MY_SUM_INT の入力パラメーターとレスポンスパラメーターのデータ型は INT です。

package com.starrocks.udf.sample;

public class SumInt {
    public static class State {
        int counter = 0; // カウンター変数
        public int serializeLength() { return 4; } // シリアル化後の長さ
    }

    public State create() {
        return new State(); // State オブジェクトを作成
    }

    public void destroy(State state) {
        // State オブジェクトを破棄
    }

    public final void update(State state, Integer val) {
        if (val != null) {
            state.counter+= val; // 値をカウンターに追加
        }
    }

    public void serialize(State state, java.nio.ByteBuffer buff) {
        buff.putInt(state.counter); // カウンターをシリアル化
    }

    public void merge(State state, java.nio.ByteBuffer buffer) {
        int val = buffer.getInt(); // バッファーから値を取得
        state.counter += val; // 値をカウンターに追加
    }

    public Integer finalize(State state) {
        return state.counter; // 最終結果を返す
    }
}

ユーザー定義クラスは、以下のメソッドを実装する必要があります。

UDAF を開発する場合は、バッファークラス java.nio.ByteBuffer を使用して中間結果を保存および表現し、ローカル変数 serializeLength を使用して中間結果のシリアル化された長さを指定する必要があります。

UDWF を開発する

UDWF は、ユーザー定義ウィンドウ関数の略語です。一般的な UDAF とは異なり、UDWF は一連の行 (ウィンドウ) の計算を実行し、各行の結果を返すために使用できます。ほとんどの場合、UDWF には、データ行を複数のグループに分割するために使用できる OVER 句が含まれています。UDWF は、各データ行が存在するグループ (ウィンドウ) に基づいて計算を実行し、各行の結果を返します。

次の例では、関数 MY_WINDOW_SUM_INT を使用します。戻り値のデータ型が BIGINT である組み込み関数 SUM とは異なり、関数 MY_WINDOW_SUM_INT の入力パラメーターとレスポンスパラメーターのデータ型は INT にすることができます。

package com.starrocks.udf.sample;

public class WindowSumInt {    
    public static class State {
        int counter = 0; // カウンター変数
        public int serializeLength() { return 4; } // シリアル化後の長さ

        @Override
        public String toString() {
            return "State{" +
                    "counter=" + counter +
                    '}';
        }
    }

    public State create() {
        return new State(); // State オブジェクトを作成
    }

    public void destroy(State state) {
        // State オブジェクトを破棄
    }

    public void update(State state, Integer val) {
        if (val != null) {
            state.counter+=val; // 値をカウンターに追加
        }
    }

    public void serialize(State state, java.nio.ByteBuffer buff) {
        buff.putInt(state.counter); // カウンターをシリアル化
    }

    public void merge(State state, java.nio.ByteBuffer buffer) {
        int val = buffer.getInt(); // バッファーから値を取得
        state.counter += val; // 値をカウンターに追加
    }

    public Integer finalize(State state) {
        return state.counter; // 最終結果を返す
    }

    public void reset(State state) {
        state.counter = 0; // カウンターをリセット
    }

    public void windowUpdate(State state,
                            int peer_group_start, int peer_group_end,
                            int frame_start, int frame_end,
                            Integer[] inputs) {
        for (int i = (int)frame_start; i < (int)frame_end; ++i) {
            state.counter += inputs[i]; // ウィンドウ内の入力データを追加
        }
    }
}

ユーザー定義クラスは、UDAF に必要なメソッドと windowUpdate() メソッドを実装する必要があります。UDWF は特別な UDAF です。

UDTF を開発する

UDTF は、データ行を読み取り、テーブルと見なすことができる複数の値を生成するために使用できます。UDTF は、行から列への変換を実装するためによく使用されます。

次の例では、関数 MY_UDF_SPLIT を使用します。関数 MY_UDF_SPLIT は、スペースを区切り文字として使用して文字列を分割することをサポートしています。入力パラメーターとレスポンスパラメーターのデータ型は STRING です。

package com.starrocks.udf.sample;

public class UDFSplit{
    public String[] process(String in) { // 入力文字列を処理するメソッド
        if (in == null) return null; // 入力が null の場合は null を返す
        return in.split(" "); // スペースで分割して配列を返す
    }
}

ユーザー定義クラスは、以下のメソッドを実装する必要があります。

構文

CREATE [GLOBAL][AGGREGATE | TABLE] FUNCTION function_name(arg_type [, ...])
RETURNS return_type
[PROPERTIES ("key" = "value" [, ...]) ]

パラメーター

スカラー UDF を作成する

次のコマンドを実行して、StarRocks にスカラー UDF を作成します。

CREATE [GLOBAL] FUNCTION MY_UDF_JSON_GET(string, string) 
RETURNS string
PROPERTIES (
    "symbol" = "com.starrocks.udf.sample.UDFJsonGet", 
    "type" = "StarrocksJar",
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

UDAF を作成する

次のコマンドを実行して、StarRocks に UDAF を作成します。

CREATE [GLOBAL] AGGREGATE FUNCTION MY_SUM_INT(INT) 
RETURNS INT
PROPERTIES 
( 
    "symbol" = "com.starrocks.udf.sample.SumInt", 
    "type" = "StarrocksJar",
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

PROPERTIES のパラメーターの説明は、「スカラー UDF を作成する」のプロパティの説明と同じです。

UDWF を作成する

次のコマンドを実行して、手順 3 で使用した UDWF を StarRocks に作成します。

CREATE [GLOBAL] AGGREGATE FUNCTION MY_WINDOW_SUM_INT(Int)
RETURNS Int
PROPERTIES 
(
    "analytic" = "true",
    "symbol" = "com.starrocks.udf.sample.WindowSumInt", 
    "type" = "StarrocksJar", 
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

analytic: 作成される関数がウィンドウ関数であるかどうかを指定します。値を true に設定します。残りのパラメーターの説明は、「スカラー UDF を作成する」で説明されているものと同じです。

UDTF を作成する

次のコマンドを実行して、手順 3 で使用した UDTF を StarRocks に作成します。

CREATE [GLOBAL] TABLE FUNCTION MY_UDF_SPLIT(string)
RETURNS string
PROPERTIES 
(
    "symbol" = "com.starrocks.udf.sample.UDFSplit", 
    "type" = "StarrocksJar", 
    "file" = "http://<YourBucketName>.oss-cn-xxxx-internal.aliyuncs.com/<YourPath>/udf-1.0-SNAPSHOT-jar-with-dependencies.jar"
);

PROPERTIES のパラメーターの説明は、「スカラー UDF を作成する」のプロパティの説明と同じです。

スカラー UDF を使用する

次のコマンドを実行して、手順 6 で作成したスカラー UDF を使用します。

SELECT MY_UDF_JSON_GET('{"key":"{\\"in\\":2}"}', '$.key.in');

UDAF を使用する

次のコマンドを実行して、手順 6 で作成した UDAF を使用します。

SELECT MY_SUM_INT(col1);

UDWF を使用する

次のコマンドを実行して、手順 6 で作成した UDWF を使用します。

SELECT MY_WINDOW_SUM_INT(intcol) 
            OVER (PARTITION BY intcol2
                  ORDER BY intcol3
                  ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)
FROM test_basic;

UDTF を使用する

次のコマンドを実行して、手順 6 で作成した UDTF を使用します。

-- 列 a、b、c1 を含むテーブル t1 が存在すると仮定します。
SELECT t1.a,t1.b,t1.c1 FROM t1;
> output:
1,2.1,"hello world"
2,2.2,"hello UDTF."

-- MY_UDF_SPLIT() 関数を使用します。
SELECT t1.a,t1.b, MY_UDF_SPLIT FROM t1, MY_UDF_SPLIT(t1.c1); 
> output:
1,2.1,"hello"
1,2.1,"world"
2,2.2,"hello"
2,2.2,"UDTF."