Flink窗口全解析：三种时间窗口、窗口处理函数使用及案例

た入场券 2022-11-01 05:45 233阅读 0赞

> 个人网站将实时更新： [https://lulaoshi.info/flink/chapter-time-window/window.html][https_lulaoshi.info_flink_chapter-time-window_window.html]

我们经常需要在一个时间窗口维度上对数据进行聚合，窗口是流处理应用中经常需要解决的问题。Flink的窗口算子为我们提供了方便易用的API，我们可以将数据流切分成一个个窗口，对窗口内的数据进行处理。本文将介绍如何在Flink上进行窗口的计算。

一个Flink窗口应用的大致骨架结构如下所示：

// Keyed Window
    stream
           .keyBy(...)               <-  按照一个Key进行分组
           .window(...)              <-  将数据流中的元素分配到相应的窗口中
          [.trigger(...)]            <-  指定触发器Trigger（可选）
          [.evictor(...)]            <-  指定清除器Evictor(可选)
           .reduce/aggregate/process()      <-  窗口处理函数Window Function
    
    // Non-Keyed Window
    stream
           .windowAll(...)           <-  不分组，将数据流中的所有元素分配到相应的窗口中
          [.trigger(...)]            <-  指定触发器Trigger（可选）
          [.evictor(...)]            <-  指定清除器Evictor(可选)
           .reduce/aggregate/process()      <-  窗口处理函数Window Function

首先，我们要决定是否对一个`DataStream`按照Key进行分组，这一步必须在窗口计算之前进行。经过`keyBy`的数据流将形成多组数据，下游算子的多个实例可以并行计算。`windowAll`不对数据流进行分组，所有数据将发送到下游算子单个实例上。决定是否分组之后，窗口的后续操作基本相同，本文所涉及内容主要针对经过`keyBy`的窗口（Keyed Window），经过`windowAll`的算子是不分组的窗口（Non-Keyed Window），它们的原理和操作与Keyed Window类似，唯一的区别在于所有数据将发送给下游的单个实例，或者说下游算子的并行度为1。

Flink窗口的骨架结构中有两个必须的两个操作：

*  使用窗口分配器（WindowAssigner）将数据流中的元素分配到对应的窗口。
 *  当满足窗口触发条件后，对窗口内的数据使用窗口处理函数（Window Function）进行处理，常用的Window Function有`reduce`、`aggregate`、`process`。

其他的`trigger`、`evictor`则是窗口的触发和销毁过程中的附加选项，主要面向需要更多自定义的高级编程者，如果不设置则会使用默认的配置。

![e66c57c960761542a4ba74d35171284a.png][]

上图是窗口的生命周期示意图，假如我们设置的是一个10分钟的滚动窗口，第一个窗口的起始时间是0:00，结束时间是0:10，后面以此类推。当数据流中的元素流入后，窗口分配器会根据时间（Event Time或Processing Time）分配给相应的窗口。相应窗口满足了触发条件，比如已经到了窗口的结束时间，会触发相应的Window Function进行计算。注意，本图只是一个大致示意图，不同的Window Function的处理方式略有不同。

![39494096012df773ea8d359f708ee69a.png][]

从数据类型上来看，一个`DataStream`经过`keyBy`转换成`KeyedStream`，再经过`window`转换成`WindowedStream`，我们要在之上进行`reduce`、`aggregate`或`process`等Window Function，对数据进行必要的聚合操作。

## WindowAssigner ##

窗口主要有两种，一种基于时间（Time-based Window），一种基于数量（Count-based Window）。本文主要讨论Time-based Window，在Flink源码中，用`TimeWindow`表示。每个`TimeWindow`都有一个开始时间和结束时间，表示一个左闭右开的时间段。Flink为我们提供了一些内置的WindowAssigner，即滚动窗口、滑动窗口和会话窗口，接下来将一一介绍如何使用。

Count-based Window根据事件到达窗口的先后顺序管理窗口，到达窗口的先后顺序和Event Time并不一致，因此Count-based Window的结果具有不确定性。

### 滚动窗口 ###

滚动窗口下窗口之间之间不重叠，且窗口长度是固定的。我们可以用`TumblingEventTimeWindows`和`TumblingProcessingTimeWindows`创建一个基于Event Time或Processing Time的滚动时间窗口。窗口的长度可以用`org.apache.flink.streaming.api.windowing.time.Time`中的`seconds`、`minutes`、`hours`和`days`来设置。

![873fef24aa4f101849a69a869c8ea757.png][]

下面的代码展示了如何使用滚动窗口。代码中最后一个例子，我们在固定长度的基础上设置了偏移（offset）。默认情况下，时间窗口会做一个对齐，比如设置一个一小时的窗口，那么窗口的起止时间是\[0:00:00.000 - 0:59:59.999)。如果设置了offset，那么窗口的起止时间将变为\[0:15:00.000 - 1:14:59.999)。offset可以用在全球不同时区设置上，如果系统时间基于格林威治标准时间（UTC-0），中国的当地时间要设置offset为`Time.hours(-8)`。

val input: DataStream[T] = ...
    
    // tumbling event-time windows
    input
        .keyBy(...)
        .window(TumblingEventTimeWindows.of(Time.seconds(5)))
        .<window function>(...)
    
    // tumbling processing-time windows
    input
        .keyBy(...)
        .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
        .<window function>(...)
    
    // 1 hour tumbling event-time windows offset by 15 minutes.
    input
        .keyBy(...)
        .window(TumblingEventTimeWindows.of(Time.hours(1), Time.minutes(15)))
        .<window function>(...)

有些代码中，设置时间使用的是`timeWindow`而非`window`，比如，`input.keyBy(...).timeWindow(Time.seconds(1))`。`timeWindow`是一种简写。当我们在执行环境设置了`TimeCharacteristic.EventTime`时，Flink对应调用`TumblingEventTimeWindows`；如果我们基于`TimeCharacteristic.ProcessingTime`，Flink使用`TumblingProcessingTimeWindows`。

### 滑动窗口 ###

滑动窗口以一个步长（Slide）不断向前滑动，窗口的长度固定。使用时，我们要设置Slide和Size。Slide的大小决定了Flink以多大的频率来创建新的窗口，Slide较小，窗口的个数会很多。Slide小于窗口的Size时，相邻窗口会重叠，一个事件会被分配到多个窗口；Slide大于Size，有些事件可能被丢掉。

![0c12373fdfc13e7548513298e59f6207.png][]

跟前面介绍的一样，我们使用`Time`类中的时间单位来定义Slide和Size，也可以设置offset。同样，`timeWindow`是一种缩写，根据执行环境中设置的时间语义来选择相应的方法初始化窗口。

val input: DataStream[T] = ...
    
    // sliding event-time windows
    input
        .keyBy(...)
        .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
        .<window function>(...)
    
    // sliding processing-time windows
    input
        .keyBy(<...>)
        .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
        .<window function>(...)
    
    // sliding processing-time windows offset by -8 hours
    input
        .keyBy(<...>)
        .window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8)))
        .<window function>(...)

