Flink 窗口

 窗口（Window）

窗口是处理无限流的核心。 窗口将流分割成有限大小的“桶”，我们可以计算窗口中的数据。

窗口程序一般有键控流（keyed streams）的窗口程序 和 非键控流（non-keyed streams）的窗口程序。

监控流（keyed streams）的窗口程序

stream
       .keyBy(...)               <-  keyed versus non-keyed windows
       .window(...)              <-  required: "assigner"
      [.trigger(...)]            <-  optional: "trigger" (else default trigger)
      [.evictor(...)]            <-  optional: "evictor" (else no evictor)
      [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
      [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
       .reduce/aggregate/apply()      <-  required: "function"
      [.getSideOutput(...)]      <-  optional: "output tag"

非监控流（non-keyed streams）的窗口程序

stream
       .windowAll(...)           <-  required: "assigner"
      [.trigger(...)]            <-  optional: "trigger" (else default trigger)
      [.evictor(...)]            <-  optional: "evictor" (else no evictor)
      [.allowedLateness(...)]    <-  optional: "lateness" (else zero)
      [.sideOutputLateData(...)] <-  optional: "output tag" (else no side output for late data)
       .reduce/aggregate/apply()      <-  required: "function"
      [.getSideOutput(...)]      <-  optional: "output tag"

窗口生命周期

窗口的创建：一旦应该属于该窗口的第一个元素到达，就会创建一个窗口。

窗口的删除：当时间（事件或处理时间）超过其结束时间戳加上用户指定的允许延迟时，该窗口将被完全删除。 Flink 保证仅删除基于时间的窗口，而不保证删除其他类型，例如 全局窗口。

例如，采用基于事件时间的窗口策略，每 5 分钟创建一次非重叠（或翻滚）窗口，并且允许延迟 1 分钟，如12:00 到 12:05 是一个窗口周期，当第一个时间戳落入该区间的元素到达时Flink创建该窗口，当水印经过 12:06 时间戳时，Flink将删除这个窗口。

此外，每个窗口都会附加一个触发器和一个函数（ProcessWindowFunction、ReduceFunction 或 AggregateFunction）。 该函数将包含要应用于窗口内容的计算，而触发器指定窗口被视为准备好应用函数的条件。 触发策略可能类似于“当窗口中的元素数量超过 4 时”或“当水印经过窗口末尾时”。 触发器还可以决定在创建和删除窗口之间的任何时间清除窗口的内容。 在这种情况下，清除仅指窗口中的元素，而不是窗口元数据。 这意味着新数据仍然可以添加到该窗口中。

除此之外，还可以指定一个 Evictor，它将能够在触发器触发后以及应用函数之前和/或之后从窗口中删除元素。

键控窗口与非键控窗口

流数据 使用 keyBy(...) 会将无限流拆分为逻辑键控流。 如果未调用 keyBy(...)，则是非键控流。

对于键控流，传入事件的任何属性都可以用作键。 拥有键控流将允许多个任务并行执行窗口计算，因为每个逻辑键控流都可以独立于其余流进行处理。 引用相同键的所有元素将被发送到相同的并行任务。

对于非键控流，原始流不会被分割成多个逻辑流，并且所有窗口逻辑将由单个任务执行，即并行度为 1。

窗口分配器（Window Assigners）

在指定了流是键控的还是非键控的后，下一步是定义窗口分配器。

窗口分配器决定了如何将元素分配到各个窗口中。

通常通过在键控流中使用window()方法，或在非键控流中使用windowAll()方法，并指定所需的WindowAssigner来实现。

窗口分配器负责将每个传入的元素分配到一个或多个窗口中。Flink为最常见的用例提供了预定义的窗口分配器，即滚动窗口、滑动窗口、会话窗口和全局窗口。

此外，用户还可以通过扩展WindowAssigner类来实现自定义窗口分配器。

除了全局窗口外，所有内置窗口分配器都是基于时间（可以是处理时间或事件时间）来分配元素的。

基于时间的窗口有一个开始时间戳（包含）和一个结束时间戳（不包含），这两个时间戳共同描述了窗口的大小。在代码中，Flink使用TimeWindow类来处理基于时间的窗口，该类提供了查询开始和结束时间戳的方法，以及一个返回给定窗口允许的最大时间戳的额外方法。

滚动窗口

滚动窗口分配器将每个元素分配到一个具有指定窗口大小的窗口中。滚动窗口具有固定的大小，并且不会重叠。

每个窗口的有开始和结束时间戳，随着时间的推移，元素会根据它们到达的时间被分配到相应的窗口中。一旦一个窗口的时间范围结束，该窗口就会被关闭，并且其中的元素会被处理（例如，进行聚合操作）。然后，一个新的窗口会立即开始，并等待新的元素被分配进来。

这种滚动窗口的分配方式适用于那些需要定期分析固定时间段内数据的场景，比如每5分钟统计一次网站访问量。由于窗口之间不重叠，每个元素只会被分配到一个窗口中，这有助于简化数据处理逻辑。

请注意，使用滚动窗口时，需要确保数据流中的时间戳是准确且有序的，如果数据流的时间戳不准确或乱序，可能会导致窗口分配错误，从而影响分析结果的准确性。

Java代码示例

DataStream<T> input = ...;
 
// tumbling event-time windows
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>);
 
// tumbling processing-time windows
input
    .keyBy(<key selector>)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .<windowed transformation>(<window function>);
 
// daily tumbling event-time windows offset by -8 hours.
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
    .<windowed transformation>(<window function>);

时间间隔可以使用Time.milliseconds(x)、Time.seconds(x)、Time.minutes(x)等函数来指定。

正如上一个例子所示，滚动窗口分配器还接受一个可选的偏移参数，该参数可用于改变窗口的对齐方式。例如，如果不设置偏移量，则每小时的滚动窗口会与纪元时间对齐，即会得到如1:00:00.000 - 1:59:59.999、2:00:00.000 - 2:59:59.999等这样的窗口。如果想要改变对齐方式，你可以设置一个偏移量。例如，如果设置了一个15分钟的偏移量，那么将得到1:15:00.000 - 2:14:59.999、2:15:00.000 - 3:14:59.999等这样的窗口。偏移量的一个重要用例是调整窗口以适应除UTC-0时区之外的其他时区。例如，在中国，需要指定一个偏移量Time.hours(-8)。

通过使用偏移量，可以确保窗口的边界与特定时区的时间边界对齐，这在处理具有时区特定需求的数据时非常有用。这样，无论Flink作业在哪个时区运行，都可以确保数据按照期望的时间范围进行窗口化，从而得到准确的分析结果。

请注意，偏移量仅影响窗口的起始时间，窗口的大小（即时间间隔）仍然保持不变。

滑动窗口

滑动窗口分配器将元素分配到固定长度的窗口中。与滚动窗口分配器类似，窗口的大小由窗口大小参数配置。而一个额外的窗口滑动参数控制着滑动窗口启动的频率。因此，如果滑动间隔小于窗口大小，滑动窗口可以重叠。在这种情况下，元素会被分配到多个窗口中。

例如，你可以设置大小为10分钟的窗口，每5分钟滑动一次。这样，你每5分钟就会得到一个包含过去10分钟内到达的事件的窗口，如以下图表所示。

在这个图表中，可以看到窗口如何随着时间的推移而滑动。每个窗口包含了一定时间段内的所有事件，这些事件根据它们到达的时间被分配到相应的窗口中。由于滑动间隔小于窗口大小，所以窗口之间存在重叠部分。这种重叠使得滑动窗口能够在时间上提供更加平滑的覆盖，适用于需要持续追踪一段时间内事件变化的场景，如流量分析、温度监测等。

使用滑动窗口时，需要仔细考虑窗口大小和滑动间隔的设置，以确保能够满足分析需求。如果窗口太大，可能会导致分析结果过于粗糙；如果窗口太小或滑动间隔太大，可能会丢失一些重要的细节信息。因此，在实际应用中，可能需要根据数据的特性和分析目标来调整这些参数。

Java代码示例

DataStream<T> input = ...;
 
// sliding event-time windows
input
    .keyBy(<key selector>)
    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    .<windowed transformation>(<window function>);
 
// sliding processing-time windows
input
    .keyBy(<key selector>)
    .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    .<windowed transformation>(<window function>);
 
// sliding processing-time windows offset by -8 hours
input
    .keyBy(<key selector>)
    .window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8)))
    .<windowed transformation>(<window function>);

