Creating a Function

A Function is custom code that is used to extend the processing capabilities of DeltaStream. There are already a number of built-in functions available for users to use in their queries. Users can add more functions by implementing them and adding them to DeltaStream.

This tutorial will demonstrate how to write a Function, add it to DeltaStream, and use it in DeltaStream queries.

Functions are only valid for approved Organizations. Please reach out to us if you want to enable Functions in your Organization.

Writing a Function

DeltaStream queries are powered by Apache Flink. So, to write Functions, users will leverage Flink's APIs (Flink documentation found here). Currently, DeltaStream supports Flink's Scalar Functions and Aggregate Functions.

To assist in writing these Functions, we've provided this repository that includes examples and templates. Users can fork or copy the template to get started. Note that in order to write Functions, we require:

Java (preferably Java 17)
Maven

For this tutorial, let's assume we have a simple use case where we want to be able to reverse a String so that we have the following behavior:

Query: SELECT reverse(col1) AS res FROM my_source_stream;
Input: { "col1": "ReverseMe" }
Output: { "res": "eMesreveR" }

Under the src/main/java directory, we can add a new class called MyStringReverser (this class can also exist in other packages under src/main/java, i.e. src/main/java/x/y/z/MyStringReverser.java) .

Here's what the MyStringReverser implementation may look like:

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

public class MyStringReverser extends ScalarFunction {

    @Override
    public void open(FunctionContext context) {
        // Initialization logic
    }

    public String eval(String s) {
        // Processing logic
        return new StringBuilder(s).reverse().toString();
    }
}

In our example, our MyStringReverser class extends ScalarFunction, the API used to define a Scalar Function typed UDF. For our Function, since we have a one-to-one mapping of inputs to outputs (i.e. for every input String we output a single reversed version of that String), a Scalar Function is appropriate. If we wanted to define an Aggregate Function instead, our class would extend Flink's AggregateFunction class instead (an example can be found here).

Important Components for Functions that `extends ScalarFunction`:

Function open(FunctionContext context): Any initialization logic exists in the open() method. This method is also where metrics can be added, using the FunctionContext (docs on adding custom metrics).
Function eval(T val): For Functions that extend ScalarFunction, they must have an eval() method. This method is where the processing logic lies. The method signature should match the input and output types of your Function. In our case, the input and output values are both Strings, but Functions can have multiple input values with different types, and the output type does not need to match the input type (this example shows how a Function can be defined for Struct input and output types). Furthermore, the same Function can have multiple eval() methods defined.

Important Components for Functions that `extends AggregateFunction`:

Function open(FunctionContext context): Any initialization logic exists in the open() method. This method is also where metrics can be added, using the FunctionContext (docs on adding custom metrics).
Function createAccumulator(): This method creates and initializes the accumulator for this Function. The accumulator is an intermediate data structure that stores the aggregated values until a final aggregation result is computed.
Function getValue(MyAccumulator acc): This method is called every time an aggregation result should be materialized. The returned value could be either an early and incomplete result (periodically emitted as data arrives) or the final result of the aggregation.
Function accumulate(MyAccumulator acc, T val): This method is called on each input Row and contains the logic for updating the accumulator. The method signature should match the input types of your Function, with the accumulator being the first parameter. The Function can have multiple accumulate() methods defined with different custom types and arguments.
(Optional) Function retract(MyAccumulator acc, T val): This method must be implemented when used in queries with OVER windows. The logic in this method should retract the input values from the accumulator instance. The Function can have multiple retract() methods defined with different custom types and arguments.
(Optional) Function merge(MyAccumulator acc, Iterable<MyAccumulator> it): This method must be implemented when used in queries with SESSION windows and bounded aggregations. The logic in this method should merge a group of accumulator instances into a single accumulator instance.

After writing your Function, build a .jar file. If you used our template, there are instructions on how to build the .jar file using Maven.