Flink中的时间和窗口（二）

╰半夏微凉° 2023-09-27 09:32 81阅读 0赞

**目录**

6.3 窗口（Window）

6.3.1 窗口的概念

6.3.2 窗口的分类

6.3.3 窗口 API 概览

6.3.4 窗口分配器（Window Assigners）

6.3.5 窗口函数（Window Functions）

--------------------

## 6.3 窗口（Window） ##

在流处理中，我们往往需要面对的是连续 不断、无休无止的无界流，不可能等到所有所有数据都到齐了才开始处理。所以聚合计算其实 只能针对当前已有的数据——之后再有数据到来，就需要继续叠加、再次输出结果。这样似乎 很“实时”，但现实中大量数据一般会同时到来，需要并行处理，这样频繁地更新结果就会给 系统带来很大负担了。

更加高效的做法是，把无界流进行**切分**，每一段数据分别进行**聚合**，结果只输出一次。这 就相当于**将无界流的聚合转化为了有界数据集的聚合**，这就是所谓的“窗口”（Window）聚合操作。窗口聚合其实是对实时性和处理效率的一个权衡。在实际应用中，我们往往更关心一段 时间内数据的统计结果，比如在过去的 1 分钟内有多少用户点击了网页。在这种情况下，我们就可以定义一个窗口，收集最近一分钟内的所有用户点击数据，然后进行聚合统计，最终输出 一个结果就可以了。

### 6.3.1 窗口的概念 ###

Flink 是一种流式计算引擎，主要是来处理无界数据流的，数据源源不断、无穷无尽。想 要更加方便高效地处理无界流，一种方式就是**将无限数据切割成有限的“数据块”进行处理**，这 就是所谓的“**窗口**”（Window）。

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

![2a38f9d67063414ba4548572747c8515.png][]

我们可以梳理一下事件时间语义下，之前例子中窗口的处理过程：

