Flink 中的时间和窗口-第六章

绝地灬酷狼 2023-10-08 23:14 50阅读 0赞

> 借鉴《尚硅谷Flink1.13版本笔记.pdf》中第六章

## Flink 中的时间和窗口 ##

在流数据处理应用中，一个很重要的操作就是窗口计算。所谓的“窗口”，就是划定的一段时间范围，也就是“时间窗”；对这范围内数据进行处理，就是窗口计算。所以窗口和时间往往是分不开的。

### 6.1 时间语义 ###

#### 6.1.1 Flink 中的时间语义 ####

![bc9334f9d82e48eb93840930f1273941.png][]

如图 6-1 所示，事件发生后，生成的数据被收集起来，首先进入分布式消息队列，后被 Flink 系统中的 Source 算子读取消费，进而向下游 的转换算子（窗口算子）传递，最终由窗口算子进行计算处理。

很明显，两个非常重要的时间点：一个是数据产生的时间，把它叫作“事件时间”（Event Time）；另一个是数据真正被处理的时刻，叫作“处理时间”（Processing Time）。 窗口操作，到底是以那种时间作为衡量标准，就是所谓的“时间语义”（Notions of Time）。由于分布式系统中网络传输的延迟和时钟漂移，处理时间相对事件发生的时间会有所滞后。

**1. 处理时间（Processing Time）**

处理时间，是指执行处理操作的机器的系统时间。 在这种时间语义下处理窗口非常简单粗暴，不需要各节点间进行协调同步，也不需要考虑数据在流中的位置，简单来说就是“我的地盘听我的”。所以处理时间是最简单的时间语义。

**2. 事件时间（Event Time）**

事件时间，指每个事件在对应的设备上发生的时间，也就是数据生成的时间。 数据一旦产生，这个时间自然就确定了，所以它可以作为一个属性嵌入到数据中。这其实就是这条数据记录的“时间戳”（Timestamp）。

在事件时间语义下，对于时间的衡量，就不看任何机器的系统时间了，而是依赖于数据本身。但由于分布式系统中网络传输延迟的不确定性，实际应用中我们要面对的数据流往往是乱序的。这种情况下，就不能简单把数据自带的时间戳当作时钟，需要用另外的标志来表示事件时间进展，在 Flink中把它叫作事件时间的“水位线”（Watermarks）。

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

#### 6.1.2 哪种时间语义更重要 ####

![d65d74c800054ecb8b28e7ddbaff8f8a.png][]

**2. 数据处理系统中的时间语义**

在计算机系统实际应用中，事件时间语义更常见。一般情况下，业务日志数据中都会记录数据生成的时间戳（timestamp），它就可作为事件时间的判断基础。

**3. 两种时间语义的对比**

通常来说，处理时间是我们计算效率的衡量标准，而事件时间更符合业务计算逻辑。所以更多时候使用事件时间；

对于处理时间而言，由于没有任何附加考虑，数据一来就直接处理，因此这种方式的流处理延迟降到最低， 效率最高。

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

### 6.2 水位线（Watermark） ###

#### 6.2.1 什么是水位线 ####

事件时间语义下，不再依赖系统时间，而是基于数据自带时间戳去定义一个时钟， 用来表示当前时间进展。于是每个并行子任务都会有一个自己的逻辑时钟，它的前进是靠数据的时间戳来驱动的。

可以把时钟以数据的形式传递出去，告诉下游任务当前时间的进展；而这个时钟的传递不会因为窗口聚合之类的运算而停滞。一种简单的想法是，在数据流中加入一个时钟标记，记录当前事件时间；这个标记可以广播到下游，当下游任务收到这个标记，就更新自己的时钟。

由于类似于水流中用来做标志的记号，在Flink 中，这种用来衡量事件时间（Event Time）进展的标记，被称作“水位线”（Watermark）。

具体实现上，水位线可看作一条特殊的数据记录，是插入到数据流中的一个标记点， 主要内容就是一个时间戳，用来指示当前的事件时间。而它插入流中的位置，就应该是在某个数据到来之后；这样就可以从这个数据中提取时间戳，作为当前水位线的时间戳。

![db30bb9810ef4e98be86db5b56b0aade.png][]

**1. 有序流中的水位线**

理想状态下，数据按照生成的先后顺序、排好队进入流中；

而实际应用中， 如果当前数据量非常大，可能会有很多数据的时间戳是相同的，这时每来一条数据就提取时间戳、插入水位线就做了大量的无用功。

所以为了提高效率，一般会每隔一段时间生成一个水位 线，这个水位线的时间戳，就是当前最新数据的时间戳，如图 6-4 所示。所以这时的水位线， 其实就是有序流中的一个周期性出现的时间标记。

![e971a5d5da1b40fd98add02f687384a5.png][]

**2. 乱序流中的水位线**

分布式系统中，数据在节点间传输，会因网络传输延迟的不确定性，导致顺序发生改变，这就是“乱序数据”。

![a8cbf87b04fd48fc994432a57bc1d153.png][]对于连续数据流，插入新的水位线时，要先判断时间戳是否比之前大，否则不再生成新的水位线，如图 6-6 所示。就是说，只有数据时间戳比当前时钟大，才推动时钟前进，这时才插入水位线。

![2bf91354d8894212bf555ce6f37e850f.png][]  
如果考虑到大量数据同时到来的处理效率，同样可以周期性生成水位线。这时只需保存之前所有数据中最大时间戳，需要插入水位线时，就直接以它作为时间戳生成新的水位线，如图 6-7 所示。

![8e0c6738ce2c4f95aabf336961c07a54.png][]

为让窗口正确收集到迟到的数据，可以等上 2 秒；也就是用当前已有数据的最大时间戳减去 2 秒，就是要插入的水位线的时间戳，如图 6-8 所示。

![9015ae4d98654663bc3206c2569db090.png][]

如果仔细观察会看到，这种“等 2 秒”的策略并不能处理所有的乱序数据。所以可以多等几秒，把时钟调得更慢。最终目的，就是让窗口能够把所有迟到数据都收进来，得到正确计算结果。对应到水位线上，其实就是要保证，当前时间已经进展到了这个时间戳，在这之后不可能再有迟到数据来了。