会话窗口

会话窗口分配器根据活动会话将元素分组。与滚动窗口和滑动窗口不同，会话窗口不会重叠，也没有固定的开始和结束时间。相反，会话窗口在没有接收到元素一段时间后关闭，即当发生一段时间的不活跃期时。会话窗口分配器可以配置为使用静态会话间隔，或者使用会话间隔提取函数来定义不活跃期的时长。当这个时段结束时，当前会话关闭，后续的元素将被分配到新的会话窗口中。

使用会话窗口在处理具有不同活跃期的数据流时非常有用，比如用户会话、网页浏览事件等。由于会话窗口是根据实际活动来定义的，因此它们能够更自然地反映数据的实际模式。

配置会话窗口时，需要确定一个合适的会话间隔。这个间隔应该足够长，以便能够捕捉到会话中的活动，但又不能太长，以免将不同的会话错误地合并在一起。如果会话模式变化很大，可以考虑使用会话间隔提取函数，以便根据每个会话的具体特性动态地调整间隔时长。

通过使用会话窗口，可以更好地理解和分析数据中的活动模式，从而得到更准确的业务洞察。

Java代码示例

DataStream<T> input = ...;
 
// event-time session windows with static gap
input
    .keyBy(<key selector>)
    .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
    .<windowed transformation>(<window function>);
    
