一、目标

数据源为Kafka ，通过Flink 时间窗口AggregateFunction方法来进行特定窗口内消息事件的次数和累计值。

本例中：使用事件时间（Event Time）、窗口为翻滚窗口（TumblingEventTimeWindows）大小为5s、聚合函数计算特定key对应的窗口内的事件次数和总的金额。

主要内容：

本例中事件类SimpleEvent介绍

Event Time和Watermark 指定和创建

Window窗口AggregateFunction 实现消息事件的次数和累计值

二、 Event 事件类介绍

例子中Event 为自定义SimpleEvent 事件类消息事件的次数和累计值

public class SimpleEvent implements Serializable {
    
    private String ID_NO;
    private BigDecimal AMT;
    private long CREATE_TIMESTAMP;

    public SimpleEvent(String ID_NO, BigDecimal AMT) {

        this.CREATE_TIMESTAMP = System.currentTimeMillis();
        this.ID_NO = ID_NO;
        this.AMT = AMT;

    }

    public String getID_NO() {
        return ID_NO;
    }

    public BigDecimal getAMT() {
        return AMT;
    }

    public long getCreationTime() {
        return this.CREATE_TIMESTAMP;
    }
}

三、Event Time事件时间处理

需要设置env.setStreamTimeCharacteristic

      env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

由于事件本身带有创建的时间戳，所以直接使用SimpleEvent自带的时间戳作为EventTime ，同时设置watermarks方式为不迟于当前最大EventTime 固定时间。主要参考Flink官方文档示例实现。

 private static class MyTimestampsAndWatermarks implements AssignerWithPeriodicWatermarks<SimpleEvent>{

        private final long maxOutOfOrderness = 3500; // 3.5 seconds

        private long currentMaxTimestamp;

        @Nullable
        @Override
        public Watermark getCurrentWatermark() {
            return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
        }

        @Override
        public long extractTimestamp(SimpleEvent element, long previousElementTimestamp) {
            long timestamp = element.getCreationTime();
            currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
            return timestamp;
        }
    }

四、Window窗口AggregateFunction 实现消息事件的次数和累计值

AggregateFunction官方接口

public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable

对应三个参数

 @param <IN>  The type of the values that are aggregated (input values) 可以理解为输入流数据类型，例子中为SimpleEvent
 @param <ACC> The type of the accumulator (intermediate aggregate state). accumulator累加器的类别，本例中为一个复合类，包括key，count，sum分别对应ID_NO,事件次数，时间累计值（总金额）
 @param <OUT> The type of the aggregated result 聚合结果类别

本例中实现的AverageAggregate类：

    private static class AverageAggregate implements AggregateFunction<SimpleEvent, AverageAccumulator, AverageAccumulator> {

        @Override
        public AverageAccumulator createAccumulator() {
            return new AverageAccumulator();
        }

        @Override
        public AverageAccumulator add(SimpleEvent value, AverageAccumulator accumulator) {
            if (accumulator.key.isEmpty()){
                accumulator.key = value.getID_NO();
            }
            accumulator.count +=1;
            accumulator.sum =accumulator.sum.add(value.getAMT());
            return accumulator;
        }

        @Override
        public AverageAccumulator getResult(AverageAccumulator accumulator) {
            //return Long.valueOf(accumulator.sum.toString())/(double)accumulator.count;
            return accumulator;
        }

        @Override
        public AverageAccumulator merge(AverageAccumulator a, AverageAccumulator b) {
            a.count = a.count+b.count;
            a.sum = a.sum.add(b.sum);
            return a;
        }
        
    }

五、主要代码

        // stream 创建 timestamp assigner  和  watermark 机制
        DataStream<SimpleEvent> withTimestampsAndWatermarks = sourceStream.assignTimestampsAndWatermarks(new MyTimestampsAndWatermarks());

        DataStream<AverageAccumulator>  averageAccumulatorStream =  withTimestampsAndWatermarks
                .keyBy(SimpleEvent::getID_NO)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .aggregate(new AverageAggregate());

        averageAccumulatorStream.map(new MapFunction<AverageAccumulator, String>() {
            @Override
            public String map(AverageAccumulator value) throws Exception {
                System.out.println("TEMP RESULT :"+value.key +" ,"+value.count+ " , "+ value.sum );
                return "";
            }
        });

自定义AverageAccumulator 类：

public class AverageAccumulator {
    String key ;
    long count;
    BigDecimal sum;
    long createTime;
    long updateTime;

    AverageAccumulator(){
        key  = "" ;
        count = 0L ;
        sum = new BigDecimal(0);
        createTime=0L;
        updateTime=0L;
    }

}

结果示例：

TEMP RESULT :525992 ,1 , 350
TEMP RESULT :525997 ,5 , 2150
TEMP RESULT :525996 ,2 , 1000
TEMP RESULT :525992 ,2 , 1600
TEMP RESULT :525995 ,2 , 700
TEMP RESULT :525991 ,4 , 1900
TEMP RESULT :525994 ,2 , 1500

完整示例，将在整理后，分享到github。