**3. 水位线的特性**

水位线就代表了当前的事件时间时钟，且可以在数据的时间戳基础上加些延迟来保证不丢数据，这一点对于乱序流的正确处理非常重要。

水位线的特性：

⚫ 水位线是插入到数据流中的一个标记，是一个特殊数据

⚫ 水位线主要内容是一个时间戳，用来表示当前事件时间的进展

⚫ 水位线是基于数据的时间戳生成的

⚫ 水位线的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进

⚫ 水位线可通过设置延迟，来保证正确处理乱序数据

⚫ 一个水位线 Watermark(t)，表示在当前流中事件时间已经达到了时间戳 t, 这代表 t 之前的所有数据都到齐，之后流中不会出现时间戳 t’ ≤ t 的数据 水位线是 Flink 流处理中保证结果正确性的核心机制，它往往会跟窗口一起配合，完成对乱序数据的正确处理。

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

#### 6.2.3 如何生成水位线 ####

**1. 生成水位线的总体原则**

希望计算结果能更加准确，可将水位线的延迟设置得更高，等待的时间越长，也就越不容易漏掉数据。不过这样处理的实时性降低了，可能为极少数的迟到数据增加了不必要的延迟。

希望处理得更快、实时性更强，可将水位线延迟设得更低。 可能迟到数据在水位线之后才到达，会导致窗口遗漏数据，计算结果不准确。

对这些 “漏网之鱼”，Flink 另外提供了窗口处理迟到数据的方法，我们会在后面介绍。

当然，如果一味地追求处理速度，可直接用处理时间语义。

Flink 中的水位线，是流处理中对低延迟和结果正确性的一个权衡机制，且把控制的权力交给程序员，我们可以在代码中定义水位线的生成策略。

**2. 水位线生成策略（Watermark Strategies）**

在Flink的DataStreamAPI中，有单 独 用 于 生 成 水 位 线 的 方 法 ：assignTimestampsAndWatermarks()，它主要用来为流中的数据分配时间戳，并生成水位线来指 示事件时间。

**3. Flink 内置水位线生成器**

// 1、有序流的水位线生成策略  特点是时间戳单调增长,不会出现迟到数据。这是周期性生成水位线的最简单的场景。简单说,直接拿当前最大的时间戳作为水位线。
        stream.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps[Event]()
          .withTimestampAssigner(
            new SerializableTimestampAssigner[Event] {
              override def extractTimestamp(t: Event, l: Long): Long = t.timestamp
            }
          ))
    
        // 2、乱序流的水位线生成策略  乱序流中需等待迟到数据,须设置固定量延迟时间。这时生成水位线时间戳,是当前数据流中最大时间戳减去延迟,相当于把表调慢,当前时钟会滞后于数据最大时间戳。
        // 事实上,有序流水位线生成器本质上和乱序流一样，相当于延迟设为0的乱序流水位线生成器,两者完全等同
        stream.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness[Event](Duration.ofSeconds(2))
          .withTimestampAssigner(
            new SerializableTimestampAssigner[Event] {
              override def extractTimestamp(t: Event, l: Long): Long = t.timestamp
            }
          ))
    
        // 3、自定义周期性水位线生成策略
        stream.assignTimestampsAndWatermarks(new WatermarkStrategy[Event] {
    
          override def createTimestampAssigner(context: TimestampAssignerSupplier.Context): TimestampAssigner[Event] = {
            new SerializableTimestampAssigner[Event] {
              override def extractTimestamp(t: Event, l: Long): Long = t.timestamp
            }
          }
    
          override def createWatermarkGenerator(context: WatermarkGeneratorSupplier.Context): WatermarkGenerator[Event] = {
            new WatermarkGenerator[Event] {
              // 定义一个延迟时间
              val delay = 5000L
              // 定义属性保存最大时间戳
              var maxTs = Long.MinValue + delay + 1
    
              override def onEvent(t: Event, l: Long, watermarkOutput: WatermarkOutput): Unit = {
                maxTs = math.max(maxTs, t.timestamp)
              }
    
              override def onPeriodicEmit(watermarkOutput: WatermarkOutput): Unit = {
                val watermark = new Watermark(maxTs - delay - 1)
                watermarkOutput.emitWatermark(watermark)
              }
            }
          }
        })

4.在自定义数据源中发送水位线

import com.wsq.chapter05_operator.Event
    import org.apache.flink.streaming.api.functions.source.SourceFunction
    import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
    import org.apache.flink.streaming.api.watermark.Watermark
    
    import java.util.Calendar
    import scala.util.Random
    
    /*
    创建自定义数据源，实现 SourceFunction 接口。重写两个关键方法：
    ⚫ run()：使用运行时上下文对象（SourceContext）向下游发送数据；
    ⚫ cancel()：通过标识位控制退出循环，达到中断数据源的效果。
     */
    
    // 实现 SourceFunction 接口，接口中的泛型是自定义数据源的类型
    class ClickSource_TS extends SourceFunction[Event] {
      // 标志位，控制循环的退出
      var running = true
    
      //重写run()，使用上下文对象sourceContext调用 collect 方法
      override def run(ctx: SourceContext[Event]): Unit = {
        // 实例化一个随机数发生器
        val random = new Random
        // 供随机选择的users数组
        val users = Array("Mary", "Bob", "Alice", "Cary")
        // 供随机选择的url数组
        val urls = Array("./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2")
        // 通过while发送数据，running默认为true，会一直发送数据
        while (running) {
          val event = Event(users(random.nextInt(users.length)), urls(random.nextInt(urls.length)), Calendar.getInstance.getTimeInMillis)
          // 为要发送的数据分配时间戳⭐
          ctx.collectWithTimestamp(event, event.timestamp)
    
          // 向下游直接发送水位线⭐
          ctx.emitWatermark(new Watermark(event.timestamp - 1L))
    
          // 调用ctx的collect方法向下游发送数据
          ctx.collect(event)
          // 隔1秒生成一个点击事件，方便观测
          Thread.sleep(2000)
        }
      }
    
      //通过将 running 置为 false 终止数据发送循环
      override def cancel(): Unit = running = false
    }

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

