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

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>(...)

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>(...)

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>(...)

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))

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

}

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)

/**
 * 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);
    }
}

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)

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
)

/**
    * 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);
    }
}

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)

/**
    * 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();
    }
}