Flink 框架中支持事件時間、攝入時間和處理時間三種。而當我們在流式計算環境中數據從 Source 產生，再到轉換和輸出，這個過程由于網絡和反壓的原因會導致消息亂序。因此，需要有一個機制來解決這個問題，這個特別的機制就是“水位線”。

Flink 的窗口和時間
根據窗口數據劃分的不同，目前 Flink 支持如下 3 種：

滾動窗口，窗口數據有固定的大小，窗口中的數據不會疊加；

滑動窗口，窗口數據有固定的大小，并且有生成間隔；

會話窗口，窗口數據沒有固定的大小，根據用戶傳入的參數進行劃分，窗口數據無疊加。

Flink 中的時間分為三種：

事件時間（Event Time），即事件實際發生的時間；
攝入時間（Ingestion Time），事件進入流處理框架的時間；
處理時間（Processing Time），事件被處理的時間。
下面的圖詳細說明了這三種時間的區別和聯系：

事件時間（Event Time

事件時間（Event Time）指的是數據產生的時間，這個時間一般由數據生產方自身攜帶，比如 Kafka 消息，每個生成的消息中自帶一個時間戳代表每條數據的產生時間。Event Time 從消息的產生就誕生了，不會改變，也是我們使用最頻繁的時間。

利用 Event Time 需要指定如何生成事件時間的“水印”，并且一般和窗口配合使用，具體會在下面的“水印”內容中詳細講解。

我們可以在代碼中指定 Flink 系統使用的時間類型為 EventTime：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//設置時間屬性為 EventTime
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

DataStream<MyEvent> stream = env.addSource(new FlinkKafkaConsumer09<MyEvent>(topic, schema, props));

stream
    .keyBy( (event) -> event.getUser() )
    .timeWindow(Time.hours(1))
    .reduce( (a, b) -> a.add(b) )
    .addSink(...);

Flink 注冊 EventTime 是通過 InternalTimerServiceImpl.registerEventTimeTimer 來實現的

可以看到，該方法有兩個入參：namespace 和 time，其中 time 是觸發定時器的時間，namespace 則被構造成為一個 TimerHeapInternalTimer 對象，然后將其放入 KeyGroupedInternalPriorityQueue 隊列中。

那么 Flink 什么時候會使用這些 timer 觸發計算呢？答案在這個方法里：

InternalTimeServiceImpl.advanceWatermark。
public void advanceWatermark(long time) throws Exception {
   currentWatermark = time;

   InternalTimer<K, N> timer;

   while ((timer = eventTimeTimersQueue.peek()) != null && timer.getTimestamp() <= time) {
      eventTimeTimersQueue.poll();
      keyContext.setCurrentKey(timer.getKey());
      triggerTarget.onEventTime(timer);
   }
}

這個方法中的 while 循環部分會從 eventTimeTimersQueue 中依次取出觸發時間小于參數 time 的所有定時器，調用 triggerTarget.onEventTime() 方法進行觸發。

這就是 EventTime 從注冊到觸發的流程。

處理時間（Processing Time）
處理時間（Processing Time）指的是數據被 Flink 框架處理時機器的系統時間，Processing Time 是 Flink 的時間系統中最簡單的概念，但是這個時間存在一定的不確定性，比如消息到達處理節點延遲等影響。

我們同樣可以在代碼中指定 Flink 系統使用的時間為 Processing Time：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

同樣，也可以在源碼中找到 Flink 是如何注冊和使用 Processing Time 的。

registerProcessingTimeTimer() 方法為我們展示了如何注冊一個 ProcessingTime 定時器：
每當一個新的定時器被加入到 processingTimeTimersQueue 這個優先級隊列中時，如果新來的 Timer 時間戳更小，那么更小的這個 Timer 會被重新注冊 ScheduledThreadPoolExecutor 定時執行器上。

Processing Time 被觸發是在 InternalTimeServiceImpl 的 onProcessingTime() 方法中：

一直循環獲取時間小于入參 time 的所有定時器，并運行 triggerTarget 的 onProcessingTime() 方法。

攝入時間（Ingestion Time）
攝入時間（Ingestion Time）是事件進入 Flink 系統的時間，在 Flink 的 Source 中，每個事件會把當前時間作為時間戳，后續做窗口處理都會基于這個時間。理論上 Ingestion Time 處于 Event Time 和 Processing Time之間。

與事件時間相比，攝入時間無法處理延時和無序的情況，但是不需要明確執行如何生成 watermark。在系統內部，攝入時間采用更類似于事件時間的處理方式進行處理，但是有自動生成的時間戳和自動的 watermark。

可以防止 Flink 內部處理數據是發生亂序的情況，但無法解決數據到達 Flink 之前發生的亂序問題。如果需要處理此類問題，建議使用 EventTime。

