关键词：

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

Flink 中的时间和窗口

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

6.1 时间语义

6.1.1 Flink 中的时间语义

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

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

1. 处理时间（Processing Time）

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

2. 事件时间（Event Time）

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

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

---

6.1.2 哪种时间语义更重要

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

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

3. 两种时间语义的对比

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

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

---

6.2 水位线（Watermark）

6.2.1 什么是水位线

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

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

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

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

1. 有序流中的水位线

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

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

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

2. 乱序流中的水位线

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

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

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

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

如果仔细观察会看到，这种“等 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）传输模式下，一个任务可能会收到来自不同分区上游子任务的数据。而不同分区的子任务时钟并不同步，所以同一时刻发给下游任务的水位线可能并不相同。这说明上游各个分区处理有快有慢，进度各不相同，这时应以最慢的那个时钟，也就是最小水位线为准。

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

---

6.3 窗口（Window）

6.3.1 窗口的概念

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

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

窗口包含起始时间、不包含结束时间。

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

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

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

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

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

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

---

6.3.2 窗口的分类

1. 按照驱动类型分类

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

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

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

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

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

（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 个数进行一次统计。

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

（2）滑动窗口（Sliding Windows）

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

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

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

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

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

（3）会话窗口（Session Windows）

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

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

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

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

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

（4）全局窗口（Global Windows）

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

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

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 所示。

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

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

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学习（十四）flink窗口时间和水位线(代码片段)

...nk支持如下3种：滚动窗口，窗口数据有固定的大小，窗口中的数据不会叠加；滑动窗口，窗口数据有固定的大小，并且有生成间隔；会话窗口，窗口数据没有固定的大小，根据用户传入的参数进行划分，窗口数据无叠加。 &nb... 查看详情

flink中的时间和窗口(代码片段)

借鉴《尚硅谷Flink1.13版本笔记.pdf》中第六章Flink中的时间和窗口在流数据处理应用中，一个很重要的操作就是窗口计算。所谓的“窗口”，就是划定的一段时间范围，也就是“时间窗”；对这范围内数据进行处理&... 查看详情

「flink」flink中的时间类型(代码片段)

Flink中的时间类型和窗口是非常重要概念，是学习Flink必须要掌握的两个知识点。Flink中的时间类型时间类型介绍Flink流式处理中支持不同类型的时间。分为以下几种：处理时间Flink程序执行对应操作的系统时间。所有基于时间的... 查看详情

flink总结之一文彻底搞懂时间和窗口(代码片段)

...（Watermark）1.什么是水位线2.水位线分类1.有序流中的水位线1、如何使用2.乱序流中的水位线2.1乱序流中如何保证数据的准确性2.2如何使用三、窗口1.什么是窗口2.窗口分类1.按照驱动类型分1.计数窗口（CountWindow）2.... 查看详情

flink窗口window(代码片段)

...时间分类　　EventTime：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink通过时间戳分配器访问事件时间戳。　　IngestionTime：是数据进入Flink的时间。　　Proc... 查看详情

flink原理学习窗口和时间(代码片段)

...生的时间，一旦确定就不会发生变化。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink通过时间戳分配器访问事件时间戳。ProcessingTime：即处理时间，是... 查看详情

flink原理学习窗口和时间(代码片段)

...生的时间，一旦确定就不会发生变化。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink通过时间戳分配器访问事件时间戳。ProcessingTime：即处理时间，是... 查看详情

flink窗口window机制详解(代码片段)

...乱序处理，和checkpoint机制）。本文我们将介绍流式处理中的窗口概念，介绍Flink内建的一些窗口和Windo 查看详情

flink和sentinel中计算当前时间所属的时间窗口算法(代码片段)

flink中：org.apache.flink.streaming.api.windowing.windows.TimeWindowpublicstaticlonggetWindowStartWithOffset(longtimestamp,longoffset,longwindowSize)//转载请标明链接：https://blog.csdn.net/wabiaoz 查看详情

flink笔记9[实验]体验窗口开启时间和关闭时间(eventtime)(代码片段)

体验窗口开启时间和关闭时间 实验数据实验代码实验结果实验分析窗口开始时间公式实验数据sensor_1,1619492107,36.2sensor_1,1619492108,36.0sensor_1,1619492109,36.5sensor_1,1619492110,34.3sensor_1,1619492111,34.3sensor_1,1619492112,34.3sensor_1,16194 查看详情

12-flink-1.10.1-flink中的时间语义和watermark(代码片段)

目录1Fink中的时间语义1.1哪种时间语义更重要2设置EventTime3水位线（Watermark）3.1水位线概念3.2watermark原理和特点4watermark的传递，引入和设置4.1 watermark的传递 4.2watermark代码中引入 4.3自定义watermark4.4 watermark的实例代... 查看详情

flink中window窗口和时间以及watermark水印(代码片段)

...计，不是按照时间，比如countWindow(100)表示当窗口中的数据有100个的时候开始计算SlidingCountWindow:累积固定个数的数据就作为一个窗口，超过指定数量个数数据开启新一个窗口计算，比如coutnWindow(100,10)窗口大小是100&... 查看详情

flink窗口和水位线(代码片段)

...的话，水位线还有一个功能就是可以控制窗口的关闭时间窗口分类时间窗口： 查看详情

flink实现自定义滑动窗口(代码片段)

...们大部分的场景，但是有时候我们需要计算一个固定时间范围内的数据，比如实时计算每天凌晨到第二天凌晨的数据，或者每天上午7点到第二天上午7点。类似于这种情况Flink默认提供的窗口是不支持的，因为Flink... 查看详情

flink流计算随笔(代码片段)

...)在流上的工作方式与批处理不同。例如，不可能计算流中的所有元素，因为流通常是无限的(×××的)。相反，流上的聚合(计数、和等)是由窗口windows限定作用域的，例如“过去5分钟的计数”或“最后100个元素的总和”。Windows可... 查看详情

11.flink四大基石window窗口的分类flink提供了很多各种场景用的windowassigner基于时间的滚动和滑动基于时间的滚动和滑动窗口基于数量的滚动和滑动(代码片段)

...nk提供了很多各种场景用的WindowAssigner12.3.代码演示–基于时间的滚动和滑动–掌握12.3.1.本地ubuntu下安装nc12.3.2.基于时间的滚动和滑动窗口12.4.代码演示-基于数量的滚动和滑动12.5.代码演示–Session会话窗口11.Flink四大基石Checkpoint... 查看详情

flink窗口函数处理数据(watermark和sideoutput)(代码片段)

...重要的解决思路，1.添加水位线Watermark2.推迟关闭窗口时间3.超时数据的side输出下面的例子是 查看详情

12-flink-1.10.1-flink中的时间语义和watermark(代码片段)

目录1Fink中的时间语义1.1哪种时间语义更重要2设置EventTime3水位线（Watermark）3.1水位线概念3.2watermark原理和特点4watermark的传递，引入和设置4.1 watermark的传递 4.2watermark代码中引入 4.3自定义watermark4.4 watermark的实例代... 查看详情