介绍

聚合函数，many to one，进来多个数据，出去一个结果，是 row 级别操作，如：聚合函数 avg、sum、min

使用

1. 定义 accumulator （累加器）

定义 accumulator ，存储了聚合的中间结果

就是普通的类，有成员变量

2. 定义函数类

定义函数类，继承 org.apache.flink.table.functions.AggregateFunction 类
必须实现下面方法：
- 覆盖方法 createAccumulator()：创建一个空的 accumulator
- 创建方法 accumulate()：对于每一行数据，会调用该方法。该方法一般有 2 个形参：
  - acc：累加器
  - 输入数据：sql中传入的数据。该形参 不能是基本数据类型，如：int、double 等，否则报错
- 覆盖方法 getValue()：当所有的数据都处理完后，通过调用该方法 计算、返回最终结果

2. 注册函数

stEnv.createTemporarySystemFunction("函数名",函数类.class);

stEnv.createTemporarySystemFunction("函数名",new 函数类());

3. 调用函数

在 SQL 中调用该函数，类似：max() 函数

案例

求平均分

数据文件

在工程的 data 目录下创建 student.csv，内容如下：

李雷,90
韩梅梅,82
lucy,76
lili,63
张三,39

acc类

public static class Acc{
    public long sum = 0;
    public int count = 0;
}

函数类

public static class AvgFun extends AggregateFunction<Double, Acc> {
    @Override
    public Double getValue(Acc acc) {
        if (acc.count == 0) {
            return null;
        } else {
            double d = ((double) acc.sum) / acc.count;
            BigDecimal decimal = new BigDecimal(String.valueOf(d));
            double res = decimal.setScale(num, BigDecimal.ROUND_HALF_UP).doubleValue();
            return res;
        }
    }
    @Override
    public Acc createAccumulator() {
        return new Acc();
    }
    Integer num = 0;

    /**
     *
     * @param acc 累加器
     * @param score 传入的列值
     * @param num 小数点后保留的位数，四舍五入
     */
    public void accumulate(Acc acc, Integer score,Integer num) {
        acc.sum += score;
        acc.count ++;
        this.num = num;
    }
}

主类

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;

public class StdAgg {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 要设置1个并行度，便于观察结果
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断

        //构建Table环境
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, bsSettings);

        /*
        创建表
        注意：表名不要出现关键字，如：order等，否则会报错
        要读取 csv文件，需要依赖flink-csv
         */
        TableResult table=stEnv.executeSql("CREATE TABLE t_student (" +
                "  name STRING," +
                "  score int" +
                ") WITH ( " +
                "  'connector' = 'filesystem'," + // 连接器，表示文件系统
                "  'path' = 'file:///D:\\bigdata_std\\flink_sql\\data\\student.csv'," + // 路径指定到文件夹 或 文件都可以
                "  'format' = 'csv'," + // 文件格式
                "  'csv.field-delimiter' = ','" + // csv列之间间隔符号
                ")");

        // 该方法废弃
//        stEnv.registerFunction("scoreFun",new ScoreFun());
        /*
        创建临时系统函数，与 createTemporaryFunction() 相比，可以直接调用
         注意：eval() 方法形参必须是对象类型，不能是 int、long这类基本数据类型
         */
        stEnv.createTemporarySystemFunction("AvgFun",AvgFun.class);
//        stEnv.createTemporarySystemFunction("AvgFun",new AvgFun());
        // 根据分数显示优良中差
        Table res=stEnv.sqlQuery("select AvgFun(score,3) from t_student ");
        System.out.println("Schema信息:");
        //打印Schema
        res.printSchema();

        //将计算后的数据，append到ds
        DataStream<Tuple2<Boolean, Row>> dsRes2 = stEnv.toRetractStream(res, Row.class);
        dsRes2.print("---");

        env.execute();
    }
}

参考：
https://blog.csdn.net/qq_22973811/article/details/124032836
https://blog.csdn.net/qq_21383435/article/details/106698462

原文出处：/show_1IX4NqjCbfa9.html