> （1）第一个数据时间戳为 2，判断之后创建第一个窗口\[0, 10），并将 2 秒数据保存进去；
> 
> （2）后续数据依次到来，时间戳均在 \[0, 10）范围内，所以全部保存进第一个窗口；
> 
> （3）11 秒数据到来，判断它不属于\[0, 10）窗口，所以创建第二个窗口\[10, 20），并将 11秒的数据保存进去。由于水位线设置延迟时间为 2 秒，所以现在的时钟是 9 秒，第一个窗口也 没有到关闭时间；
> 
> （4）之后又有 9 秒数据到来，同样进入\[0, 10）窗口中；
> 
> （5）12 秒数据到来，判断属于\[10, 20）窗口，保存进去。这时产生的水位线推进到了 10秒，所以 \[0, 10）窗口应该关闭了。第一个窗口收集到了所有的 7 个数据，进行处理计算后输 出结果，并将窗口关闭销毁；
> 
> （6）同样的，之后的数据依次进入第二个窗口，遇到 20 秒的数据时会创建第三个窗口\[20, 30）并将数据保存进去；遇到 22 秒数据时，水位线达到了 20 秒，第二个窗口触发计算，输出 结果并关闭。

这里需要注意的是，**Flink 中窗口并不是静态准备好的，而是动态创建**——**当有落在这个窗口区间范围的数据达到时，才创建对应的窗口。**另外，这里我们认为**到达窗口结束时间**时， **窗口就触发计算并关闭**，事实上“触发计算”和“窗口关闭”两个行为也可以分开。

### 6.3.2 窗口的分类 ###

**1. 按照驱动类型分类**

窗口本身是截取有界数据的一种方式，所以窗口一个非常重要的信息其实就是“怎样截取 数据”。换句话说，就是以什么标准来开始和结束数据的截取，我们把它叫作窗口的“驱动类 型”。

我们最容易想到的就是按照时间段去截取数据，这种窗口就叫作“**时间窗口**”（Time Window）。这在实际应用中最常见，之前所举的例子也都是时间窗口。除了由时间驱动之外， 窗口其实也可以由数据驱动，也就是说按照固定的个数，来截取一段数据集，这种窗口叫作“**计数窗口**”（Count Window）

![e28294d9df464b42a6aa76450593594c.png][]

**（1）时间窗口（Time Window）**

时间窗口**以时间点来定义窗口的开始（start）和结束（end）**，所以截取出的就是某一时间 段的数据。到达结束时间时，窗口不再收集数据，触发计算输出结果，并将窗口关闭销毁。所 以可以说基本思路就是“定点发车”。

用结束时间减去开始时间，得到这段时间的长度，就是窗口的大小（window size）。这里 的时间可以是不同的语义，所以我们可以定义处理时间窗口和事件时间窗口。 Flink 中有一个专门的类来表示时间窗口，名称就叫作 **TimeWindow**。这个类只有两个私有属性：start 和 end，表示窗口的开始和结束的时间戳，单位为毫秒。

private final long start; 
    private final long end;

我们可以调用公有的 getStart()和 getEnd()方法直接获取这两个时间戳。另外，TimeWindow还提供了一个 maxTimestamp()方法，用来获取窗口中能够包含数据的最大时间戳。

public long maxTimestamp() { 
     return end - 1; 
    }

**（2）计数窗口（Count Window）**

**计数窗口基于元素的个数来截取数据，到达固定的个数时就触发计算并关闭窗口。**这相当 于座位有限、“人满就发车”，是否发车与时间无关。每个窗口截取数据的个数，就是窗口的大 小。

计数窗口相比时间窗口就更加简单，我们只需指定窗口大小，就可以把数据分配到对应的 窗口中了。在 Flink 内部也并没有对应的类来表示计数窗口，底层是通过“全局窗口”（Global Window）来实现的。

**2. 按照窗口分配数据的规则分类**

时间窗口和计数窗口，只是对窗口的一个大致划分；在具体应用时，**还需要定义更加精细的规则，来控制数据应该划分到哪个窗口中去**。不同的分配数据的方式，就可以有不同的功能应用。

根据分配数据的规则，窗口的具体实现可以分为 4 类：**滚动窗口（Tumbling Window）、 滑动窗口（Sliding Window）、会话窗口（Session Window），以及全局窗口（Global Window）**。

**（1）滚动窗口（Tumbling Windows）**

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

滚动窗口可以**基于时间定义**，也可以**基于数据个数定义**；**需要的参数只有一个，就是窗口 的大小（window size）**。比如我们可以定义一个长度为 1 小时的滚动时间窗口，那么每个小时 就会进行一次统计；或者定义一个长度为 10 的滚动计数窗口，就会每 10 个数进行一次统计。

![b3e7fb3af874466ca1877aebb226ee1a.png][]

小圆点表示流中的数据，我们对数据按照 userId 做了分区。当固定了窗 口大小之后，所有分区的窗口划分都是一致的；窗口没有重叠，每个数据只属于一个窗口。滚动窗口应用非常广泛，它可以对每个时间段做聚合统计，很多 BI 分析指标都可以用它 来实现。

**（2）滑动窗口（Sliding Windows）**

区别在于，窗口之间并不是首尾相接的， 而是可以“错开”一定的位置。如果看作一个窗口的运动，那么就像是向前小步“滑动”一样。

既然是向前滑动，那么每一步滑多远，就也是可以控制的。所以定义滑动窗口的参数有两 个：除去窗口大小（window size）之外，还有一个“滑动步长”（window slide），它其实就代 表了窗口计算的频率。滑动的距离代表了下个窗口开始的时间间隔，而窗口大小是固定的，所 以也就是两个窗口结束时间的间隔；窗口在结束时间触发计算输出结果，那么滑动步长就代表了计算频率。例如，我们定义一个长度为 1 小时、滑动步长为 5 分钟的滑动窗口，那么就会统 计 1 小时内的数据，每 5 分钟统计一次。同样，滑动窗口可以基于时间定义，也可以基于数据 个数定义。

![2f84d9329d934f4384d065b574cae290.png][]

在一些场景中，可能需要统计最近一段时间内的指标，而结果的输出频率要求又很高，甚 至要求实时更新，比如股票价格的 24 小时涨跌幅统计，或者基于一段时间内行为检测的异常 报警。这时滑动窗口无疑就是很好的实现方式。

**（3）会话窗口（Session Windows）**

会话窗口顾名思义，是基于“会话”（session）来来对数据进行分组的。这里的会话类似Web 应用中 session 的概念，不过并不表示两端的通讯过程，而是借用会话超时失效的机制来 描述窗口。简**单来说，就是数据来了之后就开启一个会话窗口，如果接下来还有数据陆续到来， 那么就一直保持会话；如果一段时间一直没收到数据，那就认为会话超时失效，窗口自动关闭。** 这就好像我们打电话一样，如果时不时总能说点什么，那说明还没聊完；如果陷入了尴尬的沉 默，半天都没话说，那自然就可以挂电话了。

考虑到事件时间语义下的乱序流，这里又会有一些麻烦。相邻两个数据的时间间隔 gap大于指定的 size，我们认为它们属于两个会话窗口，前一个窗口就关闭；可在数据乱序的情况 下，可能会有迟到数据，它的时间戳刚好是在之前的两个数据之间的。这样一来，之前我们判 断的间隔中就不是“一直没有数据”，而缩小后的间隔有可能会比 size 还要小——这代表三个数据本来应该属于同一个会话窗口。

所以在 Flink 底层，对会话窗口的处理会比较特殊：**每来一个新的数据，都会创建一个新的会话窗口；然后判断已有窗口之间的距离，如果小于给定的 size，就对它们进行合并（merge） 操作。**在 Window 算子中，对会话窗口会有单独的处理逻辑。

![2d4c8f33fcd443d38d1842b870e9f2be.png][]**（4）全局窗口（Global Windows）**

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

![7736ca76d10d4c67a090efc6429ad57d.png][]

全局窗口没有结束的时间点，所以一般在希望做更加灵活的 窗口处理时自定义使用。Flink 中的计数窗口（Count Window），底层就是用全局窗口实现的。

### 6.3.3 窗口 API 概览 ###

**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 Windows）

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

在代码中，直接基于 DataStream 调用.windowAll()定义窗口。

stream.windowAll(...)

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

**2. 代码中窗口 API 的调用**

有了前置的基础，接下来我们就可以真正在代码中实现一个窗口操作了。简单来说，窗口 操作主要有两个部分：**窗口分配器（Window Assigners）**和**窗口函数（Window Functions）**。

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

其中.window()方法需要**传入一个窗口分配器**，它指明了窗口的类型；而后面的.aggregate()方法**传入一个窗口函数**作为参数，它**用来定义窗口具体的处理逻辑**。窗口分配器有各种形式， 而窗口函数的调用方法也不只.aggregate()一种，我们接下来就详细展开讲解。

另外，在实际应用中，一般都需要并行执行任务，非按键分区很少用到，所以我们之后都以按键分区窗口为例；如果想要实现非按键分区窗口，只要前面不做 keyBy，后面调用.window()时直接换成.windowAll()就可以了。

### 6.3.4 窗口分配器（Window Assigners） ###

定义窗口分配器（Window Assigners）是构建窗口算子的第一步，它的作用就是定义数据 应该被“分配”到哪个窗口。从 6.3.2 节的介绍中我们知道，窗口分配数据的规则，其实就对 应着不同的窗口类型。所以可以说，窗口分配器其实就是在指定窗口的类型。

窗口分配器最通用的定义方式，就是调用**.window()**方法。这个方法需要传入一个**WindowAssigner** 作为参数，返回 **WindowedStream**。

如果是**非按键分区窗口**，那么直接调用**.windowAll()**方法，同样传入一个 **WindowAssigner**，返回的是 **AllWindowedStream**。

窗口按照驱动类型可以分成时间窗口和计数窗口，而按照具体的分配规则，又有滚动窗口、 滑动窗口、会话窗口、全局窗口四种。除去需要自定义的全局窗口外，其他常用的类型 Flink中都给出了内置的分配器实现，我们可以方便地调用实现各种需求。

**1. 时间窗口**

（1）滚动处理时间窗口

窗口分配器由类 TumblingProcessingTimeWindows 提供，需要调用它的静态方法.of()。

stream.keyBy(...) 
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) 
    .aggregate(...)

这里.of()方法需要传入一个 Time 类型的参数 size，表示滚动窗口的大小，我们这里创建 了一个长度为 5 秒的滚动窗口。

另外，.of()还有一个重载方法，可以传入两个 Time 类型的参数：size 和 offset。第一个参数size仍然是定义窗口大小，第二个参数offset则表示偏移量。如统计今日两点到明日两点的数据，则偏移量为2.默认值为0。

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

（2）滑动处理时间窗口

窗口分配器由类 SlidingProcessingTimeWindows 提供，同样需要调用它的静态方法.of()。

stream.keyBy(...) 
    .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5))) 
    .aggregate(...)

这里.of()方法需要传入两个 Time 类型的参数：size 和 slide，前者表示滑动窗口的大小， 后者表示滑动窗口的滑动步长。我们这里创建了一个长度为 10 秒、滑动步长为 5 秒的滑动窗 口。

滑动窗口同样可以追加第三个参数，用于指定窗口起始点的偏移量，用法与滚动窗口完全 一致。

（3）处理时间会话窗口

窗口分配器由类 ProcessingTimeSessionWindows 提供，需要调用它的静态方法.withGap()或者.withDynamicGap()。

stream.keyBy(...) 
    .window(ProcessingTimeSessionWindows.withGap(Time.seconds(10))) 
    .aggregate(...)

这里**.withGap()方法需要传入一个 Time 类型的参数 size，表示会话的超时时间，也就是最 小间隔 session gap。**我们这里创建了静态会话超时时间为 10 秒的会话窗口。

.window(ProcessingTimeSessionWindows.withDynamicGap(
    new SessionWindowTimeGapExtractor<Tuple2<String, Long>>() { 
     @Override 
     public long extract(Tuple2<String, Long> element) { 
    // 提取 session gap 值返回, 单位毫秒 
     return element.f0.length() * 1000; 
     } 
    }))

这里.withDynamicGap()方法需要传入一个 SessionWindowTimeGapExtractor 作为参数，用 来定义 session gap 的动态提取逻辑。在这里，我们提取了数据元素的第一个字段，用它的长 度乘以 1000 作为会话超时的间隔。

（4）滚动事件时间窗口

窗口分配器由类 TumblingEventTimeWindows 提供，**用法与滚动处理事件窗口完全一致**。

stream.keyBy(...) 
    .window(TumblingEventTimeWindows.of(Time.seconds(5))) 
    .aggregate(...)

这里.of()方法也可以传入第二个参数 offset，用于设置窗口起始点的偏移量。

（5）滑动事件时间窗口

窗口分配器由类 SlidingEventTimeWindows 提供，**用法与滑动处理事件窗口完全一致**。

stream.keyBy(...) 
    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))) 
    .aggregate(...)

（6）事件时间会话窗口

窗口分配器由类 EventTimeSessionWindows 提供，**用法与处理事件会话窗口完全一致**。

stream.keyBy(...) 
    .window(EventTimeSessionWindows.withGap(Time.seconds(10))) 
    .aggregate(...)

**2. 计数窗口（底层由全局窗口实现）**

Flink 为我 们提供了非常方便的接口：直接调用.countWindow()方法。根据分配规则的不同，又可以分为 滚动计数窗口和滑动计数窗口两类，下面我们就来看它们的具体实现。

（1）滚动计数窗口

滚动计数窗口只需要传入一个长整型的参数 size，表示窗口的大小。

stream.keyBy(...) 
    .countWindow(10)

我们定义了一个长度为 10 的滚动计数窗口，当窗口中元素数量达到 10 的时候，就会触发 计算执行并关闭窗口。

（2）滑动计数窗口

与滚动计数窗口类似，不过需要在.countWindow()调用时传入两个参数：**size 和 slide，前者表示窗口大小，后者表示滑动步长。**

stream.keyBy(...) 
    .countWindow(10，3)

我们定义了一个长度为 10、滑动步长为 3 的滑动计数窗口。每个窗口统计 10 个数据，每 隔 3 个数据就统计输出一次结果。

**3. 全局窗口**

全局窗口是计数窗口的底层实现，一般在需要自定义窗口时使用。它的定义同样是直接调 用.window()，分配器由 GlobalWindows 类提供。

stream.keyBy(...) 
    .window(GlobalWindows.create());

需要注意使用全局窗口，**必须自行定义触发器才能实现窗口计算**，否则起不到任何作用。

### 6.3.5 窗口函数（Window Functions） ###

在窗口分配器之后，必须再接上一个定义窗 口如何进行计算的操作，这就是所谓的“**窗口函数**”（window functions）。

经**窗口分配器处理**之后，**数据可以分配到对应的窗口**中，而**数据流经过转换得到的数据类 型是 WindowedStream**。这个类型并不是 DataStream，所以并不能直接进行其他转换，而**必须 进一步调用窗口函数**，**对收集到的数据进行处理计算之后**，才能最终再次得到 **DataStream**

![ee410baea96b48e59e15ce7c5eee36f4.png][]

窗口函数定义了要对窗口中收集的数据做的计算操作，根据处理的方式可以分为两类：**增量聚合函数**和**全窗口函数**。下面我们来进行分别讲解。

**1. 增量聚合函数（incremental aggregation functions）**

窗口将数据收集起来，最基本的处理操作当然就是进行聚合。窗口对无限流的切分，可以看作得到了一个有界数据集。如果我们等到所有数据都收集齐，在窗口到了结束时间要输出结 果的一瞬间再去进行聚合，显然就不够高效了——这相当于真的在用批处理的思路来做实时流 处理。

为了提高实时性，我们可以再次将流处理的思路发扬光大：就像 DataStream 的简单聚合 一样，每来一条数据就立即进行计算，中间只要保持一个简单的聚合状态就可以了；区别只是 在于不立即输出结果，而是要等到窗口结束时间。等到窗口到了结束时间需要输出计算结果的时候，我们只需要拿出之前聚合的状态直接输出，这无疑就大大提高了程序运行的效率和实时性。

典型的增量聚合函数有两个：**ReduceFunction** 和 **AggregateFunction**。

（1）归约函数（ReduceFunction）

最基本的聚合方式就是归约（reduce）。我们在基本转换的聚合算子中介绍过 reduce 的用 法，窗口的归约聚合也非常类似，就是将窗口中收集到的数据两两进行归约。当我们进行流处理时，就是要保存一个状态；每来一个新的数据，就和之前的聚合状态做归约，这样就实现了增量式的聚合。

窗口函数中也提供了 ReduceFunction：只要基于 WindowedStream 调用.reduce()方法，然 后传入 ReduceFunction 作为参数，就可以指定以归约两个元素的方式去对窗口中数据进行聚 合了。这里的 ReduceFunction 其实与简单聚合时用到的 ReduceFunction 是同一个函数类接口， 所以使用方式也是完全一样的。

我们回忆一下，**ReduceFunction 中需要重写一个 reduce 方法**，**它的两个参数代表输入的两 个元素，而归约最终输出结果的数据类型**，**与输入的数据类型必须保持一致**。也就是说，**中间聚合的状态和输出的结果，都和输入的数据类型是一样的。**

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner; 
    import org.apache.flink.api.common.eventtime.WatermarkStrategy; 
    import org.apache.flink.api.common.functions.MapFunction; 
    import org.apache.flink.api.common.functions.ReduceFunction; 
    import org.apache.flink.api.java.tuple.Tuple2; 
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; 
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
    import 
    org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindo
    ws; 
    import org.apache.flink.streaming.api.windowing.time.Time; 
    import java.time.Duration; 
     
    public class WindowReduceExample { 
     public static void main(String[] args) throws Exception { 
     StreamExecutionEnvironment env = 
    StreamExecutionEnvironment.getExecutionEnvironment(); 
     env.setParallelism(1); 
     
     // 从自定义数据源读取数据，并提取时间戳、生成水位线 
     SingleOutputStreamOperator<Event> stream = env.addSource(new 
    ClickSource()) 
     .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoun
    dedOutOfOrderness(Duration.ZERO) 
     .withTimestampAssigner(new SerializableTimestampAssigner<Event>() 
    { 
     @Override 
     public long extractTimestamp(Event element, long recordTimestamp) 
    { 
     return element.timestamp; 
     } 
     })); stream.map(new MapFunction<Event, Tuple2<String, 
    Long>>() { 
     @Override 
     public Tuple2<String, Long> map(Event value) throws Exception { 
     // 将数据转换成二元组，方便计算 
     return Tuple2.of(value.user, 1L); 
     } 
     }) 
     .keyBy(r -> r.f0) 
     // 设置滚动事件时间窗口 
     .window(TumblingEventTimeWindows.of(Time.seconds(5))) 
     .reduce(new ReduceFunction<Tuple2<String, Long>>() { 
     @Override 
     public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, 
    Tuple2<String, Long> value2) throws Exception { 
     // 定义累加规则，窗口闭合时，向下游发送累加结果 
     return Tuple2.of(value1.f0, value1.f1 + value2.f1); 
     } 
     }) 
     .print(); 
     
     env.execute(); 
     } 
    }