#### 6.2.4 水位线的传递 ####

在“重分区”（redistributing）传输模式下，一个任务可能会收到来自不同分区上游子任务的数据。而不同分区的子任务时钟并不同步，所以同一时刻发给下游任务的水位线可能并不相同。这说明上游各个分区处理有快有慢，进度各不相同，这时应以最慢的那个时钟，也就是最小水位线为准。

![8be8f395afc848e684ee97a9f9bb1525.png][]水位线在上下游任务间的传递，巧妙地避免了分布式系统中没有统一时钟的问题，每个任务都以“处理完之前所有数据”为标准来确定自己的时钟，保证窗口处理的结果是正确的。对于有多条流合并之后进行处理的场景，水位线传递的规则类似。

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

### 6.3 窗口（Window） ###

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

Flink是流式计算引擎，是处理无界数据流的，数据源源不断。

想更加方便处理无界流，一种方式就是将无限数据切成有限“数据块”处理，这就是“窗口”（Window）。在 Flink中, 窗口是处理无界流的核心。

![a47b40d8a7854f91bc19f792dc48eae8.png][]窗口包含起始时间、不包含结束时间。

对处理时间下的窗口而言，这样理解似乎没什么问题。

然而采用事件时间语义，由于有乱序数据，需设一个延迟时间来等所有数据到齐。如上面的例子中，可以设置延迟时间为 2 秒，如图 6-12 所示，这样 0~10 秒的窗口会在时间戳为12 的数据到来之后，才真正关闭计算输出结果，这样就可以包含迟到的9秒数据。

![bbfa7c5608b24b719d4a7ebbcc345a3a.png][]但是，0~10 秒的窗口不光包含了迟到的 9 秒数据，连 11 秒和 12 秒的数据也包含进去了。

所以在 Flink 中，窗口其实并不是一个“框”，流进来的数据被框住了就只能进这一个窗口。相比之下，应把窗口理解成“桶”，如图 6-13 所示。在 Flink 中，窗口可以把流切割成有限大小的多个“存储桶”（bucket)；每个数据都会分发到对应的桶中，当到达窗口结束时间时，就对每个桶中收集的数据进行计算处理。

![9c8d68d543f74698abbcfb3537e04906.png][]注意，Flink窗口不是静态准备好的，而是动态创建——当有落在这个 窗口区间范围的数据达到时，才创建对应窗口。

另外，这里认为到达窗口结束时间时， 窗口就触发计算并关闭，事实上“触发计算”和“窗口关闭”两个行为也可以分开。

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

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

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

窗口是截取有界数据一种方式，所以窗口一个非常重要的信息就是“怎样截取数据”。

也就是以何标准来开始和结束截取，我们把它叫作窗口的“驱动类型”。

最容易想到是按时间段截取数据，这种窗口叫作“时间窗口”（Time Window）。

这最常见，之前所举的例子都是时间窗口。

除由时间驱动外， 窗口也可由数据驱动，按固定数据个数，截取一段数据集，这种窗口叫作“计数窗口”（Count Window），如图6-14所示。

![ab084dd1c6544cad942e2e051caefbe0.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;
    }

很明显，窗口中数据最大允许时间戳是end - 1，这代表了定义的窗口时间范围都是左闭右开的区间\[start，end)。

时间范围都是左闭右开的区间\[start，end)。

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

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

每个窗口截取数据的个数，就是窗口的大小。

计数窗口相比时间窗口就更简单，只需指定窗口大小，就可以把数据分配到对应的窗口中。

在 Flink 内部并没有对应的类来表示计数窗口，底层是通过“全局窗口”（Global Window）来实现。

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

时间窗口和计数窗口，只是对窗口的大致划分；在具体应用时，还需定义更精细的规则，来控制数据应该划分到哪个窗口中。

不同的分配数据的方式，就有不同的功能应用。

根据分配数据的规则，窗口的具体实现可以分为4 类：滚动窗口、滑动窗口、会话窗口，以及全局窗口。

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

滚动窗口有固定的大小，是一种对数据进行“均匀切片”的划分方式。

窗口间没有重叠， 也不会有间隔，是“首尾相接”的状态。

如果把多个窗口的创建，看作一个窗口的运动， 那就好像它在不停地向前“翻滚”一样。这是最简单的窗口形式，之前所举的例子都是滚动窗口。 滚动窗口可以基于时间定义，也可以基于数据个数定义；

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

![7d40d8fc732c40c28c281ec1364d851b.png][]

滚动窗口应用广泛，它可以对每个时间段做聚合统计，很多 BI 分析指标都可以用它来实现。

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

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

所以定义滑动窗口的参数有两个：除去窗口大小（window size）外，还有一个“滑动步长”（window slide），它就代表窗口计算的频率。同样，滑动窗口可以基于时间定义，也可以基于数据个数定义。

![3b3a8567932d474ca475b7406e6119cd.png][]

可以看到，当滑动步长小于窗口大小时，滑动窗口就会出现重叠，这时数据也可能会被同时分配到多个窗口中。

而具体的个数，就由窗口大小和滑动步长的比值（size/slide）来决定。

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

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

基于“会话”（session）来对数据进行分组。这里的会话类似 Web 应用中 session 的概念，不过并不表示两端的通讯过程，而是借用会话超时失效的机制来描述窗口。

与滑动窗口和滚动窗口不同，会话窗口只能基于时间来定义。

对于会话窗口而言，最重要的参数是会话超时时间的长度（size），也就是两个会话窗口之间的最小距离。

如果相邻两个数据到来的时间间隔（Gap）小于指定的大小（size），那说明还在保持会话，就属于同一个窗口；如果 gap 大于 size，那么新来的数据就属于新的会话窗口，而前一个窗口就应关闭。在具体实现上，可以设置静态固定的大小（size），也可以通过一个自定义的提取器（gap extractor）动态提取最小间隔 gap的值。

![d5b825084cd94c3fb80372bf233adc52.png][]在一些类似保持会话的场景下，往往可以使用会话窗口来进行数据的处理统计。

