Flink（八）【窗口】

前言

终于忙完了四门专业课的期末，确实挺累啊。今天开始继续学习 Flink ，接着上次的内容。

今日摘录：

他觉得一个人奋斗更轻松自在。跟没有干劲的人在一起厮混，只会徒增压力。

-《解忧杂货店》

1、窗口

之前我们已经了解了 Flink 中基本的聚合操作。在流处理中，我们往往需要面对的是连续不断、无休无止的无界流，不可能等到所有所有数据都到齐了才开始处理。所以聚合计算其实只能针对当前已有的数据——之后再有数据到来，就需要继续叠加、再次输出结果。这样似乎很“实时”，但现实中大量数据一般会同时到来，需要并行处理，这样频繁地更新结果就会给系统带来很大负担了。更加高效的做法是，把无界流进行切分，每一段数据分别进行聚合，结果只输出一次。这就相当于将无界流的聚合转化为了有界数据集的聚合，这就是所谓的“窗口”（Window）聚合操作。窗口聚合其实是对实时性和处理效率的一个权衡。在实际应用中，我们往往更关心一段时间内数据的统计结果，比如在过去的 1 分钟内有多少用户点击了网页。在这种情况下，我们就可以定义一个窗口，收集最近一分钟内的所有用户点击数据，然后进行聚合统计，最终输出一个结果就可以了。在 Flink 中，提供了非常丰富的窗口操作，下面我们就来详细介绍。

1.1、窗口的概念

这里的窗口和我们之前 Hive 中学到的窗口的概念是不一样的。Hive的窗口函数主要用于离线数据的聚合，是基于字段范围的，而Flink的窗口则是基于时间更多地用于处理实时数据流。

在 Flink 中，窗口可以把流切割成有限大小的多个“存储桶”（bucket)；每个数据都会分发到对应的桶中，当到达窗口结束时间时，就对每个桶中收集的数据进行计算处理。

Flink 中窗口并不是静态准备好的，而是动态创建的，只有这个窗口区间内的数据到达时，才会去创建对应的窗口。

1.2、窗口的分类

1.2.1、按照度量标准

1）时间窗口

以时间点来定义窗口的开始和结束，所以截取出的就是某一段时间的数据。到达结束时间时，窗口不再收集数据，触发计算输出结果，并将窗口关闭销毁。

2）计数窗口

基于元素个数来截取数据，元素达到固定的个数时，就触发计算并关闭窗口。

1.2.2、按照窗口分配数据的规则

1）滚动窗口

滚动窗口有固定的大小，是一种对数据进行“均匀切片”的划分方式。窗口之间没有重叠，也不会有间隔，是“首尾相接”的状态。如果我们把多个窗口的创建，看作一个窗口的运动，那就好像它在不停地向前“翻滚”一样。也正是因为滚动窗口是“无缝衔接”，所以每个数据都会被分配到一个窗口，而且只会属于一个窗口。

2）滑动窗口

与滚动窗口类似，滑动窗口的大小也是固定的。区别在于，窗口之间并不是首尾相接的，而是可以“错开”一定的位置。如果看作一个窗口的运动，那么就像是向前小步“滑动”一样。既然是向前滑动，那么每一步滑多远，就也是可以控制的。

滑动窗口的参数有两个：窗口大小（window size）和 “滑动步长”（window slide）。当滑动步长等于窗口大小时，这就像是一个滚动窗口；当滑动步长大于窗口大小时，会有数据的遗漏。所以一般情况下，我们会让滑动步长小于窗口大小，并尽量设置为整数倍的关系。

3）会话窗口

会话窗口顾名思义，是基于“会话”（session）来来对数据进行分组的。这里的会话类似Web 应用中 session 的概念，不过并不表示两端的通讯过程，而是借用会话超时失效的机制来描述窗口。简单来说，就是数据来了之后就开启一个会话窗口，如果接下来还有数据陆续到来，那么就一直保持会话；如果一段时间一直没收到数据，那就认为会话超时失效，窗口自动关闭。