代码中我们对每个用户的行为数据进行了开窗统计。与 word count 逻辑类似，首先将数 据转换成(user, count)的二元组形式（类型为 Tuple2），每条数据对应的初始 count值都是 1；然后按照用户 id 分组，在处理时间下开滚动窗口，统计每 5 秒内的用户行为数量。 对于窗口的计算，我们用 ReduceFunction 对 count 值做了增量聚合：窗口中会将当前的总 count值保存成一个归约状态，每来一条数据，就会调用内部的 reduce 方法，将新数据中的 count值叠加到状态上，并得到新的状态保存起来。等到了 5 秒窗口的结束时间，就把归约好的状态 直接输出。

这里需要注意，**我们经过窗口聚合转换输出的数据，数据类型依然是二元组 Tuple2。**

（2）聚合函数（AggregateFunction）

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

于是自然可以想到，如果**取消类型一致的限制，让输入数据、中间状态、输出结果三者类 型都可以不同**，不就可以一步直接搞定了吗？ Flink 的 Window API 中的 aggregate 就提供了这样的操作。直接基于 WindowedStream 调 用.aggregate()方法，就可以定义更加灵活的窗口聚合操作。这个方法需要传入一个AggregateFunction 的实现类作为参数。AggregateFunction 在源码中的定义如下：

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); 
    }

