E-MapReduce:Java UDF

スカラー UDF の開発

スカラー UDF は一度に 1 行を処理し、各行に対して単一の値を返します。クエリでスカラー UDF を使用すると、各行が結果セットに表示されます。一般的なスカラー関数には、UPPER、LOWER、ROUND、ABS などがあります。

次の例は、JSON オブジェクトからデータを抽出する方法を示しています。一部のビジネスシナリオでは、JSON データ内のフィールドの値が JSON オブジェクトではなく JSON 文字列である場合があります。ネストされた JSON 文字列を抽出するには、GET_JSON_STRING 関数のネストされた呼び出し (例：GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0")) を使用する必要があります。

SQL ステートメントを簡略化するために、UDF を開発して JSON 文字列を直接抽出できます (例：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 ライブラリは、フィールド値が JSON 形式の文字列であっても、パスを完全に展開できます。
            return JSONPath.read(jsonObj, key).toString();
        } catch (Exception e) {
            return null;
        }
    }
}

カスタムクラスには、次のメソッドを実装する必要があります。

ユーザー定義集計関数 (UDAF) の開発

UDAF は行のグループに対して動作し、単一の値を返します。一般的な集計関数には、SUM、COUNT、MAX、MIN などがあります。これらの関数は、各 GROUP BY グループ内の複数行のデータを集計し、グループごとに単一の結果を出力します。

次の例では、MY_SUM_INT 関数を使用します。BIGINT 値を返す組み込みの SUM 関数とは異なり、MY_SUM_INT は 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();
    }

    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;
    }
}

カスタムクラスには、次のメソッドを実装する必要があります。

UDAF を開発する場合、java.nio.ByteBuffer バッファクラスを使用して中間結果を格納し、serializeLength メソッドを使用してシリアル化された中間結果の長さを指定する必要があります。

ユーザー定義ウィンドウ関数 (UDWF) の開発

UDWF は、特殊な種類の集計関数です。通常の集計関数とは異なり、ウィンドウ関数は行のグループ (ウィンドウ) に対して値を計算し、各行に対して個別の結果を返します。通常、ウィンドウ関数には、行をパーティションに分割する OVER 句が含まれます。その後、関数は、その行が属するウィンドウに基づいて各行の結果を計算します。

次の例では、MY_WINDOW_SUM_INT 関数を使用します。BIGINT 値を返す組み込みの SUM 関数とは異なり、MY_WINDOW_SUM_INT は 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();
    }

    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];
        }
    }
}

ウィンドウ関数は特殊な種類の集計関数であるため、カスタムクラスには、必要なすべての UDAF メソッドに加えて、windowUpdate() メソッドを実装する必要があります。

ユーザー定義テーブル関数 (UDTF) の開発

UDTF は、単一の行を入力として受け取り、出力行のテーブルを生成します。UDTF は、列を複数の行に分割するなどの操作によく使用されます。

次の例では、MY_UDF_SPLIT 関数を使用します。スペースを区切り文字として使用して文字列を分割します。入力引数と戻り値はどちらも STRING 型です。

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" [, ...]) ]

パラメーター

スカラー UDF の作成

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

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 の作成

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

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 の作成

次のコマンドを実行して、前の例の 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 プロパティは、関数がウィンドウ関数であることを示します。UDWF の場合、これを true に設定します。他のパラメーターは、「スカラー UDF の作成」で説明されているものと同じです。

UDTF の作成

次のコマンドを実行して、前の例の 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 の使用

次のコマンドを実行して、前の例の 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."