**（4）全局窗口（Global Windows）**

这种窗口全局有效，会把相同 key 的所有数据都分配到同一个窗口中。

无界流的数据永无止尽，所以这种窗口也没有结束的时候，默认不会做触发计算。如果希望它能对数据进行计算处理，还需自定义“触发器”（Trigger）。

![94e81aaaec5845f5aa20cb4b0fb063a4.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。所以在实际应用中一般不推荐使用这种方式。

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()方法。这个方法需传入一个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。第一个参数还是窗口大小，第二个参数则表示窗口起始点的偏移量。

（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[(String, Long)] {
     override def extract(element: (String, Long)) { 
    // 提取 session gap 值返回, 单位毫秒
     element._1.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. 计数窗口***

计数窗口本身底层基于全局窗口（Global Window）实现。

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，如 图 6-21 所示。

![c5fc953828ae48beb5c121dbbe7c1482.png][]

窗口函数定义了要对窗口中收集的数据做的计算操作。

根据处理的方式可以分为两类：增量聚合函数和全窗口函数。

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

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

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

**（1）归约函数（ReduceFunction）**

最基本的聚合方式就是归约（reduce）。

在基本转换的聚合算子中介绍过 reduce 的用法，窗口的归约聚合也非常类似，就是将窗口中收集到的数据两两进行归约。

当我们进行流处理时，就是要保存一个状态；每来一个新的数据，就和之前的聚合状态做归约，这样就实现了增量式的聚合。 窗口函数中也提供了 ReduceFunction：只要基于 WindowedStream 调用.reduce()方法，然后传入 ReduceFunction 作为参数，就可以指定以归约两个元素的方式去对窗口中数据进行聚合了。

这里的 ReduceFunction 其实与简单聚合时用到的 ReduceFunction 是同一个函数类接口， 所以使用方式是完全一样的。 下面是使用 ReduceFunction 进行增量聚合的代码示例。

import com.atguigu.chapter05.ClickSource
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    object WindowReduceExample {
     def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(1)
     env
     .addSource(new ClickSource)
     // 数据源中的时间戳是单调递增的，所以使用下面的方法，只需要抽取时间戳就好了
     // 等同于最大延迟时间是 0 毫秒
     .assignAscendingTimestamps(_.timestamp)
     .map(r => (r.user, 1L))
     // 使用用户名对数据流进行分组
     .keyBy(_._1)
     // 设置 5 秒钟的滚动事件时间窗口
     .window(TumblingEventTimeWindows.of(Time.seconds(5)))
     // 保留第一个字段，针对第二个字段进行聚合
     .reduce((r1, r2) => (r1._1, r1._2 + r2._2))
     .print()
     env.execute()
     }
    }

运行结果如下：

(Bob,1)
    (Alice,2)
    (Mary,2)
    ……

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

**（2）聚合函数（AggregateFunction）**

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 则是我们进行聚合的中间状态类型；而输出类型就最终计算结果的类型。

AggregateFunction 接口中有四个方法：

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

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

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

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

下面一个具体例子。计算一下 PV/UV 这个比值，来表示“人均重复访问量”， 也就是平均每个用户会访问多少次页面，这在一定程度上代表了用户的黏度。 代码实现如下：

package com.atguigu.chapter06
    import com.atguigu.chapter05.{ClickSource, Event}
    import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    
    object AggregateFunctionExample {
     def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(1)
     
    env
     .addSource(new ClickSource)
     .assignAscendingTimestamps(_.timestamp)
     // 通过为每条数据分配同样的 key，来将数据发送到同一个分区
     .keyBy(_ => "key")
     .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(2)))
     .aggregate(new AvgPv)
     .print()
     env.execute()
     }
     class AvgPv extends AggregateFunction[Event, (Set[String], Double), Double] {
     // 创建空累加器，类型是元组，元组的第一个元素类型为 Set 数据结构，用来对用户名进行去重
     // 第二个元素用来累加 pv 操作，也就是每来一条数据就加一
     override def createAccumulator(): (Set[String], Double) = (Set[String](), 0L)
     // 累加规则
     override def add(value: Event, accumulator: (Set[String], Double)): 
    (Set[String], Double) = (accumulator._1 + value.user, accumulator._2 + 1L)
     // 获取窗口关闭时向下游发送的结果
     override def getResult(accumulator: (Set[String], Double)): Double = 
    accumulator._2 / accumulator._1.size
     // merge 方法只有在事件时间的会话窗口时，才需要实现，这里无需实现。
     override def merge(a: (Set[String], Double), b: (Set[String], Double)): 
    104
    (Set[String], Double) = ???
     }
    }

输出结果如下：

1.0
    1.6666666666666667
    ……

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

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

***2. 全窗口函数（Full Window Functions）***

窗口操作中的另一大类就是全窗口函数。与增量聚合函数不同，全窗口函数需先收集窗口中的数据，并在内部缓存起来，等到窗口要输出结果的时候再取出数据进行计算。

在 Flink 中，全窗口函数也有两种：WindowFunction 和 ProcessWindowFunction。

**（1）窗口函数（WindowFunction）**

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

stream
     .keyBy(<key selector>)
     .window(<window assigner>)
     .apply(new MyWindowFunction())

这个类中可以获取到包含窗口所有数据的可迭代集合（Iterable），还可以拿到窗口 （Window）本身的信息。

WindowFunction 接口在源码中实现如下：

public interface WindowFunction<IN, OUT, KEY, W extends Window> extends Function, 
    Serializable {
    void apply(KEY key, W window, Iterable<IN> input, Collector<OUT> out) throws 
    Exception;
    }

当窗口到达结束时间需要触发计算时，就会调用这里的 apply 方法。我们可以从 input 集 合中取出窗口收集的数据，结合 key 和 window 信息，通过收集器（Collector）输出结果。

这里 Collector 的用法，与 FlatMapFunction 中相同。 不过我们也看到了，WindowFunction 能提供的上下文信息较少，也没有更高级的功能。 事实上，它的作用可以被 ProcessWindowFunction 全覆盖，所以之后可能会逐渐弃用。一般在 实际应用，直接使用 ProcessWindowFunction 就行。

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