AggregateFunction 可以看作是 ReduceFunction 的通用版本，这里有三种类型：**输入类型 （IN）**、**累加器类型（ACC）**和**输出类型（OUT）**。输入类型 IN 就是输入流中元素的数据类型； 累加器类型 ACC 则是我们进行聚合的中间状态类型；而输出类型当然就是最终计算结果的类 型了。

接口中有四个方法：

⚫ createAccumulator()：**创建一个累加器**，这就是为聚合创建了一个初始状态，每个聚 合任务只会调用一次。

⚫ add()：**将输入的元素添加到累加器中。**这就是基于聚合状态，对新来的数据进行进 一步聚合的过程。方法传入两个参数：当前新到的数据 value，和当前的累加器accumulator；返回一个新的累加器值，也就是对聚合状态进行更新。每条数据到来之 后都会调用这个方法。

⚫ getResult()：**从累加器中提取聚合的输出结果。**也就是说，我们可以定义多个状态， 然后再基于这些聚合的状态计算出一个结果进行输出。比如之前我们提到的计算平均 值，就可以把 sum 和 count 作为状态放入累加器，而在调用这个方法时相除得到最终 结果。这个方法只在窗口要输出结果时调用。

⚫ merge()：**合并两个累加器**，并将合并后的状态作为一个累加器返回。这个方法只在 需要合并窗口的场景下才会被调用；最常见的合并窗口（Merging Window）的场景 就是会话窗口（Session Windows）。