与滑动窗口和滚动窗口不同，会话窗口只能基于时间来定义，而没有“会话计数窗口”的概念。

4）全局窗口

还有一类比较通用的窗口，就是“全局窗口”。这种窗口全局有效，会把相同 key 的所有数据都分配到同一个窗口中；说直白一点，就跟没分窗口一样。无界流的数据永无止尽，所以这种窗口也没有结束的时候，默认是不会做触发计算的。如果希望它能对数据进行计算处理，还需要自定义“触发器”（Trigger）。

1.3、API

1.3.1、按键分区（Keyed）和非按键分区（Non-Keyed）

在定义窗口操作之前，首先需要确定，到底是基于按键分区（Keyed）的数据流 KeyedStream来开窗，还是直接在没有按键分区的 DataStream 上开窗。也就是说，在调用窗口算子之前，是否有 keyBy 操作。

1）按键分区窗口（Keyed Windows）

经过按键分区 keyBy 操作后，数据流会按照 key 被分为多条逻辑流（logical streams），这就是 KeyedStream。基于 KeyedStream 进行窗口操作时, 窗口计算会在多个并行子任务上同时执行。相同 key 的数据会被发送到同一个并行子任务，而窗口操作会基于每个 key 进行单独的处理。所以可以认为，每个 key 上都定义了一组窗口，各自独立地进行统计计算。在代码实现上，我们需要先对 DataStream 调用.keyBy()进行按键分区，然后再调用.window()定义窗口。

stream.keyBy(...).window(...)

2）非按键分区（Non-Keyed）

如果没有进行 keyBy，那么原始的 DataStream 就不会分成多条逻辑流。这时窗口逻辑只能在一个任务（task）上执行，就相当于并行度变成了 1。所以在实际应用中一般不推荐使用这种方式。
在代码中，直接基于 DataStream 调用.windowAll()定义窗口。

stream.windowAll(...)

这里需要注意的是，对于非按键分区的窗口操作，手动调大窗口算子的并行度也是无效的，windowAll 本身就是一个非并行的操作。

1.3.2、API 的调用

窗口操作主要有两个部分：窗口分配器（Window Assigners）和窗口函数（Window Functions）。

stream.keyBy(<key selector>)
    .window(<window assigner>)
    .aggregate(<window function>)

其中.window()方法需要传入一个窗口分配器，它指明了窗口的类型；而后面的.aggregate()方法传入一个窗口函数作为参数，它用来定义窗口具体的处理逻辑。窗口分配器有各种形式，而窗口函数的调用方法也不只.aggregate()一种。另外，在实际应用中，一般都需要并行执行任务，非按键分区很少用到，所以我们之后都以按键分区窗口为例；如果想要实现非按键分区窗口，只要前面不做 keyBy，后面调用.window()时直接换成.windowAll()就可以了。

1.4、窗口分配器

用于指定窗口的类型，也就是指定窗口的度量标准（基于时间/元素个数）和分配数据的规则（滚动/滑动/会话/全局）。

窗口分配器最通用的定义方式，就是调用.window()方法。这个方法需要传入一个WindowAssigner 作为参数，返回 WindowedStream。如果是非按键分区窗口，那么直接调用.windowAll()方法，同样传入一个 WindowAssigner，返回的是 AllWindowedStream。

除去需要自定义的全局窗口外，其他三种常用的类型 Flink中都给出了内置的分配器实现，我们可以方便地调用实现各种需求。

下面的例子我们都使用按键分区的窗口类型，这里给出 keyBy 后得到的数据流对象 KeyedStream：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new WaterSensorFunction());

        KeyedStream<WaterSensor, String> sensorKs = sensorDS.keyBy(sensor -> sensor.getId());

1.4.1、基于时间的窗口

注意：这里的时间有两种语义，一种是处理时间（对于API中 xxxProcessingTimeWindows），一种是事件时间（对于API中 xxxEventProcessingWindows），这涉及到时间语义的内容，之后学到时间语义再详细说。