### 会话窗口 ###

会话窗口根据Session gap切分不同的窗口，当一个窗口在大于Session gap的时间内没有接收到新数据时，窗口将关闭。在这种模式下，窗口的长度是可变的，每个窗口的开始和结束时间并不是确定的。我们可以设置定长的Session gap，也可以使用`SessionWindowTimeGapExtractor`动态地确定Session gap的长度。

![d9892d25fd4079a51b7663e1956ba4ac.png][]

下面的代码展示了如何使用定长和可变的Session gap来建立会话窗口，其中`SessionWindowTimeGapExtractor[T]`的泛型T为数据流的类型，我们可以根据数据流中的元素来生成Session gap。

val input: DataStream[T] = ...
    
    // event-time session windows with static gap
    input
        .keyBy(...)
        .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
        .<window function>(...)
    
    // event-time session windows with dynamic gap
    input
        .keyBy(...)
        .window(EventTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor[T] {
          override def extract(element: T): Long = {
            // determine and return session gap
          }
        }))
        .<window function>(...)
    
    // processing-time session windows with static gap
    input
        .keyBy(...)
        .window(ProcessingTimeSessionWindows.withGap(Time.minutes(10)))
        .<window function>(...)
    
    
    // processing-time session windows with dynamic gap
    input
        .keyBy(...)
        .window(DynamicProcessingTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor[T] {
          override def extract(element: T): Long = {
            // determine and return session gap
          }
        }))
        .<window function>(...)