Ingestion Time 的時間類型生成相關的代碼在 AutomaticWatermarkContext 中。

水位線（WaterMark）
水位線（WaterMark）是 Flink 框架中最晦澀難懂的概念之一，有很大一部分原因是因為翻譯的原因。

WaterMark 在正常的英文翻譯中是水位，但是在 Flink 框架中，翻譯為“水位線”更為合理，它在本質上是一個時間戳。

在上面的時間類型中我們知道，Flink 中的時間：
EventTime 每條數據都攜帶時間戳；

ProcessingTime 數據不攜帶任何時間戳的信息；
IngestionTime 和 EventTime 類似，不同的是 Flink 會使用系統時間作為時間戳綁定到每條數據，可以防止 Flink 內部處理數據是發生亂序的情況，但無法解決數據到達 Flink 之前發生的亂序問題。
所以，我們在處理消息亂序的情況時，會用 EventTime 和 WaterMark 進行配合使用。

首先我們要明確幾個基本問題。

水印的本質是什么
水印的出現是為了解決實時計算中的數據亂序問題，它的本質是 DataStream 中一個帶有時間戳的元素。

如果 Flink 系統中出現了一個 WaterMark T，那么就意味著 EventTime < T 的數據都已經到達，窗口的結束時間和 T 相同的那個窗口被觸發進行計算了。

也就是說：水印是 Flink 判斷遲到數據的標準，同時也是窗口觸發的標記。

在程序并行度大于 1 的情況下，會有多個流產生水印和窗口，這時候 Flink 會選取時間戳最小的水印。

水位線是如何生成的

Flink 提供了 assignTimestampsAndWatermarks() 方法來實現水印的提取和指定，該方法接受的入參有 AssignerWithPeriodicWatermarks 和 AssignerWithPunctuatedWatermarks 兩種。

整體的類圖如下：

水位線種類

周期性水位線

我們在使用 AssignerWithPeriodicWatermarks 周期生成水印時，周期默認的時間是 200ms，這個時間的指定位置為：

@PublicEvolving
public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) {
    this.timeCharacteristic = Preconditions.checkNotNull(characteristic);
    if (characteristic == TimeCharacteristic.ProcessingTime) {
        getConfig().setAutoWatermarkInterval(0);
    } else {
        getConfig().setAutoWatermarkInterval(200);
    }
}

是否還記得上面我們在講時間類型時會通過 env.setStreamTimeCharacteristic() 方法指定 Flink 系統的時間類型，這個 setStreamTimeCharacteristic() 方法中會做判斷，如果用戶傳入的是 TimeCharacteristic.eventTime 類型，那么 AutoWatermarkInterval 的值則為 200ms ，如上述代碼所示。當前我們也可以使用 ExecutionConfig.setAutoWatermarkInterval() 方法來指定自動生成的時間間隔。

在上述的類圖中可以看出，我們需要通過 TimestampAssigner 的 extractTimestamp() 方法來提取 EventTime。

Flink 在這里提供了 3 種提取 EventTime() 的方法，分別是：

AscendingTimestampExtractor
BoundedOutOfOrdernessTimestampExtractor
IngestionTimeExtractor
這三種方法中 BoundedOutOfOrdernessTimestampExtractor() 用的最多，需特別注意，在這個方法中的 maxOutOfOrderness 參數，該參數指的是允許數據亂序的時間范圍。簡單說，這種方式允許數據遲到 maxOutOfOrderness 這么長的時間。

    public BoundedOutOfOrdernessTimestampExtractor(Time maxOutOfOrderness) {
        if (maxOutOfOrderness.toMilliseconds() < 0) {
            throw new RuntimeException("Tried to set the maximum allowed " +
                "lateness to " + maxOutOfOrderness + ". This parameter cannot be negative.");
        }
        this.maxOutOfOrderness = maxOutOfOrderness.toMilliseconds();
        this.currentMaxTimestamp = Long.MIN_VALUE + this.maxOutOfOrderness;
    }

    public abstract long extractTimestamp(T element);

    @Override
    public final Watermark getCurrentWatermark() {
        long potentialWM = currentMaxTimestamp - maxOutOfOrderness;
        if (potentialWM >= lastEmittedWatermark) {
            lastEmittedWatermark = potentialWM;
        }
        return new Watermark(lastEmittedWatermark);
    }

    @Override
    public final long extractTimestamp(T element, long previousElementTimestamp) {
        long timestamp = extractTimestamp(element);
        if (timestamp > currentMaxTimestamp) {
            currentMaxTimestamp = timestamp;
        }
        return timestamp;
    }

PunctuatedWatermark水位線

