Flink SQL Functions (UDF / PTF)

Description

Flink user-defined functions written in Java or Scala can be used in scenarios without writing a Nussknacker component. Every configured function becomes its own node in the Component Palette:

• scalar functions (ScalarFunction, UDF) appear in the User Defined Functions group,
• process table functions (ProcessTableFunction, PTF) appear in the Process Table Functions group.

The node's parameters are generated automatically from the function's signature, described with Flink's @ArgumentHint / @DataTypeHint / @FunctionHint annotations, and the output type of the node comes from the function's declared result type.

The functions are not available inside the Flink SQL Query node - each one is used as a separate node.

Configuration

The functions are enabled by the flinkSqlFunctions component provider in the model configuration. The JAR with the functions has to be on the model classPath.

components.flinkSqlFunctions {
  # Auto-discovery: every concrete UserDefinedFunction found in these packages becomes a node
  scanPackages = ["com.example.flinkudf", "org.acme.analytics"]

  # Explicit list - may be combined with scanPackages or used alone
  functions = [
    { name = "Mask", class = "com.example.flinkudf.MaskFunction" }
    { class = "com.example.flinkudf.ThresholdAlertFunction" }
  ]
}

Option	Description
scanPackages	List of package prefixes. Every concrete (non-abstract) subclass of Flink's UserDefinedFunction found in these packages on the model classpath is added. Classes that cannot be used (e.g. without a fixed signature, see below) are skipped.
functions	Explicit list of functions: class - fully qualified class name; name - optional name of the node. An explicit entry overrides a scanned function with the same name. A function that cannot be loaded is skipped and a warning is logged.

When no function is found, no node is added and a warning is logged.

Node names

The name of the node is, in this order of precedence:

1. the value of a @ComponentName("...") annotation on the function class - any runtime-retained annotation with the simple name ComponentName and a String value() method works, so the function JAR needs no Nussknacker dependency,
2. the name from the functions entry,
3. derived from the class name: CamelCase is split into words and a trailing Function is dropped - ThresholdAlertFunction becomes Threshold Alert.

Requirements for the functions

• The class must have a public no-argument constructor.
• It must have exactly one eval method - not overloaded and not var-args - whose arguments are described by @ArgumentHint (or @FunctionHint), so that Flink can derive a fixed signature. The names given in @ArgumentHint(name = ...) become the names of the node's parameters.

Scalar functions (User Defined Functions)

Each argument of eval becomes a parameter of the node, evaluated for every record, so it can use #input and the other variables. Optional arguments are not mandatory. The function's eval method is invoked directly for each record and its result is stored in the output variable. The node keeps the context: #input and the other upstream variables remain available after the node.

Argument types are mapped to their Nussknacker counterparts: an ARRAY argument takes a list (e.g. {'a', 'b'}) and a ROW argument takes a record (e.g. {name: 'n', scores: {7, 8}}).

For example, for the function:

public class MaskFunction extends ScalarFunction {
    public @DataTypeHint("STRING") String eval(
            @ArgumentHint(name = "value", type = @DataTypeHint("STRING")) String value,
            @ArgumentHint(name = "visibleSuffix", type = @DataTypeHint("INT")) Integer visibleSuffix) {
        ...
    }
}

the Mask node has the parameters value (e.g. #input.cardNumber) and visibleSuffix (e.g. 4), and its output variable is a String.

Process Table Functions

For a PTF the records entering the node are passed as the function's table argument - a table whose columns are the variables available before the node, plus record_time with the record's timestamp (as in the Flink SQL Query node). The other (scalar) arguments become parameters of the node; their values are constants fixed at design time and must be simple literals (string, number, boolean).

Parameter	Description
function arguments	One parameter per scalar argument of eval, named after the argument.
Partition by (column)	Shown when the table argument has set semantics (ArgumentTrait.SET_SEMANTIC_TABLE). A dropdown of the upstream columns, including fields nested in records (e.g. input.userId) and record_time. The stream is partitioned by this column before the function is applied; each partition is processed independently and keeps its own state.

Each row emitted by the PTF (collect(...)) becomes a new record, and the row is available in the output variable. With Partition by (column) set, Flink puts the partition column in front of the columns emitted by the function. As with Flink SQL Query, the node clears all upstream variables - #input is not available after a PTF node.

For example, a PTF counting records per partition:

@DataTypeHint("ROW<count BIGINT>")
public class RunningCountFunction extends ProcessTableFunction<Row> {
    public static class CountState { public long count = 0L; }

    public void eval(
            @StateHint CountState state,
            @ArgumentHint(value = ArgumentTrait.SET_SEMANTIC_TABLE, name = "input") Row input) {
        state.count++;
        collect(Row.of(state.count));
    }
}

is available as the Running Count node with the Partition by (column) parameter, e.g. set to input.userId.

Additional considerations

• PTFs currently process their input as append-only: a changelog (CDC) input is not supported, and a #cdcOperation variable is passed to the function as an ordinary column.
• Errors thrown by a scalar function for a record are handled by the scenario's exception handler, like errors of other nodes.