E-MapReduce：Java UDF

Kembangkan scalar UDF

Anda dapat menggunakan scalar UDF untuk melakukan operasi pada satu baris data dan menghasilkan satu baris hasil. Saat menggunakan scalar UDF untuk menanyakan data, setiap baris data akhirnya akan muncul di set hasil berdasarkan baris. Scalar UDF tipikal termasuk UPPER, LOWER, ROUND, dan ABS.

Contoh berikut menunjukkan cara mengekstrak data JSON. Misalnya, nilai bidang dalam data JSON dapat berupa string JSON, bukan objek JSON. Dalam situasi ini, untuk mengekstrak string JSON, diperlukan pemanggilan bersarang fungsi GET_JSON_STRING dalam pernyataan SQL, seperti GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0").

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

Kelas yang ditentukan pengguna harus mengimplementasikan metode berikut.

Kembangkan UDAF

Anda dapat menggunakan UDAF untuk melakukan operasi pada beberapa baris data dan menghasilkan satu baris hasil. UDAF tipikal termasuk SUM, COUNT, MAX, dan MIN. Fungsi-fungsi ini dapat digunakan untuk menggabungkan beberapa baris data dalam setiap grup GROUP BY dan menghasilkan satu baris hasil.

Dalam contoh berikut, fungsi MY_SUM_INT digunakan. Berbeda dengan fungsi bawaan SUM yang mengembalikan nilai bertipe BIGINT, tipe data parameter input dan respons untuk fungsi MY_SUM_INT adalah 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 yang ditentukan pengguna harus mengimplementasikan metode-metode berikut.

Saat mengembangkan UDAF, gunakan kelas buffer java.nio.ByteBuffer untuk menyimpan dan merepresentasikan hasil sementara, serta variabel lokal serializeLength untuk menentukan panjang serialisasi hasil tersebut.

Kembangkan UDWF

UDWF adalah singkatan dari user-defined window function. Berbeda dengan UDAF umum, UDWF dapat digunakan untuk melakukan perhitungan untuk sekumpulan baris (sebuah jendela) dan mengembalikan hasil untuk setiap baris. Dalam banyak kasus, UDWF berisi klausa OVER yang dapat digunakan untuk membagi baris data menjadi beberapa kelompok. UDWF melakukan perhitungan berdasarkan kelompok (sebuah jendela) tempat setiap baris data berada dan mengembalikan hasil untuk setiap baris.

Dalam contoh berikut, fungsi MY_WINDOW_SUM_INT digunakan. Berbeda dengan fungsi bawaan SUM yang mengembalikan nilai bertipe BIGINT, tipe data parameter input dan respons untuk fungsi MY_WINDOW_SUM_INT dapat berupa 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];
        }
    }
}

Kelas yang ditentukan pengguna harus mengimplementasikan metode yang diperlukan oleh UDAF dan metode windowUpdate(). UDWF adalah UDAF khusus.

Kembangkan UDTF

UDTF dapat digunakan untuk membaca baris data dan menghasilkan beberapa nilai yang dapat dianggap sebagai tabel. UDTF sering digunakan untuk mengimplementasikan konversi baris-ke-kolom.

Dalam contoh berikut, fungsi MY_UDF_SPLIT digunakan. Fungsi MY_UDF_SPLIT mendukung pemisahan string menggunakan spasi sebagai pembatas. Tipe data parameter input dan parameter respons adalah STRING.

package com.starrocks.udf.sample;

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

Kelas yang ditentukan pengguna harus mengimplementasikan metode berikut.

Sintaksis

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

Deskripsi parameter dalam PROPERTIES sama dengan yang dijelaskan dalam bagian properties di Buat scalar UDF.

Buat UDWF

Jalankan perintah berikut untuk membuat UDWF yang digunakan pada Langkah 3 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"
);

analytic: Menentukan apakah fungsi yang dibuat adalah fungsi jendela. Atur nilainya menjadi true. Deskripsi parameter sisanya sama dengan yang dijelaskan di Buat Scalar UDF.

Buat UDTF

Jalankan perintah berikut untuk membuat UDTF yang digunakan pada Langkah 3 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"
);

Deskripsi parameter dalam PROPERTIES sama dengan yang dijelaskan pada bagian properties di 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 yang dibuat pada Langkah 6:

-- Anggaplah tabel t1 yang berisi kolom a, b, dan c1 ada.
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."