这里只需要知道，TumblingProcessingTimeWindows和TumblingEventTimeWindows的主要区别在于时间基准和触发机制。TumblingProcessingTimeWindows基于处理时间触发计算，适用于实时性要求高的场景；而TumblingEventTimeWindows基于事件时间触发计算，对于处理延迟到达的数据更加灵活。

1）基于处理时间的滚动窗口

    // 基于时间的滚动创建 10s
        WindowedStream<WaterSensor, String, TimeWindow> tumblingWindow = sensorKs.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

这里的 of 方法还可以指定第二个参数，可以传入两个 Time 类型的参数：size 和 offset。第一个参数当然还是窗口大小，第二个参数则表示窗口起始点的偏移量。

我们知道，不同国家分布在不同的时区。标准时间戳其实就是1970 年 1 月 1 日 0 时 0 分 0 秒 0 毫秒开始计算的一个毫秒数，而这个时间是以 UTC 时间，也就是 0 时区（伦敦时间）为标准的。我们所在的时区是东八区，也就是 UTC+8，跟 UTC 有 8小时的时差。我们定义 1 天滚动窗口时，如果用默认的起始点，那么得到就是伦敦时间每天 0点开启窗口，这时是北京时间早上 8 点。那怎样得到北京时间每天 0 点开启的滚动窗口呢？只要设置-8 小时的偏移量就可以了：

sensorKs.window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))

2）基于处理时间的滑动窗口

// 基于时间的滑动窗口 窗口大小10s 滑动步长2s
        WindowedStream<WaterSensor, String, TimeWindow> slideWindow = sensorKs.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(2)));

3）基于处理时间的会话窗口

注意：会话窗口只能基于时间（处理时间或事件时间），不可以基于计数，因为我们会话窗口它划分窗口的规则只能是基于时间间隔的。

        // 基于时间的会话窗口 中间间隔多久没有数据来,就把前面的数据划分为一个窗口,这里指定5s没有数据来就把前面的数据划分为一个窗口，后面的数据是一个新的窗口
        WindowedStream<WaterSensor, String, TimeWindow> sessionWindow = sensorKs.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));

也可以动态指定间隔（每来一个数据都会修改它的间隔）：

// 基于处理时间的会话窗口 动态间隔
        sensorKs.window(ProcessingTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor<WaterSensor>() {
            @Override
            public long extract(WaterSensor sensor) {
                // 取 WaterSenor 对象的 ts 属性作为间隔，单位 ms 这里*1000方便输入
                return sensor.getTs() * 1000;  
            }
        }));

比如我们连续很快输入：

s1,5,5
s2,6,6
s3,7,7
s8,8,8

它会再我们输出完 "s8,8,8" 后再过八秒没有数据来，就输出窗口（窗口中的值就是计算的结果，具体计算的逻辑取决于我们指定的窗口函数）

4）基于事件时间的滚动窗口

// 基于事件时间的滚动窗口
        sensorKs.window(TumblingEventTimeWindows.of(Time.seconds(10)));

5）基于事件时间的滑动窗口

        // 基于事件时间的滑动窗口
        sensorKs.window(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(2)));

6）基于事件时间的会话窗口

        // 基于事件时间的会话窗口
        sensorKs.window(EventTimeSessionWindows.withGap(Time.seconds(5)));

1.4.2、基于计数的窗口

滚动计数窗口和滑动计数窗口的方法是同一个，只不过一个只需要传入一个参数，一个需要传入两个参数。

1）滚动计数窗口

// 基于计数的滚动窗口 窗口长度=5个元素
        sensorKs.countWindow(5);

2）滑动计数窗口

        // 基于计数的滑动窗口 窗口长度=5个元素，窗口步长的=2个元素，也就是说两个窗口间隔两个元素
        sensorKs.countWindow(5,2);

滑动计数窗口会在数据量每达到步长的时候计算一次，而数据量每达到窗口大小才会关闭窗口开启一个新的窗口。只要经过一个步长，就一定有一个窗口关闭。