// event-time session windows with dynamic gap
input
    .keyBy(<key selector>)
    .window(EventTimeSessionWindows.withDynamicGap((element) -> {
        // determine and return session gap
    }))
    .<windowed transformation>(<window function>);
 
// processing-time session windows with static gap
input
    .keyBy(<key selector>)
    .window(ProcessingTimeSessionWindows.withGap(Time.minutes(10)))
    .<windowed transformation>(<window function>);
    
// processing-time session windows with dynamic gap
input
    .keyBy(<key selector>)
    .window(ProcessingTimeSessionWindows.withDynamicGap((element) -> {
        // determine and return session gap
    }))
    .<windowed transformation>(<window function>);

全局窗口

全局窗口分配器将所有具有相同键的元素分配到同一个全局窗口中。这种窗口方案仅在指定了自定义触发器时才有用。否则，将不会执行任何计算，因为全局窗口没有自然的结束点来处理聚合的元素。

全局窗口提供了一个灵活的框架，允许你根据自定义逻辑来定义窗口的结束。由于全局窗口不会根据时间或其他标准自动关闭，因此需要提供一个触发器来确定何时应该关闭窗口并触发计算。

在使用全局窗口时，你需要仔细考虑如何定义触发器，以确保窗口在适当的时候关闭。触发器可以根据元素的数量、特定的时间条件、外部事件等因素来触发窗口的关闭。通过自定义触发器，实现复杂的窗口逻辑，以满足特定的业务需求。

需要注意的是，全局窗口可能会导致大量的数据在内存中积累，直到触发器被触发为止。因此，在使用全局窗口时，需要确保系统具有足够的内存和处理能力来处理可能的大数据量。此外，需要仔细设计触发器的逻辑，以避免出现窗口长时间不关闭或频繁关闭的情况，这可能会影响系统的性能和稳定性。

Java代码示例

DataStream<T> input = ...;
 
input
    .keyBy(<key selector>)
    .window(GlobalWindows.create())
    .<windowed transformation>(<window function>);

窗口函数（Window Functions）

ReduceFunction

ReduceFunction 指定如何组合输入中的两个元素以生成相同类型的输出元素。 Flink 使用ReduceFunction 增量聚合窗口的元素。

Java示例代码

DataStream<Tuple2<String, Long>> input = ...;
 
input
    .keyBy(<key selector>)
    .window(<window assigner>)
    .reduce(new ReduceFunction<Tuple2<String, Long>>() {
      public Tuple2<String, Long> reduce(Tuple2<String, Long> v1, Tuple2<String, Long> v2) {
        return new Tuple2<>(v1.f0, v1.f1 + v2.f1);
      }
    });

