開發(fā)Scalar UDF

Scalar UDF，即用戶自定義標(biāo)量函數(shù)，可以對單行數(shù)據(jù)進(jìn)行操作，輸出單行結(jié)果。當(dāng)您在查詢時使用Scalar UDF，每行數(shù)據(jù)最終都會按行出現(xiàn)在結(jié)果集中。典型的標(biāo)量函數(shù)包括UPPER、LOWER、ROUND、ABS。

以下示例以提取JSON數(shù)據(jù)功能為例進(jìn)行說明。例如，業(yè)務(wù)場景中，JSON數(shù)據(jù)中某個字段的值可能是JSON字符串而不是JSON對象，因此在提取JSON字符串時，SQL語句需要嵌套調(diào)用GET_JSON_STRING，即GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0")。

為簡化SQL語句，您可以開發(fā)一個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;
        }
    }
}

用戶自定義類必須實現(xiàn)如下方法。

開發(fā)UDAF

UDAF，即用戶自定義的聚合函數(shù)，對多行數(shù)據(jù)進(jìn)行操作，輸出單行結(jié)果。典型的聚合函數(shù)包括SUM、COUNT、MAX、MIN，這些函數(shù)對于每個GROUP BY分組中多行數(shù)據(jù)進(jìn)行聚合后，只輸出一行結(jié)果。

以下示例以MY_SUM_INT函數(shù)為例進(jìn)行說明。與內(nèi)置函數(shù)SUM（返回值為BIGINT類型）區(qū)別在于，MY_SUM_INT函數(shù)支持傳入?yún)?shù)和返回參數(shù)的類型為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;
    }
}

用戶自定義類必須實現(xiàn)如下方法。

并且，開發(fā)UDAF函數(shù)時，您需要使用緩沖區(qū)類java.nio.ByteBuffer和局部變量serializeLength，用于保存和表示中間結(jié)果，指定中間結(jié)果的序列化長度。

開發(fā)UDWF

UDWF，即用戶自定義窗口函數(shù)。跟普通聚合函數(shù)不同的是，窗口函數(shù)針對一組行（一個窗口）計算值，并為每行返回一個結(jié)果。一般情況下，窗口函數(shù)包含OVER子句，將數(shù)據(jù)行拆分成多個分組，窗口函數(shù)基于每一行數(shù)據(jù)所在的組（一個窗口）進(jìn)行計算，并為每行返回一個結(jié)果。

以下示例以MY_WINDOW_SUM_INT函數(shù)為例進(jìn)行說明。與內(nèi)置函數(shù)SUM（返回類型為BIGINT）區(qū)別在于，MY_WINDOW_SUM_INT函數(shù)支持傳入?yún)?shù)和返回參數(shù)的類型為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];
        }
    }
}

用戶自定義類必須實現(xiàn)UDAF所需要的方法（窗口函數(shù)是特殊聚合函數(shù)）以及windowUpdate()方法。

需要額外實現(xiàn)的方法

開發(fā)UDTF

UDTF，即用戶自定義表值函數(shù)，讀入一行數(shù)據(jù)，輸出多個值可視為一張表。表值函數(shù)常用于實現(xiàn)行轉(zhuǎn)列。

以下示例以MY_UDF_SPLIT函數(shù)為例進(jìn)行說明。MY_UDF_SPLIT函數(shù)支持分隔符為空格，傳入?yún)?shù)和返回參數(shù)的類型為STRING。

package com.starrocks.udf.sample;

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

用戶自定義類必須實現(xiàn)如下方法。

語法

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

參數(shù)說明

創(chuàng)建Scalar UDF

執(zhí)行如下命令，在StarRocks中創(chuàng)建之前示例中的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"
);

創(chuàng)建UDAF

執(zhí)行如下命令，在StarRocks中創(chuàng)建之前示例中的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里的參數(shù)說明與創(chuàng)建Scalar UDF相同。

創(chuàng)建UDWF

執(zhí)行如下命令，在StarRocks中創(chuàng)建先前示例中的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：所創(chuàng)建的函數(shù)是否為窗口函數(shù)，固定取值為true。其他參數(shù)說明與創(chuàng)建Scalar UDF相同。

創(chuàng)建UDTF

執(zhí)行如下命令，在StarRocks中創(chuàng)建先前示例中的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里的參數(shù)說明與創(chuàng)建Scalar UDF相同。

使用Scalar UDF

執(zhí)行如下命令，使用步驟六創(chuàng)建的Scalar UDF函數(shù)。

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

使用UDAF

執(zhí)行如下命令，使用步驟六創(chuàng)建的UDAF函數(shù)。

SELECT MY_SUM_INT(col1);

使用UDWF

執(zhí)行如下命令，使用步驟六創(chuàng)建的UDWF函數(shù)。

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

使用UDTF

執(zhí)行如下命令，使用先前示例中的UDTF。

-- 假設(shè)存在表 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() 函數(shù)。
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."