对于上面的滑动计数窗口，它每一个步长2就会滑动一次（并输出一次计算结果）。

3）全局窗口

全局窗口一般在自定义的时候才会用到。

我们可以看到，滚动计数窗口和滑动计数窗口的实现的底层就是全局窗口，只不过滚动计数窗口中指定了触发器，滑动计数窗口中指定了触发器和移除器：

1.5、窗口函数

定义了窗口分配器，我们只是知道了数据属于哪个窗口，可以将数据收集起来了；至于收集起来到底要怎么做，这就需要使用窗口函数来操作了（注意：是窗口函数而不是其他普通函数，因为经过划分窗口后的数据流是 WindowedStream 而不是 DataStream）。所以在窗口分配器之后，必须再接上一个定义窗口如何进行计算的操作，这就是所谓的“窗口函数”（window functions）。
经窗口分配器处理之后，数据可以分配到对应的窗口中，而数据流经过转换得到的数据类型是 WindowedStream。这个类型并不是 DataStream，所以并不能直接进行其他转换，而必须进一步调用窗口函数，对收集到的数据进行处理计算之后，才能最终再次得到 DataStream

1.5.1、增量聚合函数（reduce/aggregate）

1）规约函数（ReduceFunction）

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

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new WaterSensorFunction());

        KeyedStream<WaterSensor, String> sensorKs = sensorDS.keyBy(WaterSensor::getId);

        // todo 1. 指定窗口分配器：基于处理时间的滚动窗口
        WindowedStream<WaterSensor, String, TimeWindow> tumblingWindow = sensorKs.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
        // todo 2. 指定窗口函数：增量聚合的规约函数
        SingleOutputStreamOperator<WaterSensor> reduce = tumblingWindow.reduce(new ReduceFunction<WaterSensor>() {
            @Override
            public WaterSensor reduce(WaterSensor sensor1, WaterSensor sensor2) throws Exception {
                System.out.println("调用 reduce 函数，sensor1= " + sensor1 + ",sensor2= " + sensor2);
                return new WaterSensor(sensor1.getId(), sensor2.getTs(), sensor1.getVc() + sensor2.getVc());
            }
        });

        reduce.print();

        env.execute();
    }
}

运行结果：

// 窗口1
调用 reduce 函数，sensor1= WaterSensor{id='s1', ts=1, vc=1},sensor2= WaterSensor{id='s1', ts=2, vc=2}
WaterSensor{id='s1', ts=2, vc=3}    
WaterSensor{id='s3', ts=3, vc=3}
// 窗口2
WaterSensor{id='s3', ts=1, vc=1}
// 窗口3
调用 reduce 函数，sensor1= WaterSensor{id='s4', ts=1, vc=1},sensor2= WaterSensor{id='s4', ts=1, vc=5}
WaterSensor{id='s4', ts=1, vc=6}

2）聚合函数（AggregateFunction）

ReduceFunction 可以解决大多数归约聚合的问题，但是这个接口有一个限制，就是聚合状态的类型、输出结果的类型都必须和输入数据类型一样（可以看上面函数接口 ReduceFunction 中的泛型方法的定义）。这就迫使我们必须在聚合前，先将数据转换（map）成预期结果类型；而在有些情况下，还需要对状态进行进一步处理才能得到输出结果，这时它们的类型可能不同，使用 ReduceFunction 就会非常麻烦。

AggregateFunction 可以看作是 ReduceFunction 的通用版本，这里有三种类型：输入类型（IN）、累加器类型（ACC，也就是我们中间结果的类型）和输出类型（OUT）。

接口中有四个方法：