AggregateFunction

AggregateFunction 是ReduceFunction 的通用版本，具有三种类型：输入类型(IN)、累加器类型(ACC) 和输出类型(OUT)。 输入类型是输入流中元素的类型，AggregateFunction 具有一种将一个输入元素添加到累加器的方法。 该接口还具有用于创建初始累加器、将两个累加器合并为一个累加器以及从累加器提取输出（OUT 类型）的方法。 我们将在下面的示例中看到它是如何工作的。 与ReduceFunction 相同，Flink 将在窗口的输入元素到达时增量聚合它们。

Java代码示例

 
/**
 * The accumulator is used to keep a running sum and a count. The {@code getResult} method
 * computes the average.
 */
private static class AverageAggregate
    implements AggregateFunction<Tuple2<String, Long>, Tuple2<Long, Long>, Double> {
  @Override
  public Tuple2<Long, Long> createAccumulator() {
    return new Tuple2<>(0L, 0L);
  }
 
  @Override
  public Tuple2<Long, Long> add(Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) {
    return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1L);
  }
 
  @Override
  public Double getResult(Tuple2<Long, Long> accumulator) {
    return ((double) accumulator.f0) / accumulator.f1;
  }
 
  @Override
  public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
    return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
  }
}
 
DataStream<Tuple2<String, Long>> input = ...;
 
input
    .keyBy(<key selector>)
    .window(<window assigner>)
    .aggregate(new AverageAggregate());

ProcessWindowFunction

ProcessWindowFunction 获取一个包含窗口所有元素的 Iterable，以及一个可以访问时间和状态信息的 Context 对象，这使其能够提供比其他窗口函数更大的灵活性。 这是以性能和资源消耗为代价的，因为元素不能增量聚合，而是需要在内部缓冲，直到窗口被认为准备好进行处理。

源码如下

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> implements Function {
 
    /**
     * Evaluates the window and outputs none or several elements.
     *
     * @param key The key for which this window is evaluated.
     * @param context The context in which the window is being evaluated.
     * @param elements 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.
     */
    public abstract void process(
            KEY key,
            Context context,
            Iterable<IN> elements,
            Collector<OUT> out) throws Exception;
    
    /**
     * Deletes any state in the {@code Context} when the Window expires (the watermark passes its
     * {@code maxTimestamp} + {@code allowedLateness}).
     *
     * @param context The context to which the window is being evaluated
     * @throws Exception The function may throw exceptions to fail the program and trigger recovery.
     */
    public void clear(Context context) throws Exception {}
 
    /**
     * The context holding window metadata.
     */
    public abstract class Context implements java.io.Serializable {
        /**
         * Returns the window that is being evaluated.
         */
        public abstract W window();
 
        /** Returns the current processing time. */
        public abstract long currentProcessingTime();
 
        /** Returns the current event-time watermark. */
        public abstract long currentWatermark();
 
        /**
         * State accessor for per-key and per-window state.
         *
         * <p><b>NOTE:</b>If you use per-window state you have to ensure that you clean it up
         * by implementing {@link ProcessWindowFunction#clear(Context)}.
         */
        public abstract KeyedStateStore windowState();
 
        /**
         * State accessor for per-key global state.
         */
        public abstract KeyedStateStore globalState();
    }
 
}

key 参数是通过为 keyBy() 调用指定的 KeySelector 提取的密钥。 如果是元组索引键或字符串字段引用，则此键类型始终为 Tuple，必须手动将其转换为正确大小的元组才能提取键字段。

Java代码示例

DataStream<Tuple2<String, Long>> input = ...;
 
input
  .keyBy(t -> t.f0)
  .window(TumblingEventTimeWindows.of(Time.minutes(5)))
  .process(new MyProcessWindowFunction());
 
/* ... */
 
public class MyProcessWindowFunction 
    extends ProcessWindowFunction<Tuple2<String, Long>, String, String, TimeWindow> {
 
  @Override
  public void process(String key, Context context, Iterable<Tuple2<String, Long>> input, Collector<String> out) {
    long count = 0;
    for (Tuple2<String, Long> in: input) {
      count++;
    }
    out.collect("Window: " + context.window() + "count: " + count);
  }
}

