Realtime Compute for Apache Flink:Java

Precautions

• UDFs may cause dependency conflicts in the packaged JAR file. Take note of the following items when you develop UDFs:

  • Make sure that the Realtime Compute for Apache Flink version you configure for the SQL deployment corresponds to the Apache Flink version you specify in the pom.xml file.

  • Specify <scope>provided</scope> for dependencies related to Apache Flink.

  • Use the Shade plug-in to package other third-party dependencies. For more information, see Apache Maven Shade plug-in.

  For information about how to resolve dependency conflicts, see Console operations

• Frequent calls of UDFs in a SQL deployment may cause a timeout. We recommend that you upload the JAR file of the UDF as an additional dependency of the deployment and use the CREATE TEMPORARY FUNCTION statement to declare the UDF. Example: CREATE TEMPORARY FUNCTION 'GetJson' AS 'com.soybean.function.udf.MyGetJsonObjectFunction';.

UDF types

Register a UDF

• For information about how to globally register a UDF, see Register a catalog UDF.

• For information about how to register UDF for a single deployment, see Register a deployment-level UDF.

Pass UDF parameters

You can configure the parameters of a UDF in the development console of Realtime Compute for Apache Flink and obtain the parameter values within the UDF.

Named parameters

In SQL, parameter values must be passed to a function in the order in which the parameters are defined and optional parameters cannot be omitted. If the number of parameters is large, ensuring the correct parameter order can be challenging. To ensure the correct parameter order and improve usability, you can use named parameters to specify only the parameters you want to use. The following example shows how to use named parameters. In this example, a UDSF named ScalarFunction is used.

// Implement a UDSF in which the second and third 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;
	}
}

When you call the function in SQL, you can specify only the first required parameter. You can also specify the second and third optional parameters. Sample code:

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;

References

• For information about how to use a UDAF to sort and aggregate data, see Use a UDAF to sort and aggregate data.

• For information about how to develop and use Java UDFs, see UDAFs, UDSFs, and UDTFs.

• For information about how to debug and optimize Python UDFs, see Overview.

• For information about how to develop and use Python UDFs, see UDAFs, UDSFs, and UDTFs.