E-MapReduce:Java UDF

Kembangkan scalar UDF

Scalar UDF memproses satu baris dalam satu waktu dan mengembalikan satu nilai untuk setiap baris. Saat digunakan dalam kueri, setiap baris muncul dalam set hasil. Contoh fungsi scalar umum meliputi UPPER, LOWER, ROUND, dan ABS.

Contoh berikut menunjukkan cara mengekstraksi data dari objek JSON. Dalam beberapa skenario bisnis, nilai suatu bidang dalam data JSON mungkin berupa string JSON, bukan objek JSON. Untuk mengekstraksi string JSON bersarang tersebut, Anda harus menggunakan pemanggilan bersarang ke fungsi GET_JSON_STRING, seperti GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0").

Untuk menyederhanakan pernyataan SQL, Anda dapat mengembangkan UDF untuk mengekstraksi string JSON secara langsung, misalnya 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 {
            // Pustaka JSONPath dapat sepenuhnya memperluas path, bahkan jika nilai bidang adalah string berformat JSON.
            return JSONPath.read(jsonObj, key).toString();
        } catch (Exception e) {
            return null;
        }
    }
}

Kelas kustom Anda harus mengimplementasikan metode berikut.

Kembangkan UDAF

UDAF beroperasi pada sekelompok baris dan mengembalikan satu nilai. Contoh fungsi agregat umum meliputi SUM, COUNT, MAX, dan MIN. Fungsi-fungsi ini mengagregasi data dari beberapa baris dalam setiap kelompok GROUP BY dan menghasilkan satu hasil untuk setiap kelompok.

Contoh berikut menggunakan fungsi MY_SUM_INT. Berbeda dengan fungsi bawaan SUM yang mengembalikan nilai BIGINT, MY_SUM_INT menerima argumen INT dan mengembalikan 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;
    }
}

Kelas kustom Anda harus mengimplementasikan metode berikut.

Saat mengembangkan UDAF, Anda harus menggunakan kelas buffer java.nio.ByteBuffer untuk menyimpan hasil antara dan metode serializeLength untuk menentukan panjang hasil antara yang telah diserialisasi.

Kembangkan UDWF

UDWF adalah jenis khusus dari fungsi agregat. Berbeda dengan fungsi agregat biasa, fungsi jendela menghitung nilai pada sekelompok baris (jendela) dan mengembalikan hasil terpisah untuk setiap baris. Biasanya, fungsi jendela mencakup klausa OVER yang membagi baris ke dalam partisi. Fungsi tersebut kemudian menghitung hasil untuk setiap baris berdasarkan jendela tempat baris tersebut berada.

Contoh berikut menggunakan fungsi MY_WINDOW_SUM_INT. Berbeda dengan fungsi bawaan SUM yang mengembalikan nilai BIGINT, MY_WINDOW_SUM_INT menerima argumen INT dan mengembalikan 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];
        }
    }
}

Karena fungsi jendela merupakan jenis khusus dari fungsi agregat, kelas kustom Anda harus mengimplementasikan semua metode UDAF yang diperlukan, ditambah metode windowUpdate().

Kembangkan UDTF

UDTF menerima satu baris sebagai input dan menghasilkan tabel berisi beberapa baris output. UDTF sering digunakan untuk operasi seperti memisahkan satu kolom menjadi beberapa baris.

Contoh berikut menggunakan fungsi MY_UDF_SPLIT. Fungsi ini memisahkan string dengan spasi sebagai pemisah. Baik argumen input maupun nilai kembali bertipe STRING.

package com.starrocks.udf.sample;

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

Kelas kustom Anda harus mengimplementasikan metode berikut.

Sintaks

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

Parameter

Buat scalar UDF

Jalankan perintah berikut untuk membuat scalar UDF dari contoh sebelumnya di 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"
);

Buat UDAF

Jalankan perintah berikut untuk membuat UDAF dari contoh sebelumnya di 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"
);

Parameter dalam klausa PROPERTIES sama dengan yang dijelaskan dalam Buat scalar UDF.

Buat UDWF

Jalankan perintah berikut untuk membuat UDWF dari contoh sebelumnya di 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"
);

Properti analytic mengidentifikasi fungsi sebagai fungsi jendela. Untuk UDWF, atur nilai ini menjadi true. Parameter lainnya sama dengan yang dijelaskan dalam Buat scalar UDF.

Buat UDTF

Jalankan perintah berikut untuk membuat UDTF dari contoh sebelumnya di 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"
);

Parameter dalam klausa PROPERTIES sama dengan yang dijelaskan dalam Buat scalar UDF.

Gunakan scalar UDF

Jalankan perintah berikut untuk menggunakan scalar UDF yang dibuat pada Langkah 6.

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

Gunakan UDAF

Jalankan perintah berikut untuk menggunakan UDAF yang dibuat pada Langkah 6.

SELECT MY_SUM_INT(col1);

Gunakan UDWF

Jalankan perintah berikut untuk menggunakan UDWF yang dibuat pada Langkah 6.

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

Gunakan UDTF

Jalankan perintah berikut untuk menggunakan UDTF dari contoh sebelumnya.

-- Asumsikan tabel t1 ada dengan 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."

-- Gunakan fungsi 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."