## 窗口函数 ##

数据经过了`window`和`WindowAssigner`之后，已经被分配到不同的窗口里，接下来，我们要通过窗口函数，在每个窗口上对窗口内的数据进行处理。窗口函数主要分为两种，一种是增量计算，如`reduce`和`aggregate`，一种是全量计算，如`process`。增量计算指的是窗口保存一份中间数据，每流入一个新元素，新元素与中间数据两两合一，生成新的中间数据，再保存到窗口中。全量计算指的是窗口先缓存该窗口所有元素，等到触发条件后对窗口内的全量元素执行计算。

### ReduceFunction ###

使用`reduce`算子时，我们要重写一个`ReduceFunction`。`ReduceFunction`在[DataStream API: keyBy、reduce和aggregations][DataStream API_ keyBy_reduce_aggregations]一文中已经介绍，它接受两个相同类型的输入，生成一个输出，即两两合一地进行汇总操作，生成一个同类型的新元素。在窗口上进行`reduce`的原理与之类似，只不过多了一个窗口状态数据，这个状态数据的数据类型和输入的数据类型是一致的，是之前两两计算的中间结果数据。当数据流中的新元素流入后，`ReduceFunction`将中间结果和新流入数据两两合一，生成新的数据替换之前的状态数据。

case class StockPrice(symbol: String, price: Double)
    
    val input: DataStream[StockPrice] = ...
    
    senv.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
    
    // reduce的返回类型必须和输入类型StockPrice一致
    val sum = input
          .keyBy(s => s.symbol)
          .timeWindow(Time.seconds(10))
          .reduce((s1, s2) => StockPrice(s1.symbol, s1.price + s2.price))

上面的代码使用Lambda表达式对两个元组进行操作，由于对symbol字段进行了`keyBy`，相同symbol的数据都分组到了一起，接着我们将price加和，返回的结果必须也是`StockPrice`类型，否则会报错。

使用`reduce`的好处是窗口的状态数据量非常小，实现一个`ReduceFunction`也相对比较简单，可以使用Lambda表达式，也可以重写函数。缺点是能实现的功能非常有限，因为中间状态数据的数据类型、输入类型以及输出类型三者必须一致，而且只保存了一个中间状态数据，当我们想对整个窗口内的数据进行操作时，仅仅一个中间状态数据是远远不够的。

### AggregateFunction ###

`AggregateFunction`也是一种增量计算窗口函数，也只保存了一个中间状态数据，但`AggregateFunction`使用起来更复杂一些。我们看一下它的源码定义：

public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable {
    
       // 在一次新的aggregate发起时，创建一个新的Accumulator，Accumulator是我们所说的中间状态数据，简称ACC
       // 这个函数一般在初始化时调用
       ACC createAccumulator();
    
       // 当一个新元素流入时，将新元素与状态数据ACC合并，返回状态数据ACC
       ACC add(IN value, ACC accumulator);
    
       // 将两个ACC合并
       ACC merge(ACC a, ACC b);
    
       // 将中间数据转成结果数据
       OUT getResult(ACC accumulator);
    
    }

输入类型是IN，输出类型是OUT，中间状态数据是ACC，这样复杂的设计主要是为了解决输入类型、中间状态和输出类型不一致的问题，同时ACC可以自定义，我们可以在ACC里构建我们想要的数据结构。比如我们要计算一个窗口内某个字段的平均值，那么ACC中要保存总和以及个数，下面是一个平均值的示例：

case class StockPrice(symbol: String, price: Double)
    
    // IN: StockPrice
    // ACC：(String, Double, Int) - (symbol, sum, count)
    // OUT: (String, Double) - (symbol, average)
    class AverageAggregate extends AggregateFunction[StockPrice, (String, Double, Int), (String, Double)] {
    
      override def createAccumulator() = ("", 0, 0)
    
      override def add(item: StockPrice, accumulator: (String, Double, Int)) =
      (item.symbol, accumulator._2 + item.price, accumulator._3 + 1)
    
      override def getResult(accumulator:(String, Double, Int)) = (accumulator._1 ,accumulator._2 / accumulator._3)
    
      override def merge(a: (String, Double, Int), b: (String, Double, Int)) =
      (a._1 ,a._2 + b._2, a._3 + b._3)
    }
    
    senv.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
    
    val input: DataStream[StockPrice] = ...
    
    val average = input
          .keyBy(s => s.symbol)
          .timeWindow(Time.seconds(10))
          .aggregate(new AverageAggregate)