该示例显示了一个对窗口中的元素进行计数的 ProcessWindowFunction。此外，窗口函数还将有关窗口的信息添加到输出中。

请注意，使用 ProcessWindowFunction 进行简单聚合（例如计数）效率相当低。

带有增量聚合的ProcessWindowFunction

ProcessWindowFunction 可以与ReduceFunction 或AggregateFunction 结合使用，以在元素到达窗口时增量聚合元素。 当窗口关闭时，将向 ProcessWindowFunction 提供聚合结果。 这允许它增量计算窗口，同时可以访问 ProcessWindowFunction 的附加窗口元信息。 还可以使用旧版 WindowFunction 而不是 ProcessWindowFunction 进行增量窗口聚合。

ProcessWindowFunction 结合 ReduceFunction

以下示例展示了如何将增量式ReduceFunction 与ProcessWindowFunction 结合起来，以返回窗口中的最小事件以及窗口的开始时间。

Java代码示例

DataStream<SensorReading> input = ...;
 
input
  .keyBy(<key selector>)
  .window(<window assigner>)
  .reduce(new MyReduceFunction(), new MyProcessWindowFunction());
 
// Function definitions
 
private static class MyReduceFunction implements ReduceFunction<SensorReading> {
 
  public SensorReading reduce(SensorReading r1, SensorReading r2) {
      return r1.value() > r2.value() ? r2 : r1;
  }
}
 
private static class MyProcessWindowFunction
    extends ProcessWindowFunction<SensorReading, Tuple2<Long, SensorReading>, String, TimeWindow> {
 
  public void process(String key,
                    Context context,
                    Iterable<SensorReading> minReadings,
                    Collector<Tuple2<Long, SensorReading>> out) {
      SensorReading min = minReadings.iterator().next();
      out.collect(new Tuple2<Long, SensorReading>(context.window().getStart(), min));
  }
}

ProcessWindowFunction 结合 AggregateFunction

以下示例展示了如何将增量 AggregateFunction 与 ProcessWindowFunction 结合起来计算平均值，并发出键和窗口的平均值。

Java代码示例

DataStream<Tuple2<String, Long>> input = ...;
 
input
  .keyBy(<key selector>)
  .window(<window assigner>)
  .aggregate(new AverageAggregate(), new MyProcessWindowFunction());
 
// Function definitions
 
/**
 * The accumulator is used to keep a running sum and a count. The {@code getResult} method
 * computes the average.
 */
private static class AverageAggregate
    implements AggregateFunction<Tuple2<String, Long>, Tuple2<Long, Long>, Double> {
  @Override
  public Tuple2<Long, Long> createAccumulator() {
    return new Tuple2<>(0L, 0L);
  }
 
  @Override
  public Tuple2<Long, Long> add(Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) {
    return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1L);
  }
 
  @Override
  public Double getResult(Tuple2<Long, Long> accumulator) {
    return ((double) accumulator.f0) / accumulator.f1;
  }
 
  @Override
  public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
    return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
  }
}
 
private static class MyProcessWindowFunction
    extends ProcessWindowFunction<Double, Tuple2<String, Double>, String, TimeWindow> {
 
  public void process(String key,
                    Context context,
                    Iterable<Double> averages,
                    Collector<Tuple2<String, Double>> out) {
      Double average = averages.iterator().next();
      out.collect(new Tuple2<>(key, average));
  }
}

在ProcessWindowFunction中使用每个窗口的状态

除了访问键控状态（任何丰富的函数都可以）之外，ProcessWindowFunction 还可以使用作用域为函数当前正在处理的窗口的键控状态。 在这种情况下，了解每个窗口状态所指的窗口是什么非常重要。 涉及不同的“窗口”：