這種水位線的生成方式 Flink 沒有提供內置實現，它適用于根據接收到的消息判斷是否需要產生水位線的情況，用這種水印生成的方式并不多見。

舉個簡單的例子，假如我們發現接收到的數據 MyData 中以字符串 watermark 開頭則產生一個水位線：

data.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<UserActionRecord>() {

      @Override
      public Watermark checkAndGetNextWatermark(MyData data, long l) {
        return data.getRecord.startsWith("watermark") ? new Watermark(l) : null;
      }

      @Override
      public long extractTimestamp(MyData data, long l) {
        return data.getTimestamp();
      }
    });
    
class MyData{
    private String record;
    private Long timestamp;
    public String getRecord() {
        return record;
    }
    public void setRecord(String record) {
        this.record = record;
    }
    public Timestamp getTimestamp() {
        return timestamp;
    }
    public void setTimestamp(Timestamp timestamp) {
        this.timestamp = timestamp;
    }
}

案例
我們上面講解了 Flink 關于水位線和時間的生成，以及使用，下面舉一個例子來講解。

模擬一個實時接收 Socket 的 DataStream 程序，代碼中使用 AssignerWithPeriodicWatermarks 來設置水位線，將接收到的數據進行轉換，分組并且在一個10
秒，間隔是5秒的滑動窗口內獲取該窗口中第二個元素最小的那條數據。

package com.wyh.windowsApi

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.functions.{AssignerWithPeriodicWatermarks, AssignerWithPunctuatedWatermarks}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time


object WindowTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //周期性生成watermark 默認是200毫秒
    env.getConfig.setAutoWatermarkInterval(100L)

    /**
      * 從文件中讀取數據
      *
      *
      */
    //val stream = env.readTextFile("F:\flink-study\wyhFlinkSD\data\sensor.txt")

    val stream = env.socketTextStream("localhost", 7777)

    //Transform操作
    val dataStream: DataStream[SensorReading] = stream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
      //===到來的數據是升序的，準時發車，用assignAscendingTimestamps
      //指定哪個字段是時間戳 需要的是毫秒 * 1000
      //      .assignAscendingTimestamps(_.timestamp * 1000)
      //===處理亂序數據
      //      .assignTimestampsAndWatermarks(new MyAssignerPeriodic())
      //==底層也是周期性生成的一個方法 處理亂序數據 延遲1秒種生成水位 同時分配水位和時間戳 括號里傳的是等待延遲的時間
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
      override def extractTimestamp(t: SensorReading): Long = {
        t.timestamp * 1000
      }
    })

    //統計10秒內的最小溫度
    val minTemPerWindowStream = dataStream
      .map(data => (data.id, data.temperature))
      .keyBy(0)
      //      .timeWindow(Time.seconds(10)) //開時間窗口  滾動窗口 沒有數據的窗口不會觸發
      //左閉右開 包含開始 不包含結束 延遲1秒觸發的那個時間的數據不包含
      //可以直接調用底層方法，第三個參數傳offset代表時區
      //.window(SlidingEventTimeWindows.of(Time.seconds(15),Time.seconds(5),Time.hours(-8)))
      .timeWindow(Time.seconds(15), Time.seconds(5)) //滑動窗口，每隔5秒輸出一次
      .reduce((data1, data2) => (data1._1, data1._2.min(data2._2))) //用reduce做增量聚合


    minTemPerWindowStream.print("min temp")

    dataStream.print("input data")

    env.execute("window Test")

  }

}


//設置水位線（水印） 這里有兩種方式實現
//一種是周期性生成 一種是以數據的某種特性進行生成水位線（水印）
/**
  * 周期性生成watermark 默認200毫秒
  */
class MyAssignerPeriodic() extends AssignerWithPeriodicWatermarks[SensorReading] {
  val bound: Long = 60 * 1000
  var maxTs: Long = Long.MaxValue

  override def getCurrentWatermark: Watermark = {
    //定義一個規則進行生成
    new Watermark(maxTs - bound)
  }

  //用什么抽取這個時間戳
  override def extractTimestamp(t: SensorReading, l: Long): Long = {
    //保存當前最大的時間戳
    maxTs = maxTs.max(t.timestamp)
    t.timestamp * 1000
  }
}


/**
  * 亂序生成watermark
  * 每來一條數據就生成一個watermark
  */
class MyAssignerPunctuated() extends AssignerWithPunctuatedWatermarks[SensorReading] {
  override def checkAndGetNextWatermark(t: SensorReading, l: Long): Watermark = {
    new Watermark(l)
  }

  override def extractTimestamp(t: SensorReading, l: Long): Long = {
    t.timestamp * 1000
  }
}

總結

以上是生活随笔為你收集整理的Flink学习（十四） Flink 窗口、时间和水位线的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。