这几个函数的工作流程如下图所示。在计算之前要创建一个新的ACC，这时ACC还没有任何实际表示意义，当有新数据流入时，Flink会调用`add`方法，更新ACC，并返回最新的ACC，ACC是一个中间状态数据。当有一些跨节点的ACC融合时，Flink会调用`merge`，生成新的ACC。当所有的ACC最后融合为一个ACC后，Flink调用`getResult`生成结果。

![6c0662b0c4a128db63ff509cc8829bb7.png][]

### ProcessWindowFunction ###

与前两种方法不同，`ProcessWindowFunction`要对窗口内的全量数据都缓存。在Flink所有API中，`process`算子以及其对应的函数是最底层的实现，使用这些函数能够访问一些更加底层的数据，比如，直接操作状态等。它在源码中的定义如下：

/**
     * IN   输入类型
     * OUT  输出类型
     * KEY  keyBy中按照Key分组，Key的类型
     * W    窗口的类型
     */
    public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> extends AbstractRichFunction {
    
      /**
       * 对一个窗口内的元素进行处理，窗口内的元素缓存在Iterable<IN>，进行处理后输出到Collector<OUT>中
       * 我们可以输出一到多个结果
       */
        public abstract void process(KEY key, Context context, Iterable<IN> elements, Collector<OUT> out) throws Exception;
    
      /** 
        * 当窗口执行完毕被清理时，删除各类状态数据。
        */
        public void clear(Context context) throws Exception {}
    
      /**
       * 一个窗口的上下文，包含窗口的一些元数据、状态数据等。
       */
        public abstract class Context implements java.io.Serializable {
    
        // 返回当前正在处理的Window
            public abstract W window();
    
        // 返回当前Process Time
            public abstract long currentProcessingTime();
    
        // 返回当前Event Time对应的Watermark
            public abstract long currentWatermark();
    
        // 返回某个Key下的某个Window的状态
            public abstract KeyedStateStore windowState();
    
        // 返回某个Key下的全局状态
            public abstract KeyedStateStore globalState();
    
        // 迟到数据发送到其他位置
            public abstract <X> void output(OutputTag<X> outputTag, X value);
        }
    }

使用时，Flink将某个Key下某个窗口的所有元素都缓存在`Iterable<IN>`中，我们需要对其进行处理，然后用`Collector<OUT>`收集输出。我们可以使用`Context`获取窗口内更多的信息，包括时间、状态、迟到数据发送位置等。

下面的代码是一个`ProcessWindowFunction`的简单应用，我们对价格出现的次数做了统计，选出出现次数最多的输出出来。

case class StockPrice(symbol: String, price: Double)
    
    class FrequencyProcessFunction extends ProcessWindowFunction[StockPrice, (String, Double), String, TimeWindow] {
    
      override def process(key: String, context: Context, elements: Iterable[StockPrice], out: Collector[(String, Double)]): Unit = {
    
        // 股票价格和该价格出现的次数
        var countMap = scala.collection.mutable.Map[Double, Int]()
    
        for(element <- elements) {
          val count = countMap.getOrElse(element.price, 0)
          countMap(element.price) = count + 1
        }
    
        // 按照出现次数从高到低排序
        val sortedMap = countMap.toSeq.sortWith(_._2 > _._2)
    
        // 选出出现次数最高的输出到Collector
        if (sortedMap.size > 0) {
          out.collect((key, sortedMap(0)._1))
        }
    
      }
    }
    
    senv.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
    
    val input: DataStream[StockPrice] = ...
    
    val frequency = input
          .keyBy(s => s.symbol)
          .timeWindow(Time.seconds(10))
          .process(new FrequencyProcessFunction)

`Context`中有两种状态，一种是针对Key的全局状态，它是跨多个窗口的，多个窗口都可以访问；另一种是该Key下单窗口的状态，单窗口的状态只保存该窗口的数据，主要是针对`process`函数多次被调用的场景，比如处理迟到数据或自定义Trigger等场景。当使用单个窗口的状态时，要在`clear`函数中清理状态。