**（2）处理窗口函数（ProcessWindowFunction）**

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

具体使用跟 WindowFunction 非常类似，我们可以基于 WindowedStream 调用 process()方 法，传入一个 ProcessWindowFunction 的实现类。下面是一个电商网站统计每小时 UV 的例子：

import com.atguigu.chapter05.{ClickSource, Event}
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    import java.sql.Timestamp
    import scala.collection.mutable.Set
    
    object UvCountByWindowExample {
     def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(1)
     env
     .addSource(new ClickSource)
     .assignAscendingTimestamps(_.timestamp)
     // 为所有数据都指定同一个 key，可以将所有数据都发送到同一个分区
     .keyBy(_ => "key")
     .window(TumblingEventTimeWindows.of(Time.seconds(10)))
     .process(new UvCountByWindow)
     .print()
    106
     env.execute()
     }
     // 自定义窗口处理函数
     class UvCountByWindow extends ProcessWindowFunction[Event, String, String, 
    TimeWindow] {
     override def process(key: String, context: Context, elements: Iterable[Event], 
    out: Collector[String]): Unit = {
     // 初始化一个 Set 数据结构，用来对用户名进行去重
     var userSet = Set[String]()
     // 将所有用户名进行去重
     elements.foreach(userSet += _.user)
     // 结合窗口信息，包装输出内容
     val windowStart = context.window.getStart
     val windowEnd = context.window.getEnd
     out.collect(" 窗 口 ： " + new Timestamp(windowStart) + "~" + new 
    Timestamp(windowEnd) + "的独立访客数量是：" + userSet.size)
     }
     }
    }

输出结果形式如下：

窗口：...~...的独立访客数量是：2
    窗口：...~...的独立访客数量是：3
    ……

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

***3. 增量聚合和全窗口函数的结合使用***

我们已经了解了 Window API 中两类窗口函数的用法，下面做个简单的总结。

增量聚合函数处理计算会更高效。增量聚合相当于把计算量“均摊”到了窗口收集数据的过程中，自然就会比全窗口聚合更加高效、输出更加实时。

全窗口函数的优势在于提供了更多的信息，可以认为是更加“通用”的窗口操作，窗口计算更加灵活，功能更加强大。 所以在实际应用中，我们往往希望兼具这两者的优点，把它们结合在一起使用。

Flink 的 Window API 就给我们实现了这样的用法。

我们之前在调用 WindowedStream 的 reduce()和 aggregate()方法时，只是简单地直接传入了一个 ReduceFunction 或 AggregateFunction 进行增量聚合。除此之外，其实还可以传入第二个参数：一个全窗口函数，可以是 WindowFunction 或者 ProcessWindowFunction。

// ReduceFunction 与 WindowFunction 结合
    public <R> SingleOutputStreamOperator<R> reduce(
     ReduceFunction<T> reduceFunction, WindowFunction<T, R, K, W> function) 
    // ReduceFunction 与 ProcessWindowFunction 结合
    public <R> SingleOutputStreamOperator<R> reduce(
     ReduceFunction<T> reduceFunction, ProcessWindowFunction<T, R, K, W> 
    function)
    // AggregateFunction 与 WindowFunction 结合
    public <ACC, V, R> SingleOutputStreamOperator<R> aggregate(
     AggregateFunction<T, ACC, V> aggFunction, WindowFunction<V, R, K, W> 
    windowFunction)
    // AggregateFunction 与 ProcessWindowFunction 结合
    public <ACC, V, R> SingleOutputStreamOperator<R> aggregate(
     AggregateFunction<T, ACC, V> aggFunction,
     ProcessWindowFunction<V, R, K, W> windowFunction)

这样调用的处理机制是：基于第一个参数（增量聚合函数）来处理窗口数据，每来一个数据就做一次聚合；等到窗口需要触发计算时，则调用第二个参数（全窗口函数）的处理逻辑输出结果。

需要注意的是，这里的全窗口函数就不再缓存所有数据了，而是直接将增量聚合函数的结果拿来当作了 Iterable 类型的输入。一般情况下，这时的可迭代集合中就只有一个元素。

下面举一个具体的实例来说明。我们这里统计 10 秒钟的 url 浏览量，每 5 秒钟更新 一次；另外为了更加清晰地展示，还应该把窗口的起始结束时间一起输出。我们可以定义滑动窗口，并结合增量聚合函数和全窗口函数来得到统计结果。

import com.atguigu.chapter05.Event
    import com.atguigu.chapter06.EmitWatermarkInSourceFunction.ClickSource
    import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
    import 
    org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    object UrlViewCountExample {
     def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(1)
     env
     .addSource(new ClickSource)
     .assignAscendingTimestamps(_.timestamp)
     // 使用 url 作为 key 对数据进行分区
     .keyBy(_.url)
     .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
     // 注意这里调用的是 aggregate 方法
     // 增量聚合函数和全窗口聚合函数结合使用
     .aggregate(new UrlViewCountAgg, new UrlViewCountResult)
     .print()
     env.execute()
     }
     class UrlViewCountAgg extends AggregateFunction[Event, Long, Long] {
     override def createAccumulator(): Long = 0L
     // 每来一个事件就加一
     override def add(value: Event, accumulator: Long): Long = accumulator + 1L
     // 窗口闭合时发送的计算结果
     override def getResult(accumulator: Long): Long = accumulator
     override def merge(a: Long, b: Long): Long = ???
     }
     class UrlViewCountResult extends ProcessWindowFunction[Long, UrlViewCount, 
    String, TimeWindow] {
     // 迭代器中只有一个元素，是增量聚合函数在窗口闭合时发送过来的计算结果
    109
     override def process(key: String, context: Context, elements: Iterable[Long], 
    out: Collector[UrlViewCount]): Unit = {
     out.collect(UrlViewCount(
     key,
     elements.iterator.next(),
     context.window.getStart,
     context.window.getEnd
     ))}
     }
     case class UrlViewCount(url: String, count: Long, windowStart: Long, windowEnd: 
    Long)
    }

