Process Table Functions in Confluent Manager for Apache Flink

Starting with version 2.5.0, Confluent Manager for Apache Flink® (CMF) supports process table functions (PTFs) on compute pools that use Flink 2.3. A PTF is a user-defined function that takes a table as input and returns a table. It can transform each row on its own, or process the rows of each key together and keep state for every key, so you can write custom stream processing logic, such as per-key counters, in Java and call it from Flink SQL.

You package a PTF as a JAR, upload the JAR to artifact management, and register the function in an environment catalog with CREATE FUNCTION, the same way you register a user-defined function.

Prerequisites

A compute pool that runs Flink 2.3. PTFs are an Flink 2.1 feature, so compute pools that run Flink 1.19 cannot run them. For how to create a compute pool for Flink 2.3, see Flink Versions for Statements in Confluent Manager for Apache Flink.
The environment catalog and artifact management must both be enabled.
At least one compile worker. When the environment catalog and artifact management are both enabled, the Helm chart adds one compile worker by default. Do not set cmf.statementService.workers.count to 0. For details, see Compile workers.

Set the current catalog to the environment catalog, _env_<environment>, and the current database to default. Then submit the statements on this page to the Flink 2.3 compute pool.

The examples on this page use the clicks table in the examples catalog. To use the catalog, install CMF with cmf.sql.examples-catalog.enabled=true.

Row and set semantics

A PTF declares how Flink passes it each table argument:

Row semantics: The function processes each row on its own. Flink can send any row to any parallel instance of the function, so the function cannot keep state for a key. Use row semantics to filter, reshape, or expand rows.
Set semantics: The function processes the rows that share a key together. You set the key with PARTITION BY when you call the function, and the function can keep state for each key. Use set semantics for per-key logic, such as counting or deduplicating events.

Write a PTF

A PTF is a Java class that extends ProcessTableFunction and implements an eval method. Annotate the table argument with @ArgumentHint to choose row or set semantics, and give every argument a name, so that callers can pass arguments by name. The function emits output rows by calling collect.

The following function has row semantics. For each click that has a view time of at least min_view_time, it emits the click’s URL and view time:

package com.example;

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.ArgumentTrait;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.types.Row;

@DataTypeHint("ROW<url STRING, view_time INT>")
public class LongViews extends ProcessTableFunction<Row> {
    public void eval(
            @ArgumentHint(value = ArgumentTrait.ROW_SEMANTIC_TABLE, name = "input") Row input,
            @ArgumentHint(name = "min_view_time") int minViewTime) {
        Integer viewTime = input.getFieldAs("view_time");
        if (viewTime != null && viewTime >= minViewTime) {
            collect(Row.of(input.getFieldAs("url"), viewTime));
        }
    }
}

The following function has set semantics. It keeps a click count for each key in state that it declares with @StateHint, and emits each click’s URL with the count so far. A state class needs a public no-argument constructor and public fields:

package com.example;

import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.ArgumentTrait;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.StateHint;
import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.types.Row;

@DataTypeHint("ROW<url STRING, click_count BIGINT>")
public class CountClicks extends ProcessTableFunction<Row> {

    public static class ClickState {
        public long count = 0L;
    }

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

Compile the classes against org.apache.flink:flink-table-common version 2.3.0 as a provided dependency, target Java 17, and package them as a JAR:

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

For the complete PTF API, including timers, multiple table arguments, and updating input tables, see Process Table Functions in the Flink documentation.

Upload the JAR

Put the artifact metadata in a file, artifact.json:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Artifact",
  "metadata": {
    "name": "ptfs.jar"
  },
  "spec": {}
}

Upload the JAR, passing artifact.json in the artifact part:

curl -X POST http://cmf:8080/cmf/api/v1/environments/prod/artifacts \
  -F 'artifact=@artifact.json;type=application/json' \
  -F 'file=@target/ptfs.jar'

For artifact names, versions, and updates, see Manage Artifacts in Confluent Manager for Apache Flink.

Register a PTF

Register each PTF with CREATE FUNCTION, using the USING JAR clause to reference the uploaded artifact. The syntax and the cmf:// URI rules are the same as for user-defined functions.

confluent flink statement create create-long-views \
  --url http://cmf:8080 \
  --environment prod \
  --compute-pool pool-2-3 \
  --catalog _env_prod \
  --database default \
  --sql "CREATE FUNCTION long_views AS 'com.example.LongViews' USING JAR 'cmf://prod/ptfs.jar';"

Register the set-semantic function the same way, and then list the registered functions. Submit each of the following statements with the same options as the previous command:

CREATE FUNCTION count_clicks AS 'com.example.CountClicks' USING JAR 'cmf://prod/ptfs.jar';

SHOW USER FUNCTIONS;

Call a PTF

Call a PTF in the FROM clause of a query. Pass a table argument with the TABLE keyword, and pass each argument by name with =>.

A function with row semantics takes the table as it is:

SELECT * FROM long_views(input => TABLE examples.marketplace.clicks, min_view_time => 100);

The result has the columns that the function emits, url and view_time.

A function with set semantics requires a PARTITION BY clause on the table argument:

SELECT * FROM count_clicks(input => TABLE examples.marketplace.clicks PARTITION BY user_id);

The partition key columns come first in the result, followed by the columns that the function emits, so the result has the columns user_id, url, and click_count. Without PARTITION BY, the statement fails validation.

Drop a PTF

To drop a PTF, use DROP FUNCTION:

DROP FUNCTION [IF EXISTS] function_name;

Dropping a function does not affect jobs that are already running. Those jobs keep using the function version that they were compiled with.

Limitations

PTFs run only on compute pools that use Flink 2.3. Register and call them from statements on those compute pools.
The limitations of user-defined functions also apply to PTFs: they must be implemented in Java, ALTER FUNCTION and temporary functions are not supported, and function code can load only classes that match cmf.sql.classAllowlist.

Security considerations

Important

To compile a statement, a compile worker loads and runs the function class, and the Flink cluster runs it during the job. Treat a PTF JAR as trusted code. Review and scan function code before you upload it, restrict who can upload artifacts and submit statements with RBAC, and run PTF workloads on dedicated compute pools rather than shared ones. Secure the compile workers as described in Secure the compile workers. The class allowlist is a best-effort control, not a security boundary.