`ProcessWindowFunction`相比`AggregateFunction`和`ReduceFunction`的应用场景更广，能解决的问题也更复杂。但`ProcessWindowFunction`需要将窗口中所有元素作为状态存储起来，这将占用大量的存储资源，尤其是在数据量大窗口多的场景下，使用不慎可能导致整个程序宕机。比如，每天的数据在TB级，我们需要Slide为十分钟Size为一小时的滑动窗口，这种设置会导致窗口数量很多，而且一个元素会被复制好多份分给每个所属的窗口，这将带来巨大的内存压力。

### ProcessWindowFunction与增量计算相结合 ###

当我们既想访问窗口里的元数据，又不想缓存窗口里的所有数据时，可以将`ProcessWindowFunction`与增量计算函数相`reduce`和`aggregate`结合。对于一个窗口来说，Flink先增量计算，窗口关闭前，将增量计算结果发送给`ProcessWindowFunction`作为输入再进行处理。

下面的代码中，Lambda函数对所有内容进行最大值和最小值的处理，这一步是增量计算。计算的结果以数据类型`(String, Double, Double)`传递给`WindowEndProcessFunction`，`WindowEndProcessFunction`只需要将窗口结束的时间戳添加到结果`MaxMinPrice`中即可。

case class StockPrice(symbol: String, price: Double)
    
    case class MaxMinPrice(symbol: String, max: Double, min: Double, windowEndTs: Long)
    
    class WindowEndProcessFunction extends ProcessWindowFunction[(String, Double, Double), MaxMinPrice, String, TimeWindow] {
    
      override def process(key: String,
                           context: Context,
                           elements: Iterable[(String, Double, Double)],
                           out: Collector[MaxMinPrice]): Unit = {
        val maxMinItem = elements.head
        val windowEndTs = context.window.getEnd
        out.collect(MaxMinPrice(key, maxMinItem._2, maxMinItem._3, windowEndTs))
      }
    
    }
    
    senv.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
    
    val input: DataStream[StockPrice] = ...
    
    // reduce的返回类型必须和输入类型相同
    // 为此我们将StockPrice拆成一个三元组 (股票代号，最大值、最小值)
    val maxMin = input
    .map(s => (s.symbol, s.price, s.price))
    .keyBy(s => s._1)
    .timeWindow(Time.seconds(10))
    .reduce(
      ((s1: (String, Double, Double), s2: (String, Double, Double)) => (s1._1, Math.max(s1._2, s2._2), Math.min(s1._3, s2._3))),
      new WindowEndProcessFunction
    )

## Trigger ##

触发器（Trigger）决定了何时启动Window Function来处理窗口中的数据以及何时将窗口内的数据清理。增量计算窗口函数对每个新流入的数据直接进行聚合，Trigger决定了在窗口结束时将聚合结果发送出去；全量计算窗口函数需要将窗口内的元素缓存，Trigger决定了在窗口结束时对所有元素进行计算然后将结果发送出去。每个窗口都有一个默认的Trigger，比如前文这些例子都是基于Processing Time的时间窗口，当到达窗口的结束时间时，Trigger以及对应的计算被触发。如果我们有一些个性化的触发条件，比如窗口中遇到某些特定的元素、元素总数达到一定数量或窗口中的元素到达时满足某种特定的模式时，我们可以自定义一个Trigger。我们甚至可以在Trigger中定义一些提前计算的逻辑，比如在Event Time语义中，虽然Watermark还未到达，但是我们可以定义提前计算输出的逻辑，以快速获取计算结果，获得更低的延迟。

我们先看Trigger返回一个什么样的结果。当满足某个条件，Trigger会返回一个名为`TriggerResult`的结果：

*  CONTINUE：什么都不做。
 *  FIRE：启动计算并将结果发送给下游，不清理窗口数据。
 *  PURGE：清理窗口数据但不执行计算。
 *  FIRE\_AND\_PURGE：启动计算，发送结果然后清理窗口数据。