指定窗口操作时定义的窗口：这可能是 1 小时的滚动窗口或滑动 1 小时的 2 小时滑动窗口。
给定键的已定义窗口的实际实例：对于用户 ID xyz，这可能是从 12:00 到 13:00 的时间窗口。 这是基于窗口定义的，并且根据作业当前正在处理的键的数量以及事件所属的时隙，将会有许多窗口。
每个窗口的状态与这两者中的后者相关。 这意味着，如果我们处理 1000 个不同键的事件，并且所有这些键的事件当前都落入 [12:00, 13:00) 时间窗口，那么将有 1000 个窗口实例，每个实例都有自己的键控每个窗口状态。

process() 调用接收的 Context 对象上有两个方法，允许访问两种类型的状态：

globalState()，它允许访问不在窗口范围内的键控状态
windowState()，它允许访问范围也在窗口内的键控状态
如果预计同一窗口会多次触发，则此功能非常有用，当对迟到的数据进行延迟触发或当有一个进行推测性早期触发的自定义触发器时，可能会发生这种情况。 在这种情况下，将存储有关先前触发或每个窗口状态的触发次数的信息。

使用窗口状态时，清除窗口时清除该状态也很重要。 这应该发生在clear()方法中。

WindowFunction（旧版）

在某些可以使用 ProcessWindowFunction 的地方，也可以使用 WindowFunction。 这是 ProcessWindowFunction 的旧版本，提供的上下文信息较少，并且没有一些高级功能，例如每个窗口的键控状态。 该接口将在某个时候被弃用。

源码如下

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

Java代码示例

DataStream<Tuple2<String, Long>> input = ...;
input
    .keyBy(<key selector>)
    .window(<window assigner>)
    .apply(new MyWindowFunction());

触发器（Triggers）

触发器确定窗口（由窗口分配器形成）何时准备好由窗口函数处理。 每个 WindowAssigner 都带有一个默认的触发器。 如果默认触发器不能满足业务需求，可以使用trigger(...)指定自定义触发器。

触发器接口有五种方法，允许触发器对不同的事件做出反应：

每个添加到窗口的元素都会调用 onElement() 方法。
当注册的事件时间计时器触发时，将调用 onEventTime() 方法。
当注册的处理时间计时器触发时，将调用 onProcessingTime() 方法。
onMerge() 方法与有状态触发器相关，并在两个触发器相应的窗口合并时合并两个触发器的状态，例如 使用会话窗口时。
最后，clear() 方法删除相应窗口时执行所需的任何操作。

上述方法需要注意的两点是：

1.前三个决定如何通过返回 TriggerResult 来处理其调用事件。 该操作可以是以下操作之一：

• CONTINUE：什么也不做
• FIRE：触发计算
• PURGE：清除窗口中的元素
• FIRE_AND_PURGE：触发计算并随后清除窗口中的元素

2.这些方法中的任何一个都可以用于注册处理或事件时间计时器以用于将来的操作。

触发与清除

一旦触发器确定窗口已准备好进行处理，它就会触发，即返回 FIRE 或 FIRE_AND_PURGE。 这是窗口操作符发出当前窗口结果的信号。 给定一个带有 ProcessWindowFunction 的窗口，所有元素都会传递给 ProcessWindowFunction（可能在将它们传递给逐出器之后）。 带有ReduceFunction 或AggregateFunction 的Windows 只是简单地发出它们急切的聚合结果。

当触发器触发时，它可以是 FIRE 或 FIRE_AND_PURGE。 FIRE 保留窗口的内容，而 FIRE_AND_PURGE 则删除其内容。 默认情况下，预实现的触发器只是简单地触发而不清除窗口状态。

清除只会删除窗口的内容，并完整保留有关窗口的任何潜在元信息和任何触发器状态。

窗口分配器的默认触发器

WindowAssigner 的默认触发器适用于许多用例。 例如，所有事件时间窗口分配器都有一个 EventTimeTrigger 作为默认触发器。 一旦水印经过窗口末尾，就会触发此触发器。

GlobalWindow 的默认触发器是 NeverTrigger，它从不触发。 因此，在使用 GlobalWindow 时，始终必须定义自定义触发器。

通过使用trigger()指定触发器，将覆盖WindowAssigner的默认触发器。 例如，如果为 TumblingEventTimeWindows 指定 CountTrigger，将不再根据时间进度而仅根据计数来触发窗口。 如果想根据时间和计数做出反应，则必须编写自己的自定义触发器。