⚫ createAccumulator()：创建一个累加器，这就是为聚合创建了一个初始状态，每个聚合任务只会调用一次。
⚫ add()：将输入的元素添加到累加器中。这就是基于聚合状态，对新来的数据进行进一步聚合的过程。方法传入两个参数：当前新到的数据 value，和当前的累加器 accumulator；返回一个新的累加器值，也就是对聚合状态进行更新。每条数据到来之后都会调用这个方法。
⚫ getResult()：从累加器中提取聚合的输出结果。也就是说，我们可以定义多个状态，然后再基于这些聚合的状态计算出一个结果进行输出。
⚫ merge()：合并两个累加器，并将合并后的状态作为一个累加器返回。这个方法只在需要合并窗口的场景下才会被调用；最常见的合并窗口（Merging Window）的场景就是会话窗口（Session Windows）。

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

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new WaterSensorFunction());

        KeyedStream<WaterSensor, String> sensorKs = sensorDS.keyBy(WaterSensor::getId);

        // todo 1. 指定窗口分配器：基于处理时间的滚动窗口
        WindowedStream<WaterSensor, String, TimeWindow> tumblingWindow = sensorKs.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
        // todo 2. 指定窗口函数：增量聚合的规约函数
        SingleOutputStreamOperator<String> aggregate = tumblingWindow.aggregate(new AggregateFunction<WaterSensor, Integer, String>() {
            //
            @Override
            public Integer createAccumulator() {
                // 我们要存的是对象,不初始化的话是null,所以需要初始化它的值
                System.out.println("创建累加器(初始化累加器)");
                return 0;
            }

            @Override
            public Integer add(WaterSensor sensor, Integer accumulator) {
                // 计算逻辑
                System.out.println("调用add方法,sensor= "+sensor);
                return accumulator + sensor.getVc();
            }

            @Override
            public String getResult(Integer accumulator) {
                // 获取最终结果输出时 窗口触发输出
                System.out.println("调用 getResult 方法");
                return accumulator.toString();
            }

            @Override
            public Integer merge(Integer a, Integer b) {
                // 一般不用 只有会话窗口才会用到
                System.out.println("调用 merge 方法");
                return null;
            }
        });

        aggregate.print();

        env.execute();
    }
}

运行结果：

// 窗口1
创建累加器(初始化累加器)
调用add方法,sensor= WaterSensor{id='s1', ts=1, vc=1}
创建累加器(初始化累加器)
调用add方法,sensor= WaterSensor{id='s2', ts=1, vc=1}
调用 getResult 方法    //id='s1'的结果
1
调用 getResult 方法    //id='s2'的结果
1
// 窗口2
创建累加器(初始化累加器)
调用add方法,sensor= WaterSensor{id='s1', ts=1, vc=2}
调用add方法,sensor= WaterSensor{id='s1', ts=1, vc=1}
调用add方法,sensor= WaterSensor{id='s1', ts=1, vc=3}
调用 getResult 方法
6
// 窗口3
创建累加器(初始化累加器)
调用add方法,sensor= WaterSensor{id='s2', ts=2, vc=2}
调用add方法,sensor= WaterSensor{id='s2', ts=2, vc=2}
调用 getResult 方法
4
// 窗口4
创建累加器(初始化累加器)
调用add方法,sensor= WaterSensor{id='s2', ts=2, vc=1}

其实这里的增量聚合函数就是用流处理的思路来处理有界数据流，当相同 key 的数据进来时，把数据的状态不断进行更新。这就是 Flink 所谓的“有状态的流处理”，通过这种方式可以极大地提高程序运行的效率，所以在实际应用中最为常见。

1.5.2、全窗口函数（apply/process）

与增量聚合函数不同，全窗口函数需要先收集窗口中的数据，并在内部缓存起来，等到窗口要输出结果的时候再取出数据进行计算。很明显，这就是典型的批处理思路了——先攒数据，等一批都到齐了再正式启动处理流程。

但是把计算放到窗口关闭才去计算无疑是低效的，毕竟如果数据量比较大的时候，这种方式肯定没有增量聚合函数计算的快。那为什么还要使用这种方式呢？这是因为有些场景下，我们要做的计算必须基于全部的数据才有效，这时做增量聚合就没什么意义了；另外，输出的结果有可能要包含上下文中的一些信息（比如窗口的起始时间），这是增量聚合函数做不到的。所以，我们还需要有更丰富的窗口计算方式，这就可以用全窗口函数来实现。
在 Flink 中，全窗口函数也有两种：WindowFunction 和 ProcessWindowFunction。