在继续介绍Trigger的使用之前，我们可以先了解一下定时器（Timer）的使用方法。我们可以把Timer理解成一个闹钟，使用前先注册未来一个时间，当时间到达时，就像闹钟会响一样，程序会启用一个回调函数，来执行某个时间相关的任务。对于自定义Trigger来说，我们需要考虑注册时间的逻辑，当到达这个时间时，Flink会启动Window Function，清理窗口数据。

`WindowAssigner`都有一个默认的`Trigger`。比如基于Event Time的窗口会有一个`EventTimeTrigger`，每当窗口的Watermark时间戳到达窗口的结束时间，Trigger会发送`FIRE`。此外，`ProcessingTimeTrigger`对应Processing Time窗口，`CountTrigger`对应Count-based窗口。

当这些已有的Trigger无法满足我们的需求时，我们需要自定义Trigger，接下来我们看一下Flink的Trigger源码。

/**
        * T为元素类型
        * W为窗口
      */
    public abstract class Trigger<T, W extends Window> implements Serializable {
    
        /**
         * 当某窗口增加一个元素时调用onElement方法，返回一个TriggerResult
         */
        public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception;
    
        /**
         * 当一个基于Processing Time的Timer触发了FIRE时调用onProcessTime方法
         */
        public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;
    
        /**
         * 当一个基于Event Time的Timer触发了FIRE时调用onEventTime方法
         */
        public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;
    
        /**
         * 如果这个Trigger支持状态合并，则返回true
         */
        public boolean canMerge() {
            return false;
        }
    
        /**
         * 当多个窗口被合并时调用onMerge
         */
        public void onMerge(W window, OnMergeContext ctx) throws Exception {
            throw new UnsupportedOperationException("This trigger does not support merging.");
        }
    
        /**
         * 当窗口数据被清理时，调用clear方法来清理所有的Trigger状态数据
         */
        public abstract void clear(W window, TriggerContext ctx) throws Exception
    
        /**
         * 上下文，保存了时间、状态、监控以及定时器
         */
        public interface TriggerContext {
    
            /**
             * 返回当前Processing Time
             */
            long getCurrentProcessingTime();
    
            /**
             * 返回MetricGroup 
             */
            MetricGroup getMetricGroup();
    
            /**
             * 返回当前Watermark时间
             */
            long getCurrentWatermark();
    
            /**
             * 将某个time注册为一个Timer，当系统时间到达time这个时间点时，onProcessingTime方法会被调用
             */
            void registerProcessingTimeTimer(long time);
    
            /**
             * 将某个time注册为一个Timer，当Watermark时间到达time这个时间点时，onEventTime方法会被调用
             */
            void registerEventTimeTimer(long time);
    
            /**
             * 将注册的Timer删除
             */
            void deleteProcessingTimeTimer(long time);
    
            /**
             * 将注册的Timer删除
             */
            void deleteEventTimeTimer(long time);
    
            /**
             * 获取该窗口Trigger下的状态
             */
            <S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor);
    
        }
    
        /**
         * 将多个窗口下Trigger状态合并
         */
        public interface OnMergeContext extends TriggerContext {
            <S extends MergingState<?, ?>> void mergePartitionedState(StateDescriptor<S, ?> stateDescriptor);
        }
    }

接下来我们以一个提前计算的案例来解释如何使用自定义的Trigger。在股票或任何交易场景中，我们比较关注价格急跌的情况，默认窗口长度是60秒，如果价格跌幅超过5%，则立即执行Window Function，如果价格跌幅在1%到5%之内，那么10秒后触发Window Function。

class MyTrigger extends Trigger[StockPrice, TimeWindow] {
    
