Realtime Compute for Apache Flink:Java

Flink prend en charge les trois types d'UDF suivants.

Les UDF proposent une méthode optionnelle open(FunctionContext context). FunctionContext transmet les éléments de configuration personnalisés sous forme de paramètres. Le processus est le suivant :

1. Dans l'onglet Configuration de la page O&M > Deployments de la console de développement Flink, ajoutez l'élément de configuration pipeline.global-job-parameters dans la section Other Configuration de Parameters.

   pipeline.global-job-parameters: | 
     'k1:{hi,hello}',
     'k2:"str:ing,str:ing"',
     'k3:"str""ing,str:ing"'

   FunctionContext#getJobParameter ne peut récupérer que les valeurs définies dans pipeline.global-job-parameters. Vous devez inclure tous les éléments de configuration utilisés par l'UDF dans ce paramètre. Les étapes suivantes décrivent comment configurer cet élément.

   Étape	Action	Procédure	Exemple
   Étape 1	Définissez des paires clé-valeur.	Séparez la clé et la valeur par deux-points (:) et placez chaque paire clé-valeur entre guillemets simples ('). Remarque Si une clé ou une valeur contient un deux-points (:), placez-la entre guillemets doubles ("). Si une clé ou une valeur contient un deux-points (:) ou un guillemet double ("), vous devez l'échapper avec deux guillemets doubles consécutifs ("").	Si key = k1 et value = {hi,hello}, définissez la paire comme 'k1:{hi,hello}'. Si key = k2 et value = str:ing,str:ing, définissez la paire comme 'k2:"str:ing,str:ing"' Si key = k3 et value = str"ing,str:ing, définissez la paire comme 'k3:"str""ing,str:ing"'
   Étape 2	Formatez la valeur finale de pipeline.global-job-parameters sous forme de fichier YAML.	Placez chaque paire clé-valeur sur une nouvelle ligne et séparez-les par des virgules (,). Remarque Une chaîne multiligne dans un fichier YAML commence par une barre verticale (|). Chaque ligne d'une chaîne multiligne dans un fichier YAML doit avoir la même indentation.	pipeline.global-job-parameters: | 'k1:{hi,hello}', 'k2:"str:ing,str:ing"', 'k3:"str""ing,str:ing"'

2. Dans le code de l'UDF, utilisez FunctionContext#getJobParameter pour récupérer la valeur de chaque élément. L'extrait de code suivant illustre cette approche.

   Exemple :

   context.getJobParameter("k1", null); // Returns the string {hi,hello}.
   context.getJobParameter("k2", null); // Returns the string str:ing,str:ing.
   context.getJobParameter("k3", null); // Returns the string str"ing,str:ing.
   context.getJobParameter("pipeline.global-job-parameters", null); // Returns null. You can only get the content defined in pipeline.global-job-parameters, not any other job configuration item.

// Implement a user-defined scalar function. The last two input parameters are optional (isOptional = true).
public class MyFuncWithNamedArgs extends ScalarFunction  {
	private static final long serialVersionUID = 1L;

	public String eval(@ArgumentHint(name = "f1", isOptional = false, type = @DataTypeHint("STRING")) String f1,
			@ArgumentHint(name = "f2", isOptional = true, type = @DataTypeHint("INT")) Integer i2,
			@ArgumentHint(name = "f3", isOptional = true, type = @DataTypeHint("LONG")) Long l3) {

		if (i2 != null) {
			return "i2#" + i2;
		}
		if (l3 != null) {
			return "l3#" + l3;
		}
		return "default#" + f1;
	}
}

CREATE TEMPORARY FUNCTION MyNamedUdf AS 'com.aliyun.example.MyFuncWithNamedArgs';

CREATE temporary TABLE s1 (
    a INT,
    b BIGINT,
    c VARCHAR,
    d VARCHAR,
    PRIMARY KEY(a) NOT ENFORCED
) WITH (
    'connector' = 'datagen',
    'rows-per-second'='1'
);

CREATE temporary TABLE sink (
    a INT,
    b VARCHAR,
    c VARCHAR,
    d VARCHAR
) WITH (
    'connector' = 'print'
);

INSERT INTO sink
SELECT a,
    -- Specify only the first required parameter
    MyNamedUdf(f1 => c) arg1_res,
    -- Specify the first required parameter and the second optional parameter
    MyNamedUdf(f1 => c, f2 => a) arg2_res,
    -- Specify the first required parameter and the third optional parameter
    MyNamedUdf(f1 => c, f3 => d) arg3_res
FROM s1;