package com.hao.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

public class UdfTest_AggregateFunction {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //1.在创建表的DDL中直接定义时间属性
        String creatDDL = "CREATE TABLE clickTable (" +
                "user_name STRING," +
                "url STRING," +
                "ts BIGINT," +
                "et AS TO_TIMESTAMP( FROM_UNIXTIME(ts / 1000))," + //事件时间  FROM_UNIXTIME() 能转换为年月日时分秒这样的格式 转换秒
                " WATERMARK FOR et AS et - INTERVAL '1' SECOND " + //watermark 延迟一秒
                ")WITH(" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/clicks.txt'," +
                " 'format' = 'csv'" +
                ")";

        tableEnv.executeSql(creatDDL);

        //2.注册自定义聚合函数
        tableEnv.createTemporarySystemFunction("WeightedAvg", WeightedAverage.class);

        //3.调用UDF进行查询转换
        Table resultTable = tableEnv.sqlQuery("select user_name,WeightedAvg(ts,1) as w_avg from clickTable group by user_name");

        //4.转换成流打印
        tableEnv.toChangelogStream(resultTable).print();


        env.execute();
    }

    //单独定义累加器类型
    public static class WeightedAvgAccumulator{
        public long sum = 0;
        public int count = 0;
    }

    //实现自定义的聚合函数，计算平均时间戳
    public static class WeightedAverage extends AggregateFunction<Long,WeightedAvgAccumulator>{

        @Override
        public Long getValue(WeightedAvgAccumulator accumulator) {
            if (accumulator.count == 0){
                return null;
            }else{
                return accumulator.sum / accumulator.count;
            }

        }

        @Override
        public WeightedAvgAccumulator createAccumulator() {
            //初始化 累加器
            return new WeightedAvgAccumulator();
        }

        //累加计算的方法 (第一个是 WeightedAvgAccum 类型的累加器；
        //另外两个则是函数调用时输入的字段：要计算的值 iValue 和 对应的权重 iWeight。)
        public void accumulate(WeightedAvgAccumulator accumulator,Long iValue,Integer iWeight){
            accumulator.sum += iValue * iWeight;
            accumulator.count += iWeight;
        }
    }
}