1）全窗口函数（full window functions）

WindowFunction 字面上就是“窗口函数”，它其实是老版本的通用窗口函数接口。我们可以基于 WindowedStream 调用.apply()方法，传入一个 WindowFunction 的实现类。

注意：这种写法已经不推荐使用了，这里了解即可。

tumblingWindow.apply(new WindowFunction<WaterSensor, String, String, TimeWindow>() {
            /**
             * 
             * @param key 分组的 key
             * @param window 窗口对象
             * @param input 存的数据(迭代器保存着所有传进来的数据)
             * @param out 采集器
             * @throws Exception
             */
            @Override
            public void apply(String key, TimeWindow window, Iterable<WaterSensor> input, Collector<String> out) throws Exception {
                
            }
        });

这个类中可以获取到包含窗口所有数据的可迭代集合（Iterable），还可以拿到窗口（Window）本身的信息。WindowFunction 接口在源码中实现如下：

@Public
public interface WindowFunction<IN, OUT, KEY, W extends Window> extends Function, Serializable {

    /**
     * Evaluates the window and outputs none or several elements.
     *
     * @param key The key for which this window is evaluated.
     * @param window The window that is being evaluated.
     * @param input The elements in the window being evaluated.
     * @param out A collector for emitting elements.
     * @throws Exception The function may throw exceptions to fail the program and trigger recovery.
     */
    void apply(KEY key, W window, Iterable<IN> input, Collector<OUT> out) throws Exception;
}

2）窗口处理函数（ProcessWindowFunction）

ProcessWindowFunction 是 Window API 中最底层的通用窗口函数接口。之所以说它“最底层”，是因为除了可以拿到窗口中的所有数据之外，ProcessWindowFunction 还可以获取到一个“上下文对象”（Context）。这个上下文对象非常强大，不仅能够获取窗口信息，还可以访问当前的时间和状态信息。这里的时间就包括了处理时间（processing time）和事件时间水位线（event time watermark）。这就使得 ProcessWindowFunction 更加灵活、功能更加丰富。事实上， ProcessWindowFunction 是 Flink 底层 API——处理函数（process function）中的一员。

当然，这些好处是以牺牲性能和资源为代价的。作为一个全窗口函数，ProcessWindowFunction 同样需要将所有数据缓存下来、等到窗口触发计算时才使用。它其实就是一个增强版的 WindowFunction。

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

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new WaterSensorFunction());

        KeyedStream<WaterSensor, String> sensorKs = sensorDS.keyBy(WaterSensor::getId);

        // todo 1. 指定窗口分配器：基于处理时间的滚动窗口
        WindowedStream<WaterSensor, String, TimeWindow> tumblingWindow = sensorKs.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        // todo 2. 指定窗口函数：全窗口函数
        SingleOutputStreamOperator<String> process = tumblingWindow.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
            /**
             *
             * @param key 分组的 key
             * @param context 上下文
             * @param elements 全窗口存的数据
             * @param out 采集器
             * @throws Exception
             */
            @Override
            public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                long startTs = context.window().getStart();
                long endTs = context.window().getEnd();
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String start = sdf.format(new Date(startTs));
                String end = sdf.format(new Date(endTs));
                long size = elements.spliterator().estimateSize();
                out.collect("key=" + key + " 的窗口[" + start + "," + end + "]包含" + size + "条数据===>" + elements.toString());
            }
        });

        process.print();

        env.execute();
    }
}

运行结果：

key=s1 的窗口[2023-11-29 16:30:30,2023-11-29 16:30:40]包含1条数据===>[WaterSensor{id='s1', ts=1, vc=1}]
key=s2 的窗口[2023-11-29 16:30:30,2023-11-29 16:30:40]包含2条数据===>[WaterSensor{id='s2', ts=1, vc=1}, WaterSensor{id='s2', ts=1, vc=1}]