这里为了方便处理，单独定义了一个样例类 UrlViewCount 来表示聚合输出结果的数据类型，包含了 url、浏览量以及窗口的起始结束时间。

用一个 AggregateFunction 来实现增量聚合，每来一个数据就计数加1；得到的结果交给 ProcessWindowFunction，结合窗口信息包装成我们想要的 UrlViewCount，最终输出统计结果。

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

#### 6.3.6 其他 API ####

对于一个窗口算子而言，窗口分配器和窗口函数是必不可少的。除此之外，Flink 还提供 了其他一些可选的 API，让我们可以更加灵活地控制窗口行为。

***1. 触发器（Trigger）***

触发器主要是用来控制窗口什么时候触发计算。所谓的“触发计算”，本质上就是执行窗口函数，所以可以认为是计算得到结果并输出的过程。

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

stream.keyBy(...)
     .window(...)
     .trigger(new MyTrigger())

Trigger 是窗口算子的内部属性，每个窗口分配器（WindowAssigner）都会对应一个默认的触发器；对于 Flink 内置的窗口类型，它们的触发器都已经做了实现。例如，所有事件时间窗口，默认的触发器都是 EventTimeTrigger；类似还有 ProcessingTimeTrigger 和 CountTrigger。 所以一般情况下是不需要自定义触发器的，不过我们依然有必要了解它的原理。

Trigger 是一个抽象类，自定义时必须实现下面四个抽象方法：

⚫ onElement()：窗口中每到来一个元素，都会调用这个方法。

⚫ onEventTime()：当注册的事件时间定时器触发时，将调用这个方法。

⚫ onProcessingTime ()：当注册的处理时间定时器触发时，将调用这个方法。

⚫ clear()：当窗口关闭销毁时，调用这个方法。一般用来清除自定义的状态。

可以看到，除了 clear()比较像生命周期方法，其他三个方法都是对某种事件的响应。

onElement()是对流中数据元素到来的响应；而另两个则是对时间的响应。这几个方法的参数中都有一个“触发器上下文”（TriggerContext）对象，可以用来注册定时器回调（callback）。这里提到的“定时器”（Timer），其实就是我们设定的一个“闹钟”，代表未来某个时间点会执行的事件；当时间进展到设定的值时，就会执行定义好的操作。

很明显，对于时间窗口 （TimeWindow）而言，就应该是在窗口的结束时间设定了一个定时器，这样到时间就可以触发 窗口的计算输出了。关于定时器的内容，我们在后面讲解处理函数（process function）时还会提到。

上面的前三个方法可以响应事件，那它们又是怎样跟窗口操作联系起来的呢？

这就需要了解一下它们的返回值。这三个方法返回类型都是 TriggerResult，这是一个枚举类型（enum）， 其中定义了对窗口进行操作的四种类型。

⚫ CONTINUE（继续）：什么都不做

⚫ FIRE（触发）：触发计算，输出结果

⚫ PURGE（清除）：清空窗口中的所有数据，销毁窗口

⚫ FIRE\_AND\_PURGE（触发并清除）：触发计算输出结果，并清除窗口

可以看到，Trigger 除了可以控制触发计算，还可以定义窗口什么时候关闭（销毁）。 上面的四种类型，其实也就是这两个操作交叉配对产生的结果。一般我们会认为，到了窗口的结束时间，就会触发计算输出结果，然后关闭窗口——似乎这两个操作应该是同时发生的； 但 TriggerResult 的定义告诉我们，两者可以分开。

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

***2. 移除器（Evictor）***

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

stream.keyBy(...)
     .window(...)
     .evictor(new MyEvictor())

Evictor 接口定义了两个方法：

⚫ evictBefore()：定义执行窗口函数之前的移除数据操作

⚫ evictAfter()：定义执行窗口函数之后的以处数据操作

默认情况下，预实现的移除器都是在执行窗口函数（window fucntions）之前移除数据。

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

***3. 允许延迟（Allowed Lateness）***

在事件时间语义下，窗口中可能会出现数据迟到的情况。迟到数据默认会被直接丢弃，不会进入窗口进行统计计算。这样可能会导致统计结果不准确。

为了解决迟到数据的问题，Flink 提供了一个特殊的接口，可以为窗口算子设置一个“允许的最大延迟”（Allowed Lateness）。也就是说，我们可以设定允许延迟一段时间，在这段时间内，窗口不会销毁，继续到来的数据依然可以进入窗口中并触发计算。直到水位线推进到了 窗口结束时间 + 延迟时间，才真正将窗口的内容清空，正式关闭窗口。

基于 WindowedStream 调用 allowedLateness()方法，传入一个 Time 类型的延迟时间，就可以表示允许这段时间内的延迟数据。

stream.keyBy(...)
     .window(TumblingEventTimeWindows.of(Time.hours(1)))
     .allowedLateness(Time.minutes(1))

从这里可以看到，窗口的触发计算（Fire）和清除（Purge）操作确实可以分开。不过在默认情况下，允许的延迟是 0，这样一旦水位线到达了窗口结束时间就会触发计算并清除窗口，两个操作看起来就是同时发生。当窗口被清除（关闭）之后，再来的数据就会被丢弃。

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

***4. 将迟到的数据放入侧输出流***

Flink 还提供了另外一种方式处理迟到数据。我们可以将未收入窗口的迟到数据，放入“侧输出流”（side output）进行另外的处理。所谓的侧输出流，相当于是数据流的一个“分支”， 这个流中单独放置那些本该被丢弃的数据。

基于 WindowedStream 调用 sideOutputLateData() 方法，就可以实现这个功能。方法需传入一个“输出标签”（OutputTag），用来标记分支的迟到数据流。因为保存的就是流中的原始数据，所以 OutputTag 的类型与流中数据类型相同。

val stream = env.addSource(new ClickSource)
    val outputTag = new OutputTag[Event]("late")
    stream.keyBy("user")
     .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .sideOutputLateData(outputTag)

