Flink AggregateFunction窗口函数,merge何时执行

Posted 2022-06-18 二十六画生的博客

前言
在我们使用Flink DataStream API编写业务代码时，aggregate()算子和AggregateFunction无疑是非常常用的。编写一个AggregateFunction需要实现4个方法：

public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable 
    ACC createAccumulator();
 
    ACC add(IN value, ACC accumulator);
 
    OUT getResult(ACC accumulator);
 
    ACC merge(ACC a, ACC b);

前三个方法都很容易理解，但第四个merge()方法就有些令人费解了：到底什么时候需要合并两个累加器的数据呢？最近也有童鞋问到了这个问题。实际上，这个方法是专门为会话窗口（session window）服务的。下面来解析一下会话窗口。

Session Window & MergingWindowAssigner

stream.keyBy("userId").window(EventTimeSessionWindows.withGap(Time.seconds(gap)))
在普通的翻滚窗口和滑动窗口中，窗口的范围是按时间区间固定的，虽然范围有可能重合，但是处理起来是各自独立的，并不会相互影响。但是会话窗口则不同，其范围是根据事件之间的时间差是否超过gap来确定的（超过gap就形成一个新窗口），也就是说并非固定。所以，我们需要在每个事件进入会话窗口算子时就为它分配一个初始窗口，起点是它本身所携带的时间戳（这里按event time处理），终点则是时间戳加上gap的偏移量。这样的话，如果两个事件所在的初始窗口没有相交，说明它们属于不同的会话；如果相交，则说明它们属于同一个会话，并且要把这两个初始窗口合并在一起，作为新的会话窗口。多个事件则依次类推，最终形成上面图示的情况。

为了支持会话窗口的合并，它们的WindowAssigner也有所不同，称为MergingWindowAssigner，如下类图所示。

MergingWindowAssigner是一个抽象类，代码很简单，定义了用于合并窗口的mergeWindows()方法以及合并窗口时的回调MergeCallback。

public abstract class MergingWindowAssigner<T, W extends Window> extends WindowAssigner<T, W> 
    private static final long serialVersionUID = 1L;
 
    public abstract void mergeWindows(Collection<W> windows, MergeCallback<W> callback);
 
    public interface MergeCallback<W> 
        void merge(Collection<W> toBeMerged, W mergeResult);
    

所有MergingWindowAssigner实现类的mergeWindows()方法都是相同的，即直接调用TimeWindow.mergeWindows()方法。