这里，我们只用上下文对象获取了窗口的创建时间和关闭时间，其实它还有很多强大的功能，比如侧输出流等。

1.5.3、增量聚合函数结合全窗口函数

        增量聚合函数处理计算会更高效。举一个最简单的例子，对一组数据求和。大量的数据连续不断到来，全窗口函数只是把它们收集缓存起来，并没有处理；到了窗口要关闭、输出结果的时候，再遍历所有数据依次叠加，得到最终结果。而如果我们采用增量聚合的方式，那么只需要保存一个当前和的状态，每个数据到来时就会做一次加法，更新状态；到了要输出结果的时候，只要将当前状态直接拿出来就可以了。增量聚合相当于把计算量“均摊”到了窗口收集数据的过程中，自然就会比全窗口聚合更加高效、输出更加实时。
        而全窗口函数的优势在于提供了更多的信息，可以认为是更加“通用”的窗口操作。它只负责收集数据、提供上下文相关信息，把所有的原材料都准备好，至于拿来做什么我们完全可以任意发挥。这就使得窗口计算更加灵活，功能更加强大。
所以在实际应用中，我们往往希望兼具这两者的优点，把它们结合在一起使用。Flink 的Window API 就给我们实现了这样的用法。
        我们之前在调用 WindowedStream 的.reduce()和.aggregate()方法时，只是简单地直接传入了一个 ReduceFunction 或 AggregateFunction 进行增量聚合。除此之外，其实还可以传入第二个参数：一个全窗口函数，可以是 WindowFunction 或者 ProcessWindowFunction。

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

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new WaterSensorFunction());

        KeyedStream<WaterSensor, String> sensorKs = sensorDS.keyBy(WaterSensor::getId);

        // todo 1. 指定窗口分配器：基于处理时间的滚动窗口
        WindowedStream<WaterSensor, String, TimeWindow> tumblingWindow = sensorKs.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
        // todo 2. 指定窗口函数：增量聚合的规约函数
        SingleOutputStreamOperator<String> result = tumblingWindow.aggregate(
                new AggregateFunction<WaterSensor, Integer, String>() {
                    @Override
                    public Integer createAccumulator() {
                        // 我们要存的是对象,不初始化的话是null,所以需要初始化它的值
                        System.out.println("创建累加器(初始化累加器)");
                        return 0;
                    }

                    @Override
                    public Integer add(WaterSensor sensor, Integer accumulator) {
                        // 计算逻辑
                        System.out.println("调用add方法,sensor= " + sensor);
                        return accumulator + sensor.getVc();
                    }

                    @Override
                    public String getResult(Integer accumulator) {
                        // 获取最终结果输出时 窗口触发输出
                        System.out.println("调用 getResult 方法");
                        return accumulator.toString();
                    }

                    @Override
                    public Integer merge(Integer a, Integer b) {
                        // 一般不用 只有会话窗口才会用到
                        System.out.println("调用 merge 方法");
                        return null;
                    }
                }, new ProcessWindowFunction<String, String, String, TimeWindow>() {    // 输入类型=增量函数的输出类型
                    // 这里的 elements 只有一条数据，因为是从增量聚合函数直接传过来的，我们使用全窗口函数只是为了获得更多的上下文功能
                    @Override
                    public void process(String key, Context context, Iterable<String> elements, Collector<String> out) throws Exception {
                        long startTs = context.window().getStart();
                        long endTs = context.window().getEnd();
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String start = sdf.format(new Date(startTs));
                        String end = sdf.format(new Date(endTs));
                        long size = elements.spliterator().estimateSize();
                        out.collect("key=" + key + " 的窗口[" + start + "," + end + "]包含" + size + "条数据===>" + elements.toString());
                    }
                });

        result.print();

        env.execute();
    }
}

运行结果：

