When developing SQL tasks, you can create a custom function if the system's built-in functions do not meet your business needs.

Background Information #

UDF stands for User-Defined Function.
The system provides built-in string manipulation functions like SUBSTRING and REPLACE, and aggregation functions like MAX, MIN, and AVG. However, if these functions do not meet your business needs, you can write your own code to create a custom function.

Prerequisites #

Before creating a UDF, you must write the code and build a JAR package.

• Developers (Project_Dev) typically manage UDFs. Ensure the operator has been added as a project member and assigned a role. See Add project members.
• The file name of the JAR package for the UDF must comply with console requirements and can only contain uppercase and lowercase letters, numbers, underscores (_), hyphens (-), and periods (.).

Writing UDF Code #

POM Dependencies #

First, add the required Flink dependencies to your Maven project. For version, enter the version that corresponds to the Flink engine selected for the task.

1. For Flink 1.16, we recommend using version 1.16.3.
2. For Flink 1.17, we recommend using version 1.17.2:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>

Scalar Function #

ScalarFunction (scalar function): maps zero, one, or more values to a new value.

import org.apache.flink.table.functions.ScalarFunction;

public class MyScalarFunction extends ScalarFunction {
    
    // You must implement the eval method.
    public String eval(String input) {
        if (input == null) {
            return null;
        }
        return input.toUpperCase();
    }
    
    // You can overload the eval method.
    public Integer eval(Integer a, Integer b) {
        return a + b;
    }
}

Table Function #

TableFunction (table function): returns zero, one, or more rows as output.

import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

public class MyTableFunction extends TableFunction<Row> {
    
    public void eval(String str) {
        if (str == null || str.isEmpty()) {
            return;
        }
        
        for (String s : str.split(",")) {
            Row row = new Row(2);
            row.setField(0, s);
            row.setField(1, s.length());
            collect(row);
        }
    }
}

Aggregation Function #

AggregateFunction (aggregation function): aggregates data from multiple rows.

import org.apache.flink.table.functions.AggregateFunction;

public class MyAggregateFunction extends AggregateFunction<Double, MyAccumulator> {
    
    // Create an accumulator.
    @Override
    public MyAccumulator createAccumulator() {
        return new MyAccumulator();
    }
    
    // Get the final result.
    @Override
    public Double getValue(MyAccumulator accumulator) {
        if (accumulator.count == 0) {
            return null;
        }
        return accumulator.sum / accumulator.count;
    }
    
    // Accumulation method
    public void accumulate(MyAccumulator accumulator, Double value) {
        if (value != null) {
            accumulator.sum += value;
            accumulator.count++;
        }
    }
    
    // Optional retract method (for bounded OVER windows)
    public void retract(MyAccumulator accumulator, Double value) {
        if (value != null) {
            accumulator.sum -= value;
            accumulator.count--;
        }
    }
    
    // Optional merge method (for session windows and group aggregations)
    public void merge(MyAccumulator accumulator, Iterable<MyAccumulator> iterable) {
        for (MyAccumulator other : iterable) {
            accumulator.sum += other.sum;
            accumulator.count += other.count;
        }
    }
    
    // Optional reset method
    public void resetAccumulator(MyAccumulator accumulator) {
        accumulator.sum = 0.0;
        accumulator.count = 0L;
    }
    
    // Custom accumulator class
    public static class MyAccumulator {
        public double sum = 0.0;
        public long count = 0L;
    }
}

Register and use UDFs #

Compile the UDF JAR package and upload dependency files #

In Job Development, upload and select the UDF JAR package under Dependency Files in the right-side parameter configuration sidebar.

Register the UDF in the SQL client #

In SQL, register and reference UDFs as follows:

-- Register function
CREATE FUNCTION my_scalar_func AS 'com.example.udf.MyScalarFunction';

-- Use function
SELECT my_scalar_func(name) FROM source_table;