E-MapReduce:Fonction UDF Java

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>

Développer une scalar UDF

Une scalar UDF traite une ligne à la fois et renvoie une seule valeur pour chaque ligne. Lorsque vous utilisez une scalar UDF dans une requête, chaque ligne apparaît dans l'ensemble de résultats. Les fonctions scalaires typiques incluent UPPER, LOWER, ROUND et ABS.

L'exemple suivant montre comment extraire des données d'un objet JSON. Dans certains scénarios métier, la valeur d'un champ dans les données JSON peut être une chaîne JSON plutôt qu'un objet JSON. Pour extraire la chaîne JSON imbriquée, vous devez utiliser un appel imbriqué à la fonction GET_JSON_STRING, tel que GET_JSON_STRING(GET_JSON_STRING('{"key":"{\\"k0\\":\\"v0\\"}"}', "$.key"), "$.k0").

Pour simplifier l'instruction SQL, vous pouvez développer une UDF qui extrait directement la chaîne JSON, par exemple 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;
        }
    }
}

Votre classe personnalisée doit implémenter la méthode suivante.

Développer une UDAF

Une UDAF opère sur un groupe de lignes et renvoie une seule valeur. Les fonctions d'agrégation typiques incluent SUM, COUNT, MAX et MIN. Ces fonctions agrègent les données de plusieurs lignes au sein de chaque groupe GROUP BY et produisent un seul résultat pour chaque groupe.

L'exemple suivant utilise la fonction MY_SUM_INT. Contrairement à la fonction intégrée SUM, qui renvoie une valeur BIGINT, MY_SUM_INT accepte un argument INT et renvoie 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;
    }
}

Votre classe personnalisée doit implémenter les méthodes suivantes.

Lors du développement d'une UDAF, vous devez utiliser la classe de tampon java.nio.ByteBuffer pour stocker le résultat intermédiaire et la méthode serializeLength pour spécifier la longueur du résultat intermédiaire sérialisé.

Développer une UDWF

Une UDWF est un type spécial de fonction d'agrégation. Contrairement à une fonction d'agrégation classique, une fonction de fenêtre calcule une valeur sur un groupe de lignes (une fenêtre) et renvoie un résultat distinct pour chaque ligne. Généralement, une fonction de fenêtre inclut une clause OVER qui divise les lignes en partitions. La fonction calcule ensuite un résultat pour chaque ligne en fonction de la fenêtre de lignes à laquelle elle appartient.

L'exemple suivant utilise la fonction MY_WINDOW_SUM_INT. Contrairement à la fonction intégrée SUM, qui renvoie une valeur BIGINT, MY_WINDOW_SUM_INT accepte un argument INT et renvoie 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];
        }
    }
}

Étant donné qu'une fonction de fenêtre est un type spécial de fonction d'agrégation, votre classe personnalisée doit implémenter toutes les méthodes UDAF requises, ainsi que la méthode windowUpdate().

Développer une UDTF

Une UDTF prend une seule ligne en entrée et produit une table de lignes en sortie. Les UDTF sont souvent utilisées pour des opérations telles que la division d'une colonne en plusieurs lignes.

L'exemple suivant utilise la fonction MY_UDF_SPLIT. Elle divise une chaîne en utilisant un espace comme séparateur. L'argument d'entrée et la valeur de retour sont tous deux de type STRING.

package com.starrocks.udf.sample;

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

Votre classe personnalisée doit implémenter la méthode suivante.

mvn package

Syntaxe

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

Paramètres

Créer une scalar UDF

Exécutez la commande suivante pour créer la scalar UDF de l'exemple précédent dans 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"
);

Créer une UDAF

Exécutez la commande suivante pour créer la UDAF de l'exemple précédent dans 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"
);

Les paramètres de la clause PROPERTIES sont identiques à ceux décrits dans la section Créer une scalar UDF.

Créer une UDWF

Exécutez la commande suivante pour créer la UDWF de l'exemple précédent dans 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 propriété analytic identifie la fonction comme une fonction de fenêtre. Pour une UDWF, définissez cette valeur sur true. Les autres paramètres sont identiques à ceux décrits dans la section Créer une scalar UDF.

Créer une UDTF

Exécutez la commande suivante pour créer la UDTF de l'exemple précédent dans 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"
);

Les paramètres de la clause PROPERTIES sont identiques à ceux décrits dans la section Créer une scalar UDF.

Utiliser la scalar UDF

Exécutez la commande suivante pour utiliser la scalar UDF créée à l'étape 6.

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

Utiliser la UDAF

Exécutez la commande suivante pour utiliser la UDAF créée à l'étape 6.

SELECT MY_SUM_INT(col1);

Utiliser la UDWF

Exécutez la commande suivante pour utiliser la UDWF créée à l'étape 6.

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

Utiliser la UDTF

Exécutez la commande suivante pour utiliser la UDTF de l'exemple précédent.

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