E-MapReduce：Java UDF

開發Scalar UDF

Scalar UDF，即使用者自訂純量涵式，可以對單行資料進行操作，輸出單行結果。當您在查詢時使用Scalar 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語句，您可以開發一個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 (obj == 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函數為例進行說明。與內建函數SUM（傳回值為BIGINT類型）區別在於，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();
    }

    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函數為例進行說明。與內建函數SUM（傳回型別為BIGINT）區別在於，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();
    }

    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，即使用者自訂表格值函數，讀入一行資料，輸出多個值可視為一張表。資料表值函式常用於實現行轉列。

以下樣本以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;
        return in.split(" ");
    }
}

使用者自訂類必須實現如下方法。

文法

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

參數說明

建立Scalar UDF

執行如下命令，在StarRocks中建立之前樣本中的Scalar 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裡的參數說明與建立Scalar UDF相同。

建立UDWF

執行如下命令，在StarRocks中建立先前樣本中的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"
);

analytic：所建立的函數是否為視窗函數，固定取值為true。其他參數說明與建立Scalar UDF相同。

建立UDTF

執行如下命令，在StarRocks中建立先前樣本中的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"
);

PROPERTIES裡的參數說明與建立Scalar UDF相同。

使用Scalar UDF

執行如下命令，使用步驟六建立的Scalar UDF函數。

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

使用UDAF

執行如下命令，使用步驟六建立的UDAF函數。

SELECT MY_SUM_INT(col1);

使用UDWF

執行如下命令，使用步驟六建立的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。

-- 假設存在表 t1，其列 a、b、c1 資訊如下。
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."