      override def onElement(element: StockPrice,
                             time: Long,
                             window: TimeWindow,
                             triggerContext: Trigger.TriggerContext): TriggerResult = {
        val lastPriceState: ValueState[Double] = triggerContext.getPartitionedState(new ValueStateDescriptor[Double]("lastPriceState", classOf[Double]))
    
        // 设置返回默认值为CONTINUE
        var triggerResult: TriggerResult = TriggerResult.CONTINUE
    
        // 第一次使用lastPriceState时状态是空的,需要先进行判断
        // 状态数据由Java端生成，如果是空，返回一个null
        // 如果直接使用Scala的Double，需要使用下面的方法判断是否为空
        if (Option(lastPriceState.value()).isDefined) {
          if ((lastPriceState.value() - element.price) > lastPriceState.value() * 0.05) {
            // 如果价格跌幅大于5%，直接FIRE_AND_PURGE
            triggerResult = TriggerResult.FIRE_AND_PURGE
          } else if ((lastPriceState.value() - element.price) > lastPriceState.value() * 0.01) {
            val t = triggerContext.getCurrentProcessingTime + (10 * 1000 - (triggerContext.getCurrentProcessingTime % 10 * 1000))
            // 给10秒后注册一个Timer
            triggerContext.registerProcessingTimeTimer(t)
          }
        }
        lastPriceState.update(element.price)
        triggerResult
      }
    
      // 我们不用EventTime，直接返回一个CONTINUE
      override def onEventTime(time: Long, window: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = {
        TriggerResult.CONTINUE
      }
    
      override def onProcessingTime(time: Long, window: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = {
        TriggerResult.FIRE_AND_PURGE
      }
    
      override def clear(window: TimeWindow, triggerContext: Trigger.TriggerContext): Unit = {
        val lastPrice: ValueState[Double] = triggerContext.getPartitionedState(new ValueStateDescriptor[Double]("lastPrice", classOf[Double]))
        lastPrice.clear()
      }
    }
    
    senv.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
    
    val input: DataStream[StockPrice] = ...
    
    val average = input
          .keyBy(s => s.symbol)
          .timeWindow(Time.seconds(60))
          .trigger(new MyTrigger)
          .aggregate(new AverageAggregate)

在自定义Trigger时，如果使用了状态，一定要使用`clear`方法将状态数据清理，否则随着窗口越来越多，状态数据会越积越多。

## Evictor ##

清除器（Evictor）是在`WindowAssigner`和`Trigger`的基础上的一个可选选项，用来清除一些数据。我们可以在Window Function执行前或执行后调用Evictor。

/**
        * T为元素类型
        * W为窗口
      */
    public interface Evictor<T, W extends Window> extends Serializable {
    
        /**
         * 在Window Function前调用
       */
        void evictBefore(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);
    
        /**
         * 在Window Function后调用
         */
        void evictAfter(Iterable<TimestampedValue<T>> elements, int size, W window, EvictorContext evictorContext);
    
    
        /**
         * Evictor的上下文
         */
        interface EvictorContext {
    
            long getCurrentProcessingTime();
    
            MetricGroup getMetricGroup();
    
            long getCurrentWatermark();
        }
    }

`evictBefore`和`evictAfter`分别在Window Function之前和之后被调用，窗口的所有元素被放在了`Iterable<TimestampedValue<T>>`，我们要实现自己的清除逻辑。当然，对于增量计算的`ReduceFunction`和`AggregateFunction`，我们没必要使用Evictor。

Flink提供了几个实现好的Evictor：

*  `CountEvictor`保留一定数目的元素，多余的元素按照从前到后的顺序先后清理。
 *  `TimeEvictor`保留一个时间段的元素，早于这个时间段的元素会被清理。

[https_lulaoshi.info_flink_chapter-time-window_window.html]: https://link.zhihu.com/?target=https%3A//lulaoshi.info/flink/chapter-time-window/window.html
[e66c57c960761542a4ba74d35171284a.png]: /images/20221024/47153270f4d247df869cbe40ce594101.png
[39494096012df773ea8d359f708ee69a.png]: /images/20221024/407b04a7e5ff497abd53e6897baccc29.png
[873fef24aa4f101849a69a869c8ea757.png]: /images/20221024/230274ba1f0b47bca7771d295f69ceed.png
[0c12373fdfc13e7548513298e59f6207.png]: /images/20221024/ed6fde9d407c495494b96da904a5c3cf.png
[d9892d25fd4079a51b7663e1956ba4ac.png]: /images/20221024/2fb04837eab04159bc39846aa8a4e93b.png
[DataStream API_ keyBy_reduce_aggregations]: https://link.zhihu.com/?target=https%3A//mp.weixin.qq.com/s/2vcKteQIyj31sVrSg1R_2Q
[6c0662b0c4a128db63ff509cc8829bb7.png]: /images/20221024/46768c3b1de24be3945701787dea640a.png