// 窗口1
调用add方法,sensor= WaterSensor{id='s1', ts=1, vc=1}
调用add方法,sensor= WaterSensor{id='s1', ts=1, vc=2}
调用 getResult 方法    //这里调用 getResult 方法并不会输出结果 因为结果被作为输入参数传给了全窗口函数
key=s1 的窗口[2023-11-29 16:48:00,2023-11-29 16:48:10]包含1条数据===>[3]
创建累加器(初始化累加器)
调用add方法,sensor= WaterSensor{id='s2', ts=2, vc=2}
调用 getResult 方法
key=s2 的窗口[2023-11-29 16:48:10,2023-11-29 16:48:20]包含1条数据===>[2]
// 窗口2
创建累加器(初始化累加器)
调用add方法,sensor= WaterSensor{id='s1', ts=1, vc=1}
调用add方法,sensor= WaterSensor{id='s1', ts=1, vc=1}
创建累加器(初始化累加器)
调用add方法,sensor= WaterSensor{id='s2', ts=2, vc=2}
调用 getResult 方法    
key=s1 的窗口[2023-11-29 16:48:20,2023-11-29 16:48:30]包含1条数据===>[2]
调用 getResult 方法
key=s2 的窗口[2023-11-29 16:48:20,2023-11-29 16:48:30]包含1条数据===>[2]

可以看到，我们结合了两者的优点：

增量聚合：来一条算一条，存储中间结果，不会把计算的压力攒到最后。
全窗口函数：可以通过上下文获取更多的功能。

当然，我们的增量聚合函数 reduce 也可以和全窗口函数结合使用，同样是传入两个参数（一个需要实现函数接口 ReduceFunction ，另一个需要实现抽象类 ProcessWindowFunction （或者 WindowFunction 只不过这个现在不常用了））。

1.6、其他 API

其实前面的窗口都有默认的触发器，比如滚动计数窗口中指定了触发器，滑动计数窗口中指定了触发器和移除器。这里理解就好。

1.6.1、触发器（Trigger）

触发器主要是用来控制窗口什么时候触发计算，也就是什么时候执行窗口函数，可以理解为计算得到结果的过程。

基于 WindowedStream 调用 tigger() 方法，就可以传入一个自定义的窗口触发器（Trigger）。

tumblingWindow.trigger(new Trigger<WaterSensor, TimeWindow>() {
            @Override
            public TriggerResult onElement(WaterSensor element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
                return null;
            }

            @Override
            public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                return null;
            }

            @Override
            public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                return null;
            }

            @Override
            public void clear(TimeWindow window, TriggerContext ctx) throws Exception {

            }
        });

拿我们基于处理时间的滚动窗口来说，TumblingProcessingTimeWindows 有一个 getDefaultTrigger 方法来选择默认的触发器，这个方法会返回一个类 ProcessingTimeTrigger 的实例，这个类 ProcessingTimeTrigger 继承自抽象类 Tigger，主要实现了这么几个方法：

onElement()：窗口中每到来一条数据，都会调用这个方法。
onEventTime()：当注册的事件时间定时器触发时，将调用这个方法。
onProcessingTime ()：当注册的处理时间定时器触发时，将调用这个方法。
clear()：当窗口关闭销毁时，调用这个方法。一般用来清除自定义的状态。

我们可以看上面基于计数的滑动窗口的底层代码，我们可以发现，它有默认的触发器和移除器：

我们可以查看它的触发器的内部代码：

1.6.2、移除器（Evictor）

移除器主要用来自定义移除某些数据的逻辑。基于 WindowedStream 调用 evictor() 方法，就可以传入一个自定义的移除器。Evictor 是一个接口，不同的窗口类型有自己预先实现的移除器。

tumblingWindow.evictor(new Evictor<WaterSensor, TimeWindow>() {
            @Override
            public void evictBefore(Iterable<TimestampedValue<WaterSensor>> elements, int size, TimeWindow window, EvictorContext evictorContext) {
                
            }

            @Override
            public void evictAfter(Iterable<TimestampedValue<WaterSensor>> elements, int size, TimeWindow window, EvictorContext evictorContext) {

            }
        });

不管是触发器还是移除器，一般都不需要我们自定义，但要知道它的底层原理。