分类： 大数据

flink Window Assigners是Flink流处理中用于将数据流划分为有限大小的时间窗口的机制之一。Window Assigners负责将数据流中的数据分配到不同的时间窗口中。时间窗口可以基于不同的标准进行划分，如时间戳、事件数量或其他自定义标准。

Flink提供了多个Window Assigners实现，包括滚动窗口（Tumbling Windows）、滑动窗口（Sliding Windows）、会话窗口（Session Windows）等。滚动窗口是基于固定大小的时间段来划分数据流的，而滑动窗口则是根据固定大小和滑动间隔来划分数据流的。会话窗口根据一定的时间间隔来划分数据流，并根据数据之间的间隔来决定何时关闭会话窗口。

使用Window Assigners，可以将数据流中的数据分配到不同的时间窗口中，从而方便地进行聚合、计算和分析。

一、窗口（Window）

窗口（Window）是处理无界流的关键所在。窗口可以将数据流装入大小有限的“桶”中，再对每个“桶”加以处理。

1. 滚动窗口（Tumbling Windows）

滚动窗口有固定的大小，是一种对数据进行均匀切片的划分方式。窗口之间没有重叠，也不会有间隔，是“首尾相接”的状态。滚动窗口可以基于时间定义，也可以基于数据个数定义；需要的参数只有一个，就是窗口的大小（window size）。

2. 滑动窗口（Sliding Windows）

与滚动窗口类似，滑动窗口的大小也是固定的。区别在于，窗口之间并不是首尾相接的，而是可以“错开”一定的位置。如果看作一个窗口的运动，那么就像是向前小步“滑动”一样。定义滑动窗口的参数有两个：除去窗口大小（window size）之外，还有一个滑动步长（window slide），代表窗口计算的频率。

3. 会话窗口（Session Windows）

借用会话超时失效的机制来描述窗口简单来说，就是数据来了之后就开启一个会话窗口，如果接下来还有数据陆续到来，那么就一直保持会话；如果一段时间一直没收到数据，那就认为会话超时失效，窗口自动关闭。与滑动窗口和滚动窗口不同，会话窗口只能基于时间来定义，而没有“会话计数窗口”的概念。

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

4. 全局窗口（Global Windows）

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

二、Keyed Streams和Non-Keyed Streams

Flink 窗口在 keyed streams 和 non-keyed streams 上使用的基本结构。 我们可以看到，这两者唯一的区别仅在于：keyed streams 要调用 keyBy(…)后再调用 window(…) ， 而 non-keyed streams 只用直接调用 windowAll(…)。

在Apache Flink中，数据流可以被分为两种类型：Keyed Streams和Non-Keyed Streams。

Keyed Streams是指在处理数据流时需要对数据进行分区，并将数据按照某个Key进行分组的数据流。Keyed Streams中的每个元素都有一个Key，通过这个Key来确定数据所属的分区。在Flink中，Keyed Streams通常用于聚合操作，如GroupBy和Window。

相反，Non-Keyed Streams是指在处理数据流时不需要按照某个Key进行分组的数据流。在Non-Keyed Streams中，每个元素都是相互独立的，没有固定的分区和分组关系。Non-Keyed Streams通常用于一些基于时间的操作，如TimeWindow和ProcessFunction。

区分Keyed Streams和Non-Keyed Streams的方法是，当定义数据流时，如果调用了DataStream的keyBy()方法，那么这个数据流就是Keyed Streams。如果没有调用keyBy()方法，则这个数据流就是Non-Keyed Streams。

    import org.apache.flink.streaming.api.scala._

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream: DataStream[(String, Int)] = env
      .socketTextStream("localhost", 9999)
      .map { value =>
        val Array(key, count) = value.split(",")
        (key, count.toInt)
      }
      .keyBy(_._1) // 调用keyBy方法，将数据按照Tuple2的第一个元素进行分组，因此这个数据流是Keyed Streams

注意的是，在Scala中，对于Tuple类型的数据，可以直接使用元组的语法来访问元素。例如，可以使用_1来访问Tuple的第一个元素，使用2来访问第二个元素。因此，在上面的示例中，keyBy(._1)可以理解为按照Tuple的第一个元素进行分组。

    import org.apache.flink.streaming.api.scala._

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream: DataStream[String] = env
      .socketTextStream("localhost", 9999)
      .flatMap { value =>
        value.split(" ")
      } // 没有调用keyBy方法，因此这个数据流是Non-Keyed Streams