所以可以看到，AggregateFunction 的工作原理是：

> 首先调用 createAccumulator()为任务初 始化一个状态(累加器)；而后每来一个数据就调用一次 add()方法，对数据进行聚合，得到的 结果保存在状态中；等到了窗口需要输出时，再调用 getResult()方法得到计算结果。很明显， 与 ReduceFunction 相同，AggregateFunction 也是增量式的聚合；而**由于输入、中间状态、输出的类型可以不同，使得应用更加灵活方便。**

下面来看一个具体例子。我们知道，在电商网站中，**PV（页面浏览量）**和 **UV（独立访客 数）**是非常重要的两个流量指标。一般来说，**PV 统计的是所有的点击量**；而**对用户 id 进行去 重之后，得到的就是 UV**。所以有时我们会用 PV/UV 这个比值，来表示“**人均重复访问量**”， 也就是平均每个用户会访问多少次页面，这在一定程度上代表了用户的粘度。

代码实现如下：

package com.atguigu.chapter06;
    
    import com.atguigu.chapter05.ClickSource;
    import com.atguigu.chapter05.Event;
    import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
    import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    import org.apache.flink.api.common.functions.AggregateFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    
    import java.time.Duration;
    import java.util.HashSet;
    
    public class WindowAggregateTest_PvUV {
        //开窗统计pv页面浏览量和uv独立访客数，两者相除得到:人均重复访问量
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.getConfig().setAutoWatermarkInterval(100);
    
            SingleOutputStreamOperator<Event> stream=env.addSource(new ClickSource())
    
                    //1.WaterMark的生成器
                    .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)//变成有序
                            .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                @Override
                                //2.时间戳的提取器
                                public long extractTimestamp(Event event, long l) {
                                    return event.timestamp;
                                }
                            })
                    );
    
            stream.print("data");
    
            //所有数据放在一起统计pv和uv
            stream.keyBy(date -> true)
            .window(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(2)))
                    .aggregate(new AvgPv())
                            .print();
    
            env.execute();
        }
        //自定义一个AggregateFunction,用Long保存pv个数，用HashSet做Uv去重
        public static class AvgPv implements AggregateFunction<Event, Tuple2<Long, HashSet<String>>,Double>{
    
            @Override
            public Tuple2<Long, HashSet<String>> createAccumulator() {
                return Tuple2.of(0L,new HashSet<>());
            }
    
            @Override
            public Tuple2<Long, HashSet<String>> add(Event event, Tuple2<Long, HashSet<String>> longHashSetTuple2) {
                //每来一条数据，pv个数+1，将user放入hashSet中
                longHashSetTuple2.f1.add(event.user);
                return Tuple2.of(longHashSetTuple2.f0+1,longHashSetTuple2.f1);
            }
    
            @Override
            public Double getResult(Tuple2<Long, HashSet<String>> longHashSetTuple2) {
               //窗口触发时，输出pv和uv的值
                return (double)longHashSetTuple2.f0/longHashSetTuple2.f1.size();
            }
    
            @Override
            public Tuple2<Long, HashSet<String>> merge(Tuple2<Long, HashSet<String>> longHashSetTuple2, Tuple2<Long, HashSet<String>> acc1) {
                return null;
            }
        }
    }

