E-MapReduce:Java UDF

Develop a scalar UDF

You can use scalar UDFs to perform operations on a single row of data and generate a single row of results. When you use scalar UDFs to query data, each row of data will eventually appear in the result set by row. Typical scalar UDFs include UPPER, LOWER, ROUND, and ABS.

The following example shows how to extract JSON data. For example, the value of a field in JSON data may be a JSON string instead of a JSON object. In this case, if you want to extract the JSON string, nested calling of the function GET_JSON_STRING in the SQL statement is required, which is GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0").

To simplify the SQL statement, you can develop a UDF to directly extract the JSON string, such as 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 {
            // The JSONPath library can be fully expanded, even if the value of a field is a JSON string.
            return JSONPath.read(jsonObj, key).toString();
        } catch (Exception e) {
            return null;
        }
    }
}

User-defined classes must implement the following method.

Develop a UDAF

You can use UDAFs to perform operations on multiple rows of data and generate a single row of results. Typical UDAFs include SUM, COUNT, MAX, and MIN. These functions can be used to aggregate multiple rows of data in each GROUP BY group and generate a single row of results.

In the following example, the function MY_SUM_INT is used. Different from the built-in function SUM whose return value is of the BIGINT data type, the data types of the input parameters and response parameters of the function MY_SUM_INT are 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;
    }
}

User-defined classes must implement the following methods.

When you develop a UDAF, you must use the buffer class java.nio.ByteBuffer to save and represent intermediate results and use the local variable serializeLength to specify the serialized length of the intermediate results.

Develop a UDWF

UDWF is the abbreviation for user-defined window function. Different from a common UDAF, a UDWF can be used to perform calculation for a set of rows (a window) and return a result for each row. In most cases, a UDWF contains an OVER clause that can be used to split data rows into multiple groups. The UDWF performs calculation based on the group (a window) in which each row of data resides and returns a result for each row.

In the following example, the function MY_WINDOW_SUM_INT is used. Different from the built-in function SUM whose return value is of the BIGINT data type, the data types of the input parameters and response parameters of the function MY_WINDOW_SUM_INT can be 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];
        }
    }
}

User-defined classes must implement the methods required by UDAFs and the windowUpdate() method. UDWFs are special UDAFs.

Develop a UDTF

A UDTF can be used to read a row of data and generate multiple values that can be considered as a table. UDTFs are commonly used to implement row-to-column conversion.

In the following example, the function MY_UDF_SPLIT is used. The function MY_UDF_SPLIT supports splitting strings by using a space as the delimiter. The data types of the input parameters and response parameters are STRING.

package com.starrocks.udf.sample;

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

User-defined classes must implement the following method.

Syntax

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

Parameters

Create a scalar UDF

Run the following commands to create a scalar UDF in 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"
);

Create a UDAF

Run the following commands to create a UDAF in 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"
);

The descriptions of the parameters in PROPERTIES are the same as those in properties in Create a scalar UDF.

Create a UDWF

Run the following commands to create the UDWF that is used in Step 3 in 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: Specifies whether the created function is a window function. Set the value to true. The descriptions of the remaining parameters are the same as those described in Create a scalar UDF.

Create a UDTF

Run the following commands to create the UDTF that is used in Step 3 in 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"
);

The descriptions of the parameters in PROPERTIES are the same as those in properties in Create a scalar UDF.

Use the scalar UDF

Run the following command to use the scalar UDF created in Step 6:

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

Use the UDAF

Run the following command to use the UDAF created in Step 6:

SELECT MY_SUM_INT(col1);

Use the UDWF

Run the following command to use the UDWF created in Step 6:

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

Use the UDTF

Run the following commands to use the UDTF created in Step 6:

-- Assume that table t1 that contains columns a, b, and c1 exists. 
SELECT t1.a,t1.b,t1.c1 FROM t1;
> output:
1,2.1,"hello world"
2,2.2,"hello UDTF."

-- Use the MY_UDF_SPLIT() function. 
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."