将迟到数据放入侧输出流之后，还应该可以将它提取出来。基于窗口处理完成之后的 DataStream，调用 getSideOutput()方法，传入对应的输出标签，就可以获取到迟到数据所在的流了。

val winAggStream = stream.keyBy(...)
     .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .sideOutputLateData(outputTag)
    .aggregate(new MyAggregateFunction)
    val lateStream = winAggStream.getSideOutput(outputTag)

这里注意，getSideOutput()是 DataStream 的方法，获取到的侧输出流数据类型应该和 OutputTag 指定的类型一致，与窗口聚合之后流中的数据类型可以不同。

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

#### **6.3.7 窗口的生命周期** ####

熟悉了窗口 API 的使用，再回头梳理一下窗口本身的生命周期，这也是对窗口所有操作的总结。

***1. 窗口的创建***

窗口的类型和基本信息由窗口分配器（window assigners）指定，但窗口不会预先创建好， 而是由数据驱动创建。当第一个应该属于这个窗口的数据元素到达时，就会创建对应的窗口。

***2. 窗口计算的触发***

除了窗口分配器，每个窗口还会有自己的窗口函数（window functions）和触发器（trigger）。 窗口函数可以分为增量聚合函数和全窗口函数，主要定义了窗口中计算的逻辑；而触发器则是指定调用窗口函数的条件。

对于不同的窗口类型，触发计算的条件也会不同。Flink 预定义的窗口类型都有对应内置的触发器。

***3. 窗口的销毁***

一般情况下，当时间达到了结束点，就会直接触发计算输出结果、进而清除状态销毁窗口。 这时窗口的销毁可以认为和触发计算是同一时刻。这里需要注意，Flink 中只对时间窗口 （TimeWindow）有销毁机制；由于计数窗口（CountWindow）是基于全局窗口（GlobalWindw） 实现的，而全局窗口不会清除状态，所以就不会被销毁。

在特殊的场景下，窗口的销毁和触发计算会有所不同。事件时间语义下，如果设置了允许延迟，那么在水位线到达窗口结束时间时，仍然不会销毁窗口；窗口真正被完全删除的时间点， 是窗口的结束时间加上用户指定的允许延迟时间。

***4. 窗口 API 调用总结***

到目前为止，已经彻底明白了 Flink 中窗口的概念和 Window API 的调用，再用一张图做一个完整总结，如图 6-22 所示。

![9c0bcefda1b54866a6c75d21c36c342a.png][]Window API 首先按照时候按键分区分成两类。keyBy()之后的 KeyedStream，可以调用window()方法声明按键分区窗口（Keyed Windows）；而如果不做 keyBy()，DataStream 也可以直接调用 windowAll()声明非按键分区窗口。之后的方法调用就完全一样了。

接下来先是通过 window()/windowAll()方法定义窗口分配器，得到 WindowedStream； 然 后通过 各 种 转 换 方 法 （ reduce()/aggregate()/apply()/process() ） 给出窗口函数 (ReduceFunction/AggregateFunction/ProcessWindowFunction)，定义窗口的具体计算处理逻辑， 转换之后重新得到 DataStream。这两者必不可少，是窗口算子（WindowOperator）最重要的组 成部分。

此外，在这两者之间，还可以基于 WindowedStream 调用.trigger()自定义触发器、调用.evictor()定义移除器、调用 allowedLateness()指定允许延迟时间、调用 sideOutputLateData() 将迟到数据写入侧输出流，这些都是可选的 API，一般不需要实现。而如果定义了侧输出流， 可以基于窗口聚合之后的 DataStream 调用 getSideOutput()获取侧输出流。

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

### 6.4 迟到数据的处理 ###

指某水位线之后到来的数据，它的时间戳是在水位线之前的。所以只有在事件时间语义下，讨论迟到数据的处理才有意义。

#### 6.4.1 设置水位线延迟时间 ####

水位线是事件时间的进展，它是整个应用的全局逻辑时钟。水位线生成之后，会随着数据在任务间流动，从而给每个任务指明当前的事件时间。所以从这个意义上讲，水位线是一 个覆盖万物的存在，它并不只针对事件时间窗口有效。

水位线其实是所有事件时间定时器触发的判断标准。那么水位线的延迟，当然也就是全局时钟的滞后。

既然水位线这么重要，就不应该把它的延迟设置得太大，否则流处理的实时性就会降低。因为水位线的延迟主要是用来对付分布式网络传输导致的数据乱序，而网络传输的乱序程度一般并不会很大，大多集中在几毫秒至几百毫秒。所以实际应用中，往往会给水位线设置一个“能够处理大多数乱序数据的小延迟”，视需求一般设在毫秒~秒级。 当我们设置了水位线延迟时间后，所有定时器就都会按照延迟后的水位线来触发。如果一 个数据所包含的时间戳，小于当前的水位线，那么它就是所谓的“迟到数据”。

#### 6.4.2 允许窗口处理迟到数据 ####

除设置水位线延迟外，Flink的窗口也可以设置延迟时间，允许继续处理迟到数据的。 这种情况下，由于大部分乱序数据已被水位线的延迟等到了，所以往往迟到的数据不会太多。

这样，我们会在水位线到达窗口结束时间时，先快速地输出一个近似正确的计算结果； 然后保持窗口继续等到延迟数据，每来一条数据，窗口就会再次计算，并将更新后的结果输出。 这样就可以逐步修正计算结果，最终得到准确的统计值了。

这其实就是著名的 Lambda 架构。原先需要两套独立的系统来同时保证实时性和结果的最终正确性，如今 Flink 一套系统就全部搞定了。

#### 6.4.3 将迟到数据放入窗口侧输出流 ####

还可以用窗口的侧输出流，来收集关窗以后的迟到数据。这种方式是最后“兜底”方法， 只能保证数据不丢失；因为窗口已经真正关闭，所以是无法基于之前窗口的结果直接做更新的。

只能将之前的窗口计算结果保存下来，然后获取侧输出流中的迟到数据，判断数据所属的窗口，手动对结果进行合并更新。尽管烦琐，实时性也不够强，但能够保证最终结果正确。

