E-MapReduce:Java UDF

スカラー UDF

スカラ UDF は、evaluate メソッドを public メンバーメソッドとして実装する必要があります。メソッドシグネチャによって SQL パラメーターおよび戻り値の型が決まります。これらは CREATE FUNCTION 文で宣言する型と一致している必要があります（データ型マッピングをご参照ください）。

以下の例では、ドット区切りのパス式を使用してネストされた JSON 値を抽出する MY_UDF_JSON_GET を実装しています。これは、ネストされた GET_JSON_STRING(GET_JSON_STRING(...)) パターンを、単一の呼び出し MY_UDF_JSON_GET('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key.k0") に置き換えます。

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

public class UDFJsonGet {
    public final String evaluate(String jsonObj, String key) {
        if (jsonObj == null || key == null) return null;
        try {
            // JSONPath.read はネストされた JSON 文字列を完全に展開します
            return JSONPath.read(jsonObj, key).toString();
        } catch (Exception e) {
            return null;
        }
    }
}

UDAF

UDAF は、グループごとの複数の行を単一の結果に集約します。これは、中間結果を保持するための State 内部クラスを使用し、実行ノード間でデータを送信する際に StarRocks によってシリアル化および逆シリアル化されます。

必須メソッド — すべての UDAF で 6 つすべてを実装します：

中間状態バッファー — java.nio.ByteBuffer を使用して中間結果を格納します：

以下の例では、MY_SUM_INT（INT 型の入力と出力を持つ合計関数）を実装しています（組み込みの SUM 関数とは異なり、常に BIGINT を返します）：

package com.starrocks.udf.sample;

public class SumInt {
    public static class State {
        int counter = 0;
        public int serializeLength() { return 4; } // INT = 4 バイト
    }

    public State create() {
        return new State();
    }

    public void destroy(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;
    }
}

UDWF

ユーザー定義ウィンドウ関数 (UDWF) は、グループごとに 1 つの結果を返すのではなく、入力行ごとに 1 つの結果を返す特殊な UDAF です。これは、OVER 句を使用してパーティションとウィンドウフレームを定義し、標準の UDAF インターフェイスに windowUpdate メソッドを追加します。

6 つの UDAF メソッドすべてに加えて、windowUpdate を実装します：

`windowUpdate` パラメーター：

以下の例では、INT のウィンドウ合計である MY_WINDOW_SUM_INT を実装します:

package com.starrocks.udf.sample;

public class WindowSumInt {
    public static class State {
        int counter = 0;
        public int serializeLength() { return 4; }
    }

    public State create() {
        return new State();
    }

    public void destroy(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];
        }
    }
}

ウィンドウ関数の構文の詳細については、「ウィンドウ関数」をご参照ください。

UDTF

UDTF は 1 つの入力行を読み取り、単一の列に複数の行を返します。配列を返す process メソッドを実装する必要があります。

以下の例では、文字列をスペースで分割する MY_UDF_SPLIT を実装します：

package com.starrocks.udf.sample;

public class UDFSplit{
    public String[] process(String in) {
        if (in == null) return null;
        return in.split(" ");
    }
}

構文

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

パラメーター

PROPERTIES パラメーター

スカラー 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 の作成

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"
);

UDWF の作成

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"
);

UDTF の作成

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"
);

スカラー UDF

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

UDAF

SELECT MY_SUM_INT(col1);

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

-- テーブル t1 には a, b, c1 の列があると仮定します
SELECT t1.a, t1.b, t1.c1 FROM t1;
-- 出力：
-- 1, 2.1, "hello world"
-- 2, 2.2, "hello UDTF."

-- c1 を 1 行に 1 単語ずつ分割します
SELECT t1.a, t1.b, MY_UDF_SPLIT FROM t1, MY_UDF_SPLIT(t1.c1);
-- 出力：
-- 1, 2.1, "hello"
-- 1, 2.1, "world"
-- 2, 2.2, "hello"
-- 2, 2.2, "UDTF."