内置触发器和自定义触发器

Flink 带有一些内置触发器。

• （已经提到的）EventTimeTrigger 根据水印测量的事件时间进度触发。
• ProcessingTimeTrigger 根据处理时间触发。
• 一旦窗口中的元素数量超过给定限制，CountTrigger 就会触发。
• PurgingTrigger 将另一个触发器作为参数，并将其转换为清除触发器。

如果需要实现自定义触发器，应该查看Trigger抽象类。 请注意，该 API 仍在不断发展，并且可能会在 Flink 的未来版本中发生变化。

清除器（Evictors）

除了 WindowAssigner 和 Trigger 之外，Flink 的窗口模型还允许指定可选的 Evictor。 这可以使用 evictor(...) 方法来完成（如本文开头所示）。 清除器能够在触发器触发之后以及应用窗口函数之前和/或之后从窗口中删除元素。 为此，Evictor 接口有两种方法：

/**
 * Optionally evicts elements. Called before windowing function.
 *
 * @param elements The elements currently in the pane.
 * @param size The current number of elements in the pane.
 * @param window The {@link Window}
 * @param evictorContext The context for the Evictor
 */
void evictBefore(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);
 
/**
 * Optionally evicts elements. Called after windowing function.
 *
 * @param elements The elements currently in the pane.
 * @param size The current number of elements in the pane.
 * @param window The {@link Window}
 * @param evictorContext The context for the Evictor
 */
void evictAfter(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);

evictBefore() 包含在窗口函数之前应用的清除逻辑，而 evictAfter() 包含在窗口函数之后应用的清除逻辑。 在应用窗口函数之前被清除的元素将不会被窗口函数处理。

Flink 附带了三个预先实现的清除器。 这些都是：

• CountEvictor：保留窗口中用户指定数量的元素，并丢弃窗口缓冲区开头的剩余元素。
• DeltaEvictor：采用 DeltaFunction 和阈值，计算窗口缓冲区中最后一个元素与其余每个元素之间的增量，并删除增量大于或等于阈值的元素。
• TimeEvictor：以毫秒为单位的间隔作为参数，对于给定的窗口，它在其元素中查找最大时间戳 max_ts 并删除时间戳小于 max_ts - interval的所有元素。

默认情况下，所有预先实现的清除器都在窗口函数之前应用其逻辑。

指定清除器可以防止任何预聚合，因为在应用计算之前必须将窗口的所有元素传递给清除器。这意味着带有清除器的窗口将创建更多的状态。

Flink 不保证窗口内元素的顺序。这意味着虽然清除器可以从窗口的开头删除元素，但这些元素不一定是最先或最后到达的元素。

允许延迟

当使用事件时间窗口时，元素可能会延迟到达，即 Flink 用于跟踪事件时间进度的水印已经超过了元素所属窗口的结束时间戳。 

默认情况下，当水印超过窗口末尾时，迟到的元素将被删除。 然而，Flink 允许为窗口操作符指定允许的最大延迟时间。 允许的延迟指定元素在被删除之前可以延迟的时间，其默认值为 0。在水印经过窗口末尾但在经过窗口末尾之前到达的元素加上允许的延迟， 仍然添加到窗口中。 根据所使用的触发器，迟到但未删除的元素可能会导致窗口再次触发。 EventTimeTrigger 就是这种情况。

为了实现这一点，Flink 会保留窗口的状态，直到其允许的延迟到期。 一旦发生这种情况，Flink 将删除窗口并删除其状态。

默认情况下，允许的延迟设置为 0。也就是说，在水印之后到达的元素将被丢弃。

Java代码示例

DataStream<T> input = ...;
 
input
    .keyBy(<key selector>)
    .window(<window assigner>)
    .allowedLateness(<time>)
    .<windowed transformation>(<window function>);

使用 GlobalWindows 窗口分配器时，不会将任何数据视为迟到，因为全局窗口的结束时间戳为 Long.MAX_VALUE。

将延迟数据作为侧输出

使用 Flink 的侧输出功能，可以获得最新丢弃的数据流。

