An aggregation function aggregates one or more rows of data into a single scalar value.

How to Implement #

1. Add the flink-table-common dependency to pom.xml. Set the version based on the Flink engine version selected for the task.

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

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

2. Code Implementation
   Extend org.apache.flink.table.functions.AggregateFunction and implement the following three methods:

   • createAccumulator(): creates an accumulator to store the aggregation state.

     tip

     When creating an accumulator, the comparator cannot be a lambda expression. Place the comparator in a class member variable because the constructed object cannot be serialized, which will cause the Flink task to fail.

   • accumulate(): called for each input row to update the accumulator.
   • getValue(): called after all rows have been processed to compute and return the final result.
     In addition to the three required methods above, the following methods only need to be implemented in certain cases.
   • retract(): required for bounded OVER windows.
   • merge(): required for many batch aggregations and session window aggregations.
   • resetAccumulator(): required for many batch aggregations.
   • getResultType(): required if the result type is not a primitive type or a simple POJO.
   • getAccumulatorType(): required if the accumulator type is not a primitive type or a simple POJO.

Example Demo #

This article provides a simple example of an Aggregation function that calculates sum and count.

/** * Accumulator for WeightedAvg. */
public static class WeightedAvgAccum {
    public long sum = 0;
    public int count = 0;
}

/** * Weighted Average user-defined aggregate function. */
public static class WeightedAvg extends AggregateFunction<Long, WeightedAvgAccum> {

    // Initialize the accumulator.
    @Override
    public WeightedAvgAccum createAccumulator() {
        return new WeightedAvgAccum();
    }

    // Get the result.
    @Override
    public Long getValue(WeightedAvgAccum acc) {
        if (acc.count == 0) {
            return null;
        } else {
            return acc.sum / acc.count;
        }
    }
    
    // Specifies the return value type. In this example, it returns a Long. You can omit this method, as Flink can infer it automatically.
    // This is added to demonstrate the method's usage.
    // 1. For basic or POJO types, such as Integer, Long, and String, you do not need to write this method.
    // 2. For complex types, such as array, map, and row, you must write this method.
    // 3. The getAccumulatorType method is similar, but it returns the Accumulator's type.
    @Override
        public TypeInformation<Row[]> getResultType() {
            // In this example, it is Long.
                return Types.LONG;
        
        // Here are a few examples:
        // 1. Map<String, Integer>
        // Types.MAP(Types.STRING, Types.INT)
        // 2. An array of objects, such as String[] or Object[]
        // Types.OBJECT_ARRAY(Types.STRING)
        // 3. An array of a primitive type, such as int[] or Integer[]
        // Types.PRIMITIVE_ARRAY(Types.INT);
        // 4. A Row type, for example, with two columns: name varchar, id int.
        // Types.ROW_NAMED(new String[]{"name", "id"}, new TypeInformation[]{Types.STRING, Types.INT})
        // 5. For more common types, see org.apache.flink.api.common.typeinfo.Types
        }

        // Accumulate a message
    public void accumulate(WeightedAvgAccum acc, long iValue, int iWeight) {
        acc.sum += iValue * iWeight;
        acc.count += iWeight;
    }

        // Retract a message
    public void retract(WeightedAvgAccum acc, long iValue, int iWeight) {
        acc.sum -= iValue * iWeight;
        acc.count -= iWeight;
    }
    
    // Merges multiple accumulators. For example, session windows use this.
    public void merge(WeightedAvgAccum acc, Iterable<WeightedAvgAccum> it) {
        Iterator<WeightedAvgAccum> iter = it.iterator();
        while (iter.hasNext()) {
            WeightedAvgAccum a = iter.next();
            acc.count += a.count;
            acc.sum += a.sum;
        }
    }
    
    // Resets the accumulator.
    public void resetAccumulator(WeightedAvgAccum acc) {
        acc.count = 0;
        acc.sum = 0L;
    }
}

How to Use #

To use a custom Aggregation function, see Create a Custom Function.