第08講：Flink 窗口、時間和水印

Flink系列文章

  1. 第01講：Flink 的應用場景和架構模型
  2. 第02講：Flink 入門程式 WordCount 和 SQL 實現
  3. 第03講：Flink 的編程模型與其他框架比較
  4. 第04講：Flink 常用的 DataSet 和 DataStream API
  5. 第05講：Flink SQL & Table 編程和案例
  6. 第06講：Flink 集群安裝部署和 HA 配置
  7. 第07講：Flink 常見核心概念分析
  8. 第08講：Flink 窗口、時間和水印
  9. 第09講：Flink 狀態與容錯

本課時主要介紹 Flink 中的時間和水印。

我們在之前的課時中反覆提到過窗口和時間的概念，Flink 框架中支援事件時間、攝入時間和處理時間三種。而當我們在流式計算環境中數據從 Source 產生，再到轉換和輸出，這個過程由於網路和反壓的原因會導致消息亂序。因此，需要有一個機制來解決這個問題，這個特別的機制就是「水印」。

我們在第 05 課時中講解過 Flink 窗口的實現，根據窗口數據劃分的不同，目前 Flink 支援如下 3 種：

  • 滾動窗口，窗口數據有固定的大小，窗口中的數據不會疊加；
  • 滑動窗口，窗口數據有固定的大小，並且有生成間隔；
  • 會話窗口，窗口數據沒有固定的大小，根據用戶傳入的參數進行劃分，窗口數據無疊加。

Flink 中的時間分為三種：

  • 事件時間（Event Time），即事件實際發生的時間；
  • 攝入時間（Ingestion Time），事件進入流處理框架的時間；
  • 處理時間（Processing Time），事件被處理的時間。

下面的圖詳細說明了這三種時間的區別和聯繫：

image (18).png

事件時間（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 來實現的：

image (19).png

可以看到，該方法有兩個入參：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 的。

image (20).png

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

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

image (21).png

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

攝入時間（Ingestion Time）

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

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

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

Ingestion Time 的時間類型生成相關的程式碼在 AutomaticWatermarkContext 中：

image (22).png

image (23).png

我們可以看出，這裡會設置一個 watermark 發送定時器，在 watermarkInterval 時間之後觸發。

處理數據的程式碼在 processAndCollect() 方法中：

image (24).png

水印（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 兩種。

整體的類圖如下：

image (25).png

水印種類

周期性水印

我們在使用 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 來設置水印，將接收到的數據進行轉換，分組並且在一個 5
秒的窗口內獲取該窗口中第二個元素最小的那條數據。

複製程式碼

public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

    //設置為eventtime事件類型 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

    //設置水印生成時間間隔100ms env.getConfig().setAutoWatermarkInterval(100);
    DataStream<String> dataStream = env
            .socketTextStream("127.0.0.1", 9000)
            .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {

                private Long currentTimeStamp = 0L;
                //設置允許亂序時間
                private Long maxOutOfOrderness = 5000L;

                @Override
                public Watermark getCurrentWatermark() {

                    return new Watermark(currentTimeStamp - maxOutOfOrderness);

                }
                @Override

                public long extractTimestamp(String s, long l) {

                    String[] arr = s.split(",");

                    long timeStamp = Long.parseLong(arr[1]);

                    currentTimeStamp = Math.max(timeStamp, currentTimeStamp);

                    System.err.println(s + ",EventTime:" + timeStamp + ",watermark:" + (currentTimeStamp - maxOutOfOrderness));

                    return timeStamp;

                }

            });


    dataStream.map(new MapFunction<String, Tuple2<String, Long>>() {

        @Override
        public Tuple2<String, Long> map(String s) throws Exception {

            String[] split = s.split(",");

            return new Tuple2<String, Long>(split[0], Long.parseLong(split[1]));

        }

    })
            .keyBy(0)
           .window(TumblingEventTimeWindows.of(Time.seconds(5)))

            .minBy(1)

            .print();

    env.execute("WaterMark Test Demo");
}

我們第一次試驗的數據如下：

複製程式碼

flink,1588659181000
flink,1588659182000
flink,1588659183000
flink,1588659184000
flink,1588659185000

可以做一個簡單的判斷，第一條數據的時間戳為 1588659181000，窗口的大小為 5 秒，那麼應該會在 flink,1588659185000 這條數據出現時觸發窗口的計算。

我們用 nc -lk 9000 命令啟動埠，然後輸出上述試驗數據，看到控制台的輸出：

image (26).png

很明顯，可以看到當第五條數據出現後，窗口觸發了計算。

下面再模擬一下數據亂序的情況，假設我們的數據來源如下：

複製程式碼

flink,1588659181000
flink,1588659182000
flink,1588659183000
flink,1588659184000
flink,1588659185000
flink,1588659180000
flink,1588659186000
flink,1588659187000
flink,1588659188000
flink,1588659189000
flink,1588659190000

其中的 flink,1588659180000 為亂序消息，來看看會發生什麼？

image (27).png

可以看到，時間戳為 1588659180000 的這條消息並沒有被處理，因為此時程式碼中的允許亂序時間 private Long maxOutOfOrderness = 0L 即不處理亂序消息。

下面修改 private Long maxOutOfOrderness = 5000L，即代表允許消息的亂序時間為 5 秒，然後把同樣的數據發往 socket 埠。

可以看到，我們把所有數據發送出去僅觸發了一次窗口計算，並且輸出的結果中 watermark 的時間往後順延了 5 秒鐘。所以，maxOutOfOrderness 的設置會影響窗口的計算時間和水印的時間，如下圖所示：

image (28).png

假如我們繼續向 socket 中發送數據：

複製程式碼

flink,1588659191000
flink,1588659192000
flink,1588659193000
flink,1588659194000
flink,1588659195000

可以看到下一次窗口的觸發時間：

image (29).png

在這裡要特別說明，Flink 在用時間 + 窗口 + 水印來解決實際生產中的數據亂序問題，有如下的觸發條件：

  • watermark 時間 >= window_end_time；
  • 在 [window_start_time,window_end_time) 中有數據存在，這個窗口是左閉右開的。

此外，因為 WaterMark 的生成是以對象的形式發送到下游，同樣會消耗記憶體，因此水印的生成時間和頻率都要進行嚴格控制，否則會影響我們的正常作業。

點擊這裡下載本課程源碼

總結

這一課時我們學習了 Flink 的時間類型和水印生成，內容偏多並且水印部分理解起來需要時間，建議你結合源碼再進一步學習。

關注公眾號：大數據技術派，回復資料，領取1024G資料。

Tags:

VirMach 便宜 VPS

QNews

QNews