Realtime Compute for Apache Flink:Java

Flink admite los siguientes tres tipos de UDF.

Las UDF proporcionan un método opcional open(FunctionContext context). FunctionContext pasa elementos de configuración personalizados como parámetros. El proceso es el siguiente:

1. En la pestaña Configuration de la página O&M > Deployments de la consola de desarrollo de Flink, agregue el elemento de configuración pipeline.global-job-parameters en la sección Other Configuration de Parameters.

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

   FunctionContext#getJobParameter solo puede recuperar valores definidos en pipeline.global-job-parameters. Debe incluir todos los elementos de configuración que utiliza la UDF en este parámetro. A continuación se describe cómo configurar este elemento.

   Paso	Acción	Procedimiento	Ejemplo
   Paso 1	Defina pares clave-valor.	Separe la clave y el valor con dos puntos (:) y encierre cada par clave-valor entre comillas simples ('). Nota Si una clave o un valor contiene dos puntos (:), encierre dicha clave o valor entre comillas dobles ("). Si una clave o un valor contiene dos puntos (:) o comillas dobles ("), debe escaparlo con dos comillas dobles consecutivas ("").	Si key = k1 y value = {hi,hello}, defina el par como 'k1:{hi,hello}'. Si key = k2 y value = str:ing,str:ing, defina el par como 'k2:"str:ing,str:ing"' Si key = k3 y value = str"ing,str:ing, defina el par como 'k3:"str""ing,str:ing"'
   Paso 2	Formatee el valor final de pipeline.global-job-parameters como un archivo YAML.	Coloque cada par clave-valor en una nueva línea y sepárelos con comas (,). Nota Una cadena multilínea en un archivo YAML comienza con una barra vertical (|). Cada línea de una cadena multilínea en un archivo YAML debe tener la misma sangría.	pipeline.global-job-parameters: | 'k1:{hi,hello}', 'k2:"str:ing,str:ing"', 'k3:"str""ing,str:ing"'

2. En el código de la UDF, utilice FunctionContext#getJobParameter para recuperar el valor de cada elemento. El siguiente código es un ejemplo.

   Ejemplo:

   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;