代码中我们创建了事件时间滑动窗口，统计 10 秒钟的“人均 PV”，每 2 秒统计一次。由 于聚合的状态还需要做处理计算，因此窗口聚合时使用了更加灵活的 AggregateFunction。为了 统计 UV，我们用一个 HashSet 保存所有出现过的用户 id，实现自动去重；而 PV 的统计则类 似一个计数器，每来一个数据加一就可以了。所以这里的状态，定义为包含一个 HashSet 和一 个 count 值的二元组（Tuple2, Long>），每来一条数据，就将 user 存入 HashSet， 同时 count 加 1。这里的 count 就是 PV，而 HashSet 中元素的个数（size）就是 UV；所以最终 窗口的输出结果，就是它们的比值。这里没有涉及会话窗口，所以 merge()方法可以不做任何操作。

另外，Flink 也为窗口的聚合提供了一系列预定义的简单聚合方法，可以直接基于WindowedStream 调用。主要包括.sum()/max()/maxBy()/min()/minBy()，与 KeyedStream 的简单 聚合非常相似。它们的底层，其实都是通过 AggregateFunction 来实现的。

通过 ReduceFunction 和 AggregateFunction 我们可以发现，增量聚合函数其实就是在用流 处理的思路来处理有界数据集，核心是保持一个聚合状态，当数据到来时不停地更新状态。这就是 Flink 所谓的“有状态的流处理”，通过这种方式可以极大地提高程序运行的效率，所以 在实际应用中最为常见。

[2a38f9d67063414ba4548572747c8515.png]: https://img-blog.csdnimg.cn/2a38f9d67063414ba4548572747c8515.png
[e28294d9df464b42a6aa76450593594c.png]: https://img-blog.csdnimg.cn/e28294d9df464b42a6aa76450593594c.png
[b3e7fb3af874466ca1877aebb226ee1a.png]: https://img-blog.csdnimg.cn/b3e7fb3af874466ca1877aebb226ee1a.png
[2f84d9329d934f4384d065b574cae290.png]: https://img-blog.csdnimg.cn/2f84d9329d934f4384d065b574cae290.png
[2d4c8f33fcd443d38d1842b870e9f2be.png]: https://img-blog.csdnimg.cn/2d4c8f33fcd443d38d1842b870e9f2be.png
[7736ca76d10d4c67a090efc6429ad57d.png]: https://img-blog.csdnimg.cn/7736ca76d10d4c67a090efc6429ad57d.png
[ee410baea96b48e59e15ce7c5eee36f4.png]: https://img-blog.csdnimg.cn/ee410baea96b48e59e15ce7c5eee36f4.png