E-MapReduce:Java UDF

project
|--pom.xml
|--src
|  |--main
|  |  |--java
|  |  |--resources
|  |--test
|--target

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>udf</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.76</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-dependency-plugin</artifactId>
                <version>2.10</version>
                <executions>
                    <execution>
                        <id>copy-dependencies</id>
                        <phase>package</phase>
                        <goals>
                            <goal>copy-dependencies</goal>
                        </goals>
                        <configuration>
                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.3.0</version>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

Desarrollar una scalar UDF

Una scalar UDF procesa una fila a la vez y devuelve un único valor por cada fila. Cuando se utiliza una scalar UDF en una consulta, cada fila aparece en el conjunto de resultados. Las funciones escalares típicas incluyen UPPER, LOWER, ROUND y ABS.

El siguiente ejemplo muestra cómo extraer datos de un objeto JSON. En algunos escenarios de negocio, el valor de un campo en datos JSON puede ser una cadena JSON en lugar de un objeto JSON. Para extraer la cadena JSON anidada, debe usar una llamada anidada a la función GET_JSON_STRING, como GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0").

Para simplificar la instrucción SQL, puede desarrollar una UDF que extraiga la cadena JSON directamente; por ejemplo, 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 fully expand the path, even if a field value is a JSON-formatted string.
            return JSONPath.read(jsonObj, key).toString();
        } catch (Exception e) {
            return null;
        }
    }
}

Su clase personalizada debe implementar el siguiente método.

Desarrollar una UDAF

Una UDAF opera sobre un grupo de filas y devuelve un único valor. Las funciones de agregación típicas incluyen SUM, COUNT, MAX y MIN. Estas funciones agregan datos de varias filas dentro de cada grupo GROUP BY y generan un único resultado por grupo.

El siguiente ejemplo utiliza la función MY_SUM_INT. A diferencia de la función integrada SUM, que devuelve un valor BIGINT, MY_SUM_INT toma un argumento INT y devuelve un 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;
    }
}

Su clase personalizada debe implementar los siguientes métodos.

Al desarrollar una UDAF, debe utilizar la clase de búfer java.nio.ByteBuffer para almacenar el resultado intermedio y el método serializeLength para especificar la longitud del resultado intermedio serializado.

Desarrollar una UDWF

Una UDWF es un tipo especial de función de agregación. A diferencia de una función de agregación normal, una función de ventana calcula un valor sobre un grupo de filas (una ventana) y devuelve un resultado separado para cada fila. Por lo general, una función de ventana incluye una cláusula OVER que divide las filas en particiones. Luego, la función calcula un resultado para cada fila basándose en la ventana de filas a la que pertenece.

El siguiente ejemplo utiliza la función MY_WINDOW_SUM_INT. A diferencia de la función integrada SUM, que devuelve un valor BIGINT, MY_WINDOW_SUM_INT toma un argumento INT y devuelve un 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];
        }
    }
}

Dado que una función de ventana es un tipo especial de función de agregación, su clase personalizada debe implementar todos los métodos requeridos para UDAF, además del método windowUpdate().

Desarrollar una UDTF

Una UDTF toma una sola fila como entrada y produce una tabla de filas de salida. Las UDTF se utilizan frecuentemente para operaciones como dividir una columna en varias filas.

El siguiente ejemplo utiliza la función MY_UDF_SPLIT. Esta divide una cadena utilizando un espacio como separador. Tanto el argumento de entrada como el valor devuelto son del tipo STRING.

package com.starrocks.udf.sample;

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

Su clase personalizada debe implementar el siguiente método.

mvn package

Sintaxis

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

Parámetros

Crear una scalar UDF

Ejecute el siguiente comando para crear la scalar UDF del ejemplo anterior en 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"
);

Crear una UDAF

Ejecute el siguiente comando para crear la UDAF del ejemplo anterior en 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"
);

Los parámetros en la cláusula PROPERTIES son los mismos que se describen en Create a scalar UDF.

Crear una UDWF

Ejecute el siguiente comando para crear la UDWF del ejemplo anterior en 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"
);

La propiedad analytic identifica la función como una función de ventana. Para una UDWF, establézcala en true. Los demás parámetros son los mismos que se describen en Create a scalar UDF.

Crear una UDTF

Ejecute el siguiente comando para crear la UDTF del ejemplo anterior en 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"
);

Los parámetros en la cláusula PROPERTIES son los mismos que se describen en Create a scalar UDF.

Utilizar la scalar UDF

Ejecute el siguiente comando para utilizar la scalar UDF creada en el paso 6.

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

Utilizar la UDAF

Ejecute el siguiente comando para utilizar la UDAF creada en el paso 6.

SELECT MY_SUM_INT(col1);

Utilizar la UDWF

Ejecute el siguiente comando para utilizar la UDWF creada en el paso 6.

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

Utilizar la UDTF

Ejecute el siguiente comando para utilizar la UDTF del ejemplo anterior.

-- Assume that table t1 exists with columns a, b, and c1.
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."