首先需要指定要在窗口流上使用 sideOutputLateData(OutputTag) 获取延迟数据。 然后，可以获得窗口操作结果的侧输出流：

Java代码示例

final OutputTag<T> lateOutputTag = new OutputTag<T>("late-data"){};
 
DataStream<T> input = ...;
 
SingleOutputStreamOperator<T> result = input
    .keyBy(<key selector>)
    .window(<window assigner>)
    .allowedLateness(<time>)
    .sideOutputLateData(lateOutputTag)
    .<windowed transformation>(<window function>);
 
DataStream<T> lateStream = result.getSideOutput(lateOutputTag);

关于延迟元素的考虑

当指定允许的延迟大于 0 时，窗口及其内容将在水印经过窗口末尾后保留。 在这些情况下，当一个迟到但未删除的元素到达时，它可能会触发窗口的另一次触发。 这些触发称为延迟触发，因为它们是由延迟事件触发的，与窗口第一次触发的主触发相反。 在会话窗口的情况下，延迟触发可能会进一步导致窗口合并，因为它们可能会“弥合”两个预先存在的未合并窗口之间的间隙。

延迟触发发出的元素应被视为先前计算的更新结果，即您的数据流将包含同一计算的多个结果。 根据应用程序，需要考虑这些重复的结果或删除它们的重复项。

处理窗口结果

窗口操作的结果又是一个数据流，结果元素中不会保留有关窗口操作的信息，因此如果想保留有关窗口的元信息，则必须在 ProcessWindowFunction 的结果元素中手动编码该信息。 在结果元素上设置的唯一相关信息是元素时间戳。 这被设置为已处理窗口的最大允许时间戳，即结束时间戳 - 1，因为窗口结束时间戳是独占的。 请注意，这对于事件时间窗口和处理时间窗口都是如此。 即，在窗口操作之后，元素始终具有时间戳，但这可以是事件时间时间戳或处理时间时间戳。 对于处理时间窗口，这没有特殊含义，但对于事件时间窗口，这与水印与窗口的交互方式一起启用了具有相同窗口大小的连续窗口操作。 

水印与窗口的交互

当水印到达窗口操作符时，会触发两件事：

• 水印触发最大时间戳（即结束时间戳 - 1）小于新水印的所有窗口的计算
• 水印（按原样）转发到下游操作

直观地说，一旦下游操作收到该水印，水印就会“清除”所有在下游操作中被视为后期的窗口。

连续的窗口操作

如前所述，计算窗口结果的时间戳的方式以及水印与窗口交互的方式允许将连续的窗口操作串在一起。 当想要执行两个连续的窗口操作（其中想要使用不同的键但仍希望来自同一上游窗口的元素最终出现在同一下游窗口中）时，这可能很有用。

Java代码示例

DataStream<Integer> input = ...;
 
DataStream<Integer> resultsPerKey = input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .reduce(new Summer());
 
DataStream<Integer> globalResults = resultsPerKey
    .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
    .process(new TopKWindowFunction());

在此示例中，第一个操作的时间窗口 [0, 5) 的结果也将在后续窗口操作中的时间窗口 [0, 5) 中结束。 这允许计算每个键的总和，然后在第二次操作中计算同一窗口内的前 k 个元素。

有用的状态大小考虑

窗口可以在很长一段时间内（例如几天、几周或几个月）进行定义，因此会积累非常大的状态。 在估计窗口计算的存储需求时，需要记住一些规则：

Flink 为每个元素所属的窗口创建一个副本。 鉴于此，翻滚窗口保留每个元素的一份副本（一个元素恰好属于一个窗口，除非它被延迟删除）。 相反，滑动窗口为每个元素创建多个元素，如“窗口分配器”部分中所述。 因此，大小为 1 天、滑动 1 秒的滑动窗口可能不是一个好主意。

ReduceFunction 和 AggregateFunction 可以显着降低存储需求，因为它们急切地聚合元素并且每个窗口仅存储一个值。 相反，仅使用 ProcessWindowFunction 就需要累积所有元素。

使用清除器可以防止任何预聚合，因为在应用计算之前窗口的所有元素都必须通过清除器。

完成！enjoy it!