E-MapReduce:Java UDF

Scalar UDF

Scalar UDF harus mengimplementasikan metode evaluate sebagai metode anggota public. Signature metode ini menentukan tipe parameter dan tipe kembalian SQL — tipe tersebut harus sesuai dengan tipe yang Anda deklarasikan dalam pernyataan CREATE FUNCTION (lihat Pemetaan tipe data).

Contoh berikut mengimplementasikan MY_UDF_JSON_GET, yang mengekstraksi nilai JSON bersarang menggunakan ekspresi path bertitik. Fungsi ini menggantikan pola bersarang GET_JSON_STRING(GET_JSON_STRING(...)) dengan satu panggilan: 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 sepenuhnya mengekspansi string JSON bersarang
            return JSONPath.read(jsonObj, key).toString();
        } catch (Exception e) {
            return null;
        }
    }
}

UDAF

UDAF mengagregasi beberapa baris per kelompok menjadi satu hasil. Fungsi ini menggunakan kelas dalam State untuk menyimpan hasil antara, yang diserialisasi dan dideserialisasi oleh StarRocks saat mentransmisikan data antar node eksekusi.

Metode wajib — implementasikan keenam metode berikut untuk setiap UDAF:

Buffer state antara — gunakan java.nio.ByteBuffer untuk menyimpan hasil antara:

Contoh berikut mengimplementasikan MY_SUM_INT, penjumlahan INT-in/INT-out (berbeda dengan SUM bawaan yang selalu mengembalikan BIGINT):

package com.starrocks.udf.sample;

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

    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 (user-defined window function) adalah UDAF khusus yang mengembalikan satu hasil per baris input, bukan satu hasil per kelompok. Fungsi ini menggunakan klausa OVER untuk menentukan partisi dan frame jendela, serta menambahkan metode windowUpdate ke antarmuka UDAF standar.

Implementasikan keenam metode UDAF ditambah windowUpdate:

Parameter `windowUpdate`:

Contoh berikut mengimplementasikan MY_WINDOW_SUM_INT, penjumlahan jendela 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];
        }
    }
}

Untuk informasi lebih lanjut tentang sintaks fungsi jendela, lihat Window functions.

UDTF

UDTF membaca satu baris input dan mengembalikan beberapa baris, semuanya dalam satu kolom. Fungsi ini harus mengimplementasikan metode process, yang mengembalikan array.

Contoh berikut mengimplementasikan MY_UDF_SPLIT, yang memisahkan string berdasarkan spasi:

package com.starrocks.udf.sample;

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

Sintaks

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

Parameter

Parameter PROPERTIES

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

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

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

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

Scalar 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

-- Asumsikan tabel t1 memiliki kolom a, b, dan c1
SELECT t1.a, t1.b, t1.c1 FROM t1;
-- Output:
-- 1, 2.1, "hello world"
-- 2, 2.2, "hello UDTF."

-- Pisahkan c1 menjadi satu kata per baris
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."