所以总结起来，Flink 处理迟到数据，对于结果的正确性有三重保障：水位线的延迟，窗口允许迟到数据，将迟到数据放入窗口侧输出流。 我们可以回忆一下之前 6.3.5 小节统计每个 url 浏览次数的代码 UrlViewCountExample，稍作改进，增加处理迟到数据的功能。具体代码如下。

import java.time.Duration
    import com.atguigu.chapter05.Event
    import com.atguigu.chapter06.UrlViewCountExample.UrlViewCount
    import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, 
    WatermarkStrategy}
    import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
    import 
    org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    object ProcessLateDataExample {
     def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(1)
    115
     // 为了方便测试，读取 socket 文本流进行处理
     val stream = env
     .socketTextStream("localhost", 7777)
     .map(data => {
     val fields = data.split(",")
     Event(fields(0).trim, fields(1).trim, fields(2).trim.toLong)
     })
     // 方式一：设置 watermark 延迟时间，2 秒钟
     .assignTimestampsAndWatermarks(WatermarkStrategy
     // 最大延迟时间设置为 5 秒钟
     .forBoundedOutOfOrderness[Event](Duration.ofSeconds(2))
     .withTimestampAssigner(new SerializableTimestampAssigner[Event] {
     // 指定时间戳是哪个字段
     override def extractTimestamp(element: Event, recordTimestamp: Long): Long 
    = element.timestamp
     })
     )
     // 定义侧输出流标签
     val outputTag = OutputTag[Event]("late")
     val result = stream
     .keyBy(_.url)
     .window(TumblingEventTimeWindows.of(Time.seconds(10)))
     // 方式二：允许窗口处理迟到数据，设置 1 分钟的等待时间
     .allowedLateness(Time.minutes(1))
     // 方式三：将最后的迟到数据输出到侧输出流
     .sideOutputLateData(outputTag)
     .aggregate(new UrlViewCountAgg, new UrlViewCountResult)
     // 打印输出
     result.print("result")
    116
     result.getSideOutput(outputTag).print("late")
     // 为方便观察，可以将原始数据也输出
     stream.print("input")
     env.execute()
     }
     class UrlViewCountAgg extends AggregateFunction[Event, Long, Long] {
     override def createAccumulator(): Long = 0L
     // 每来一个事件就加一
     override def add(value: Event, accumulator: Long): Long = accumulator + 1L
     // 窗口闭合时发送的计算结果
     override def getResult(accumulator: Long): Long = accumulator
     override def merge(a: Long, b: Long): Long = ???
     }
     class UrlViewCountResult extends ProcessWindowFunction[Long, UrlViewCount, 
    String, TimeWindow] {
     // 迭代器中只有一个元素，是增量聚合函数在窗口闭合时发送过来的计算结果
     override def process(key: String, context: Context, elements: Iterable[Long], 
    out: Collector[UrlViewCount]): Unit = {
     out.collect(UrlViewCount(
     key,
     elements.iterator.next(),
     context.window.getStart,
     context.window.getEnd
     ))
     }}
    }

[bc9334f9d82e48eb93840930f1273941.png]: https://img-blog.csdnimg.cn/bc9334f9d82e48eb93840930f1273941.png
[d65d74c800054ecb8b28e7ddbaff8f8a.png]: https://img-blog.csdnimg.cn/d65d74c800054ecb8b28e7ddbaff8f8a.png
[db30bb9810ef4e98be86db5b56b0aade.png]: https://img-blog.csdnimg.cn/db30bb9810ef4e98be86db5b56b0aade.png
[e971a5d5da1b40fd98add02f687384a5.png]: https://img-blog.csdnimg.cn/e971a5d5da1b40fd98add02f687384a5.png
[a8cbf87b04fd48fc994432a57bc1d153.png]: https://img-blog.csdnimg.cn/a8cbf87b04fd48fc994432a57bc1d153.png
[2bf91354d8894212bf555ce6f37e850f.png]: https://img-blog.csdnimg.cn/2bf91354d8894212bf555ce6f37e850f.png
[8e0c6738ce2c4f95aabf336961c07a54.png]: https://img-blog.csdnimg.cn/8e0c6738ce2c4f95aabf336961c07a54.png
[9015ae4d98654663bc3206c2569db090.png]: https://img-blog.csdnimg.cn/9015ae4d98654663bc3206c2569db090.png
[8be8f395afc848e684ee97a9f9bb1525.png]: https://img-blog.csdnimg.cn/8be8f395afc848e684ee97a9f9bb1525.png
[a47b40d8a7854f91bc19f792dc48eae8.png]: https://img-blog.csdnimg.cn/a47b40d8a7854f91bc19f792dc48eae8.png
[bbfa7c5608b24b719d4a7ebbcc345a3a.png]: https://img-blog.csdnimg.cn/bbfa7c5608b24b719d4a7ebbcc345a3a.png
[9c8d68d543f74698abbcfb3537e04906.png]: https://img-blog.csdnimg.cn/9c8d68d543f74698abbcfb3537e04906.png
[ab084dd1c6544cad942e2e051caefbe0.png]: https://img-blog.csdnimg.cn/ab084dd1c6544cad942e2e051caefbe0.png
[7d40d8fc732c40c28c281ec1364d851b.png]: https://img-blog.csdnimg.cn/7d40d8fc732c40c28c281ec1364d851b.png
[3b3a8567932d474ca475b7406e6119cd.png]: https://img-blog.csdnimg.cn/3b3a8567932d474ca475b7406e6119cd.png
[d5b825084cd94c3fb80372bf233adc52.png]: https://img-blog.csdnimg.cn/d5b825084cd94c3fb80372bf233adc52.png
[94e81aaaec5845f5aa20cb4b0fb063a4.png]: https://img-blog.csdnimg.cn/94e81aaaec5845f5aa20cb4b0fb063a4.png
[c5fc953828ae48beb5c121dbbe7c1482.png]: https://img-blog.csdnimg.cn/c5fc953828ae48beb5c121dbbe7c1482.png
[9c0bcefda1b54866a6c75d21c36c342a.png]: https://img-blog.csdnimg.cn/9c0bcefda1b54866a6c75d21c36c342a.png