flink 时间窗口

flink强大的窗口功能，是相较于其他流计算引擎比较有优势的地方。
flink中窗口是如何设计的？
一共有四个要素。
1）window assigner。对于流式的数据，如何来界定一个window，也就是把数据划分给正确的窗口。
2） 窗口触发器，trigger。提供灵活的cut point ，满足不同触发逻辑的需求。
3） 窗口Incepter，evict。提供灵活的cut point，在串口计算前后可以做一些定制化的事情。
4） 窗口处理行数，processer。提供强大的处理接口，可以对窗口中的数据进行运算。

1）首先看window assigner.
通常我们想把数据按照一定区间做一个划分，比如10分钟。每一个数据有一个 event_time和process_time的时间概念。
process_time 位系统处理这个数据的时间，而event——time则是用户可以自己定义的和业务逻辑相关的时间。
比如，处理一个股票trade流， trade 的交易所发生时间可以作为event——time，flink 处理这个流的时间位process——time。
对于使用event_time的流数据，因为网络延迟的原因，通常会存在乱序，比如我定义一个【00：00，00:10】的窗口
那么有些00：09 的数据，可能会在00：11的时候到达，flink 的窗口机制，必须要给与一定的容错性来处理这样的乱序问题。
flink采用一种watermark水位线的机制，watermark作为流数据中的特殊事件存在。对于watermark，通常使用periodical 的机制生成，
这样可以提高效率，下面举个例子说明具体的算法。
容忍的延迟位1分钟
event_time, watermark
00:01 00:01-1=00:00
00:05 00:05-1 =00:04
00:02 00:04(取当前最大)
00:03 00:04(取当前最大)
00:06 00:05-1 =00:05
00:10 00:10-1 =00:09
00:11 00:11-1 =00:10
因为watermark到达 窗口边界，这个时候就要触发窗口计算了。后面来的数据要么被丢弃，
要么就是进入用户自定义的output中。
对于多并行度的流，每个子并行流都需要处理窗口的问题，达到触发提交进行窗口闭合。

2）trigger，一般默认的窗口触发是 watermark 和窗口的 endTime 相等的时候，
flink 提供灵活的自定义方式，用户可以任意使用窗口触发机制来满足自己的需求。
需要对窗口制定一个trigger，如下的CountTriggerDebug，当窗口中的数据达到2条就触发

 DataStream<Tuple2<String, List<Integer>>> reduce = map.keyBy("f0")
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .trigger(CountTriggerDebug.of(2))
                .reduce((new ReduceFunction<Tuple2<String, List<Integer>>>() {
                    @Override
                    public Tuple2<String, List<Integer>> reduce(Tuple2<String, List<Integer>> value1, Tuple2<String, List<Integer>> value2) {
                        value1.f1.add(value2.f1.get(0));
                        return value1;
                    }
                }));

    @Override
    public TriggerResult onElement(Object element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
        ReducingState<Long> count = ctx.getPartitionedState(stateDesc);
        count.add(1L);
        if (count.get() >= maxCount) {
            System.out.println("触发器触发窗口函数对该窗口计算,同时清除该窗口的计数状态,--" + count.get());
            count.clear();
//            return TriggerResult.FIRE;
            return TriggerResult.FIRE_AND_PURGE;
        }
        System.out.println("触发器仅对该窗口的计数状态进行加一操作--" + count.get());
        return TriggerResult.CONTINUE;
    }

使用trigger的时候，需要关注状态 是使用 FIRE_AND_PURGE 还是 FIRE。
如果只是FIRE的话，窗口中所有数据会传到下游。如果使用FIRE_PURGE,则下游只会收到增量数据，不至于影响性能。

3）Evictor
Evictor机制提供给在窗口计算前和计算后做一些事情的能力。比如删除一些特定的数据等。
指定一个实现了Evictor接口的类就可以达到这个效果。

        DataStream<Tuple2<String, List<Integer>>> reduce = keyedStream
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .evictor(CountEvictor.of(3))
                .reduce((new ReduceFunction<Tuple2<String, List<Integer>>>() {
                    @Override
                    public Tuple2<String, List<Integer>> reduce(Tuple2<String, List<Integer>> value1, Tuple2<String, List<Integer>> value2) {
                        value1.f1.add(value2.f1.get(0));
                        return value1;
                    }
                }));

public interface Evictor<T, W extends Window> extends Serializable {
    void evictBefore(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);
    void evictAfter(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);

4） Processer
window提供了 reduce，aggregate和 process函数。需要注意的是
reduce和aggreate 是可以增量计算的，也就是不用等待所有的数据都到了窗口之后在计算。
可以预先计算好一部分，比如一些统计功能是可以增量运算的，这样可以很好的提高性能。
而对于一些复杂的功能，是需要全量数据作为先决条件的，这就需要用process 函数。