Flink視窗背景

Flink認為Batch是Streaming的一個特例，因此Flink底層引擎是一個流式引擎，在上面實作了流處理和批處理，而Window就是從Streaming到Batch的橋梁，通俗講，Window是用來對一個無限的流設定一個有限的集合，從而在有界的資料集上進行操作的一種機制，流上的集合由Window來劃定范圍，比如“計算過去10分鐘”或者“最后50個元素的和”，Window可以由時間（Time Window）（比如每30s）或者資料（Count Window）（如每100個元素）驅動，DataStream API提供了Time和Count的Window，

一個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

Flink視窗的骨架結構中有兩個必須的兩個操作：

• 使用視窗分配器（WindowAssigner）將資料流中的元素分配到對應的視窗，
• 當滿足視窗觸發條件后，對視窗內的資料使用視窗處理函式（Window Function）進行處理，常用的Window Function有reduce、aggregate、process

滾動視窗

基于時間驅動

將資料依據固定的視窗長度對資料進行切分，滾動視窗下視窗之間之間不重疊，且視窗長度是固定的，我們可以用TumblingEventTimeWindows和TumblingProcessingTimeWindows創建一個基于Event Time或Processing Time的滾動時間視窗，視窗的長度可以用org.apache.flink.streaming.api.windowing.time.Time中的seconds、minutes、hours和days來設定，

//關鍵處理案例
KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = mapStream.keyBy(0);
// 基于時間驅動，每隔10s劃分一個視窗
WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> timeWindow =
keyedStream.timeWindow(Time.seconds(10));
// 基于事件驅動, 每相隔3個事件(即三個相同key的資料), 劃分一個視窗進行計算
// WindowedStream<Tuple2<String, Integer>, Tuple, GlobalWindow> countWindow =
keyedStream.countWindow(3);
// apply是視窗的應用函式，即apply里的函式將應用在此視窗的資料上，
timeWindow.apply(new MyTimeWindowFunction()).print();
// countWindow.apply(new MyCountWindowFunction()).print();

基于事件驅動

當我們想要每100個用戶的購買行為作為驅動，那么每當視窗中填滿100個”相同”元素了，就會對視窗進行計算，很好理解，下面是一個實作案例

public class MyCountWindowFunction implements WindowFunction<Tuple2<String, Integer>,
  String, Tuple, GlobalWindow> {
    @Override
    public void apply(Tuple tuple, GlobalWindow window, Iterable<Tuple2<String, Integer>>
      input, Collector<String> out) throws Exception {
      SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
      int sum = 0;
      for (Tuple2<String, Integer> tuple2 : input){
        sum += tuple2.f1;
      }
      
      //無用的時間戳，默認值為： Long.MAX_VALUE,因為基于事件計數的情況下，不關心時間，
      long maxTimestamp = window.maxTimestamp();
      out.collect("key:" + tuple.getField(0) + " value: " + sum + "| maxTimeStamp :"+ maxTimestamp + "," + format.format(maxTimestamp)
      );
  }
}

滑動時間視窗

動視窗是固定視窗的更廣義的一種形式，滑動視窗由固定的視窗長度和滑動間隔組成，特點：視窗長度固定，可以有重疊，滑動視窗以一個步長（Slide）不斷向前滑動，視窗的長度固定，使用時，我們要設定Slide和Size，Slide的大小決定了Flink以多大的頻率來創建新的視窗，Slide較小，視窗的個數會很多，Slide小于視窗的Size時，相鄰視窗會重疊，一個事件會被分配到多個視窗；Slide大于Size，有些事件可能被丟掉

基于時間的滾動視窗

//基于時間驅動，每隔5s計算一下最近10s的資料
// WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> timeWindow =
keyedStream.timeWindow(Time.seconds(10), Time.seconds(5));
SingleOutputStreamOperator<String> applyed = countWindow.apply(new WindowFunction<Tuple3<String, String, String>, String, String, GlobalWindow>() {
    @Override
    public void apply(String s, GlobalWindow window, Iterable<Tuple3<String, String, String>> input, Collector<String> out) throws Exception {
        Iterator<Tuple3<String, String, String>> iterator = input.iterator();
        StringBuilder sb = new StringBuilder();
        while (iterator.hasNext()) {
            Tuple3<String, String, String> next = iterator.next();
            sb.append(next.f0 + ".." + next.f1 + ".." + next.f2);
        }
//                window.
        out.collect(sb.toString());
    }
}); 

基于事件的滾動視窗

/**
* 滑動視窗：視窗可重疊
* 1、基于時間驅動
* 2、基于事件驅動
*/
WindowedStream<Tuple3<String, String, String>, String, GlobalWindow> countWindow = keybyed.countWindow(3,2);

SingleOutputStreamOperator<String> applyed = countWindow.apply(new WindowFunction<Tuple3<String, String, String>, String, String, GlobalWindow>() {
    @Override
    public void apply(String s, GlobalWindow window, Iterable<Tuple3<String, String, String>> input, Collector<String> out) throws Exception {
        Iterator<Tuple3<String, String, String>> iterator = input.iterator();
        StringBuilder sb = new StringBuilder();
        while (iterator.hasNext()) {
            Tuple3<String, String, String> next = iterator.next();
            sb.append(next.f0 + ".." + next.f1 + ".." + next.f2);
        }
//                window.
        out.collect(sb.toString());
    }
});

會話時間視窗

由一系列事件組合一個指定時間長度的timeout間隙組成，類似于web應用的session，也就是一段時間沒有接收到新資料就會生成新的視窗，在這種模式下，視窗的長度是可變的，每個視窗的開始和結束時間并不是確定的，我們可以設定定長的Session gap，也可以使用SessionWindowTimeGapExtractor動態地確定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>(...)

視窗函式

在視窗劃分完畢后，就是要對視窗內的資料進行處理，一是增量計算對應reduce 和aggregate，二是全量計算對應process ,增量計算指的是視窗保存一份中間資料，每流入一個新元素，新元素與中間資料兩兩合一，生成新的中間資料，再保存到視窗中，全量計算指的是視窗先快取該視窗所有元素，等到觸發條件后對視窗內的全量元素執行計算

參考

https://cloud.tencent.com/developer/article/1584926

吳邪，小三爺，混跡于后臺，大資料，人工智能領域的小菜鳥，
更多請關注

標籤：其他

上一篇：AI入門與專案實戰（基礎概念）

下一篇：Beam Search快速理解及代碼決議(下)