Realtime Compute for Apache Flink:Java UDAFs

Prerequisites

Before you begin, make sure you have:

• IntelliJ IDEA installed

• Maven installed

• Java development environment configured

Steps

1. Download and decompress ASI_UDX_Demo to your local machine. The decompressed folder ASI_UDX-main contains:

   • pom.xml — Maven project configuration, including coordinates, dependencies, and build rules.

   • \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java — sample UDAF implementation.

2. In IntelliJ IDEA, select File > Open, then select the ASI_UDX-main folder.

3. Open pom.xml in the \ASI_UDX-main\ directory and configure dependencies. The file includes the minimum dependency for Flink 1.12:

   • If your job has no additional dependencies, skip to the next step.

   • If your job requires additional dependencies, add them to pom.xml.

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

   Use the latest minor version for the Apache Flink major version that matches your Ververica Runtime (VVR) version. For VVR-to-Flink version mappings, see Overview.

4. Open \ASI_UDX-main\src\main\java\ASI_UDAF\ASI_UDAF.java and implement your aggregation logic. The sample implements cumulative summation:

   package ASI_UDAF;
   
   import org.apache.flink.table.functions.AggregateFunction;
   
   import java.util.Iterator;
   
   public class ASI_UDAF{
       public static class AccSum{
           public long sum;
       }
   
       public static class MySum extends AggregateFunction<Long, AccSum>{
   
           @Override
           public Long getValue(AccSum acSum){
               return acSum.sum;
           }
   
           @Override
           public AccSum createAccumulator(){
               AccSum acCount= new AccSum();
               acCount.sum=0;
               return acCount;
           }
   
           public void accumulate(AccSum acc,long num){
               acc.sum += num;
           }
   
           /**
           * Supports retracting a message generated by an upstream operator.
           */
           public void retract(AccSum acc,long num){
               acc.sum -= num;
           }
   
           /**
           * Supports local-global two-stage aggregate optimization.
           */
           public void merge(AccSum acc,Iterable<AccSum> it){
               Iterator<AccSum> iter=it.iterator();
               while(iter.hasNext()){
                   AccSum accSum=iter.next();
                   if(null!=accSum){
                       acc.sum+=accSum.sum;
                   }
               }
           }
       }
   }

5. From the directory containing pom.xml, run:

   mvn package -Dcheckstyle.skip

   The UDAF is packaged successfully when ASI_UDX-1.0-SNAPSHOT.jar appears in \ASI_UDX-main\target\.

Method 1: Use a registered UDAF (recommended)

Register the UDAF once, then reuse it across deployments. For registration steps, see Manage UDFs.

Once registered as ASI_UDAF$MySum, call it directly in your SQL:

CREATE TEMPORARY TABLE ASI_UDAF_Source (
  a BIGINT NOT NULL
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE ASI_UDAF_Sink (
  sum  BIGINT
) WITH (
  'connector' = 'print'
);

INSERT INTO ASI_UDAF_Sink
SELECT `ASI_UDAF$MySum`(a)
FROM ASI_UDAF_Source;

Method 2: Upload a JAR to a specific deployment

On the Flink Data Studio > ETL page, upload the JAR package using Additional dependency files under More configurations. Then define a temporary function alias in the job's SQL.

The JAR is scoped to that deployment only and cannot be shared with other deployments.

If the temporary function is named mysum:

CREATE TEMPORARY TABLE ASI_UDAF_Source (
  a   BIGINT
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE ASI_UDAF_Sink (
  sum  BIGINT
) WITH (
  'connector' = 'print'
);

CREATE TEMPORARY FUNCTION `mysum` AS 'ASI_UDAF.ASI_UDAF$MySum'; -- Create the temporary function mysum.

INSERT INTO ASI_UDAF_Sink
SELECT `mysum`(a)
FROM ASI_UDAF_Source;