前言

本文隸屬于專欄《1000個問題搞定大資料技術體系》，該專欄為筆者原創，參考請注明來源，不足和錯誤之處請在評論區幫忙指出，謝謝！

本專欄目錄結構和參考文獻請見1000個問題搞定大資料技術體系

正文

1、watermark的作用

watermark是用于處理亂序事件的，而正確的處理亂序事件，通常用watermark機制結合window來實作，
我們知道，流處理從事件產生，到流經source，再到operator，中間是有一個程序和時間的，雖然大部分情況下，流到operator的資料都是按照事件產生的時間順序來的，但是也不排除由于網路、背壓等原因，導致亂序的產生（out-of-order或者說late element），
但是對于late element，我們又不能無限期的等下去，必須要有個機制來保證一個特定的時間后，必須觸發window去進行計算了，這個特別的機制，就是watermark，

2、watermark解決遲到的資料

實時系統中，由于各種原因造成的延時，造成某些訊息發到flink的時間延時于事件產生的時間，如果基于event time構建window，但是對于late element，我們又不能無限期的等下去，必須要有個機制來保證一個特定的時間后，必須觸發window去進行計算了，這個特別的機制，就是watermark，

Watermarks(水位線)就是來處理這種問題的機制

參考google的DataFlow設計，
是event time處理進度的標志，
表示比watermark更早(更老)的事件都已經到達(沒有比水位線更低的資料 )，
基于watermark來進行視窗觸發計算的判斷，

有序的資料流watermark：

在某些情況下，基于Event Time的資料流是有續的(相對event time)，在有序流中，watermark就是一個簡單的周期性標記，
在這里插入圖片描述

無序的資料流watermark：

在更多場景下，基于Event Time的資料流是無續的(相對event time)，
在無序流中，watermark至關重要，她告訴operator比watermark更早(更老/時間戳更小)的事件已經到達， operator可以將內部事件時間提前到watermark的時間戳(可以觸發window計算啦)
在這里插入圖片描述

并行流當中的watermark：

通常情況下， watermark在source函式中生成，但是也可以在source后任何階段，如果指定多次 watermark，后面指定的 watermarker會覆寫前面的值， source的每個sub task獨立生成水印，
watermark通過operator時會推進operators處的當前event time，同時operators會為下游生成一個新的watermark，
多輸入operator(union、 keyBy、 partition)的當前event time是其輸入流event time的最小值，
注意：多并行度的情況下，watermark對齊會取所有channel最小的watermark

在這里插入圖片描述

3、watermark如何生成

通常，在接收到source的資料后，應該立刻生成watermark；但是，也可以在source后，應用簡單的map或者filter操作，然后再生成watermark，

生成watermark的方式主要有2大類：

With Periodic Watermarks
With Punctuated Watermarks

第一種可以定義一個最大允許亂序的時間，這種情況應用較多，
我們主要來圍繞Periodic Watermarks來說明，下面是生成periodic watermark的方法：

4、watermark處理順序資料

需求：定義一個視窗為10s，通過資料的event time時間結合watermark實作延遲10s的資料也能夠正確統計
我們通過資料的eventTime來向前推10s，得到資料的watermark，
代碼實作：

package com.shockang.study.bigdata.flink.watermark

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.text.SimpleDateFormat
import scala.collection.mutable.ArrayBuffer
import scala.util.Sorting

object FlinkWaterMark2 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.api.scala._
    //設定flink的資料處理時間為eventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)
    val tupleStream: DataStream[(String, Long)] = env.socketTextStream("node01", 9000).map(x => {
      val strings: Array[String] = x.split(" ")
      (strings(0), strings(1).toLong)
    })

    //注冊我們的水印
    val waterMarkStream: DataStream[(String, Long)] = tupleStream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[(String, Long)] {
      var currentTimemillis: Long = 0L
      var timeDiff: Long = 10000L
      val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

      /* //獲取當前資料的waterMark
       override def getNext: Watermark = {
       }*/
      override def getCurrentWatermark: Watermark = {
        val watermark = new Watermark(currentTimemillis - timeDiff)
        watermark
      }

      //抽取資料的eventTime
      override def extractTimestamp(element: (String, Long), l: Long): Long = {
        val enventTime = element._2
        currentTimemillis = Math.max(enventTime, currentTimemillis)
        val id = Thread.currentThread().getId
        println("currentThreadId:" + id + ",key:" + element._1 + ",eventtime:[" + element._2 + "|" + sdf.format(element._2) + "],currentMaxTimestamp:[" + currentTimemillis + "|" + sdf.format(currentTimemillis) + "],watermark:[" + this.getCurrentWatermark.getTimestamp + "|" + sdf.format(this.getCurrentWatermark.getTimestamp) + "]")
        enventTime
      }
    })
    waterMarkStream.keyBy(0)
      .window(TumblingEventTimeWindows.of(Time.seconds(10)))
      .apply(new MyWindowFunction2).print()
    env.execute()
  }
}


class MyWindowFunction2 extends WindowFunction[(String, Long), String, Tuple, TimeWindow] {
  override def apply(key: Tuple, window: TimeWindow, input: Iterable[(String, Long)],
                     out: Collector[String]): Unit = {
    val keyStr = key.toString
    val arrBuf = ArrayBuffer[Long]()
    val ite = input.iterator
    while (ite.hasNext) {
      val tup2 = ite.next()
      arrBuf.append(tup2._2)
    }
    val arr = arrBuf.toArray
    Sorting.quickSort(arr) //對資料進行排序，按照eventTime進行排序
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    val result = "聚合資料的key為：" + keyStr + "," + "視窗當中資料的條數為：" + arr.length + "," + "視窗當中第一條資料為：" + sdf.format(arr.head) + "," + "視窗當中最后一條資料為：" + sdf.format(arr.last) + "," + "視窗起始時間為：" + sdf.format(window.getStart) + "," + "視窗結束時間為：" + sdf.format(window.getEnd) + "！！！！！看到這個結果，就證明視窗已經運行了"
    out.collect(result)
  }
}

輸入測驗資料

注意：如果需要觸發flink的視窗呼叫，必須滿足兩個條件
1：waterMarkTime > eventTime
2：視窗內有資料

資料輸入測驗：

按照十秒鐘統計一次，我們程式會將時間劃分成為以下時間間隔段
2019-10-01 10:11:00  到  2019-10-01 10:11:10
2019-10-01 10:11:10  到  2019-10-01 10:11:20
2019-10-01 10:11:20  到  2019-10-01 10:11:30
2019-10-01 10:11:30  到  2019-10-01 10:11:40
2019-10-01 10:11:40  到  2019-10-01 10:11:50
2019-10-01 10:11:50  到  2019-10-01 10:12:00
	
順序計算：
觸發資料計算的條件依據為兩個
第一個waterMark時間大于資料的eventTime時間，第二個視窗之內有資料
我們這里的waterMark直接使用eventTime的最大值減去10秒鐘

0001 1569895882000	 資料eventTime為：2019-10-01 10:11:22  資料waterMark為  2019-10-01 10:11:12
0001 1569895885000	 資料eventTime為：2019-10-01 10:11:25  資料waterMark為  2019-10-01 10:11:15
0001 1569895888000	 資料eventTime為：2019-10-01 10:11:28  資料waterMark為  2019-10-01 10:11:18

0001 1569895890000	 資料eventTime為：2019-10-01 10:11:30  資料waterMark為  2019-10-01 10:11:20
0001 1569895891000	 資料eventTime為：2019-10-01 10:11:31  資料waterMark為  2019-10-01 10:11:21
0001 1569895895000	 資料eventTime為：2019-10-01 10:11:35  資料waterMark為  2019-10-01 10:11:25
0001 1569895898000	 數據eventTime為：2019-10-01 10:11:38  資料waterMark為  2019-10-01 10:11:28

0001 1569895900000	 資料eventTime為：2019-10-01 10:11:40  資料waterMark為  2019-10-01 10:11:30  觸發第一條到第三條資料計算，資料包前不包后，不會計算2019-10-01 10:11:30 這條資料
0001 1569895911000	 資料eventTime為：2019-10-01 10:11:51  資料waterMark為  2019-10-01 10:11:41  觸發2019-10-01 10:11:20到2019-10-01 10:11:28時間段的額資料計算，資料包前不包后，不會觸發2019-10-01 10:11:30這條資料的計算

5、watermark處理亂序資料

輸入測驗資料
接著繼續輸入以下亂序資料，驗證flink亂序資料的問題是否能夠解決

亂序資料
0001 1569895948000	 資料eventTime為：2019-10-01 10:12:28  資料waterMark為  2019-10-01 10:12:18  
0001 1569895945000	 資料eventTime為：2019-10-01 10:12:25  資料waterMark為  2019-10-01 10:12:18  
0001 1569895947000	 資料eventTime為：2019-10-01 10:12:27  資料waterMark為  2019-10-01 10:12:18  

0001 1569895950000	 資料eventTime為：2019-10-01 10:12:30  資料waterMark為  2019-10-01 10:12:20  

0001 1569895960000	 資料eventTime為：2019-10-01 10:12:40  資料waterMark為  2019-10-01 10:12:30  觸發計算 waterMark > eventTime 并且視窗內有資料，觸發 2019-10-01 10:12:28到2019-10-01 10:12:27 這三條資料的計算，資料包前不包后，不會觸發2019-10-01 10:12:30 這條資料的計算
0001 1569895949000	 資料eventTime為：2019-10-01 10:12:29  資料waterMark為  2019-10-01 10:12:30  遲到太多的資料，flink直接丟棄，可以設定flink將這些遲到太多的資料保存起來，便于排查問題

6、比watermark更晚的資料如何解決

如果我們設定資料的watermark為每條資料的eventtime往后一定的時間，例如資料的eventtime為2019-08-20 15:30:30，程式的window視窗為10s，然后我們設定的watermark為2019-08-20 15:30:40，
那么如果某一條資料eventtime為2019-08-20 15:30:32，到達flink程式的時間為2019-08-20 15:30:45 該怎么辦，這條資料比視窗的watermark時間還要晚了5S鐘該怎么辦？對于這種比watermark還要晚的資料，flink有三種處理方式

1、直接丟棄

我們輸入一個亂序很多的（其實只要 Event Time < watermark 時間）資料來測驗下：輸入：【輸入兩條內容】
late element
0001 1569895948000 資料eventTime為：2019-10-01 10:12:28 資料直接丟棄
0001 1569895945000 資料eventTime為：2019-10-01 10:12:25 資料直接丟棄
注意：此時并沒有觸發 window，因為輸入的資料所在的視窗已經執行過了，flink 默認對這些遲到的資料的處理方案就是丟棄，

2、allowedLateness 指定允許資料延遲的時間

在某些情況下，我們希望對遲到的資料再提供一個寬容的時間，
Flink 提供了 allowedLateness 方法可以實作對遲到的資料設定一個延遲時間，在指定延遲時間內到達的資料還是可以觸發 window 執行的，

修改代碼：

waterMarkStream
  .keyBy(0)
  .window(TumblingEventTimeWindows.of(Time.seconds(3))).allowedLateness(Time.seconds(2))//允許資料遲到2S
//function: (K, W, Iterable[T], Collector[R]) => Unit
.apply(new MyWindowFunction).print()

驗證資料遲到性：
輸入資料：

更改代碼之后重啟我們的程式，然后從新輸入之前的資料

0001 1569895882000
0001 1569895885000
0001 1569895888000
0001 1569895890000
0001 1569895891000
0001 1569895895000
0001 1569895898000
0001 1569895900000
0001 1569895911000
0001 1569895948000
0001 1569895945000
0001 1569895947000
0001 1569895950000
0001 1569895960000
0001 1569895949000

驗證資料的延遲性：定義資料僅僅延遲2S的資料重新接收，重新計算

0001 1569895948000	 資料eventTime為：2019-10-01 10:12:28  觸發資料計算  資料waterMark為  2019-10-01 10:12:30
0001 1569895945000	 資料eventTime為：2019-10-01 10:12:25  觸發資料計算  資料waterMark為  2019-10-01 10:12:30

0001 1569895958000	 資料eventTime為：2019-10-01 10:12:38  不會觸發資料計算 資料waterMark為  2019-10-01 10:12:30  waterMarkTime  <  eventTime，所以不會觸發計算

將資料的waterMark調整為41秒就可以觸發上面這條資料的計算了

0001 1569895971000	 資料eventTime為：2019-10-01 10:12:51  資料waterMark為  2019-10-01 10:12:41

又會繼續觸發0001 1569895958000 這條資料的計算了

3、sideOutputLateData 收集遲到的資料

通過 sideOutputLateData 可以把遲到的資料統一收集，統一存盤，方便后期排查問題，

需要先調整代碼：

package com.shockang.study.bigdata.flink.watermark

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.text.SimpleDateFormat
import scala.collection.mutable.ArrayBuffer
import scala.util.Sorting


object FlinkWaterMark {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.api.scala._
    //設定time型別為eventtime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //暫時定義并行度為1
    env.setParallelism(1)
    val text = env.socketTextStream("node01", 9000)
    val inputMap: DataStream[(String, Long)] = text.map(line => {
      val arr = line.split(" ")
      (arr(0), arr(1).toLong)
    })

    //給我們的資料注冊waterMark
    val waterMarkStream: DataStream[(String, Long)] = inputMap
      .assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks[(String, Long)] {
        var currentMaxTimestamp = 0L

        //watermark基于eventTime向后推遲10秒鐘，允許訊息最大亂序時間為10s
        val waterMarkDiff: Long = 10000L

        val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

        //獲取下一個水印
        override def checkAndGetNextWatermark(t: (String, Long), l: Long): Watermark = {
          val watermark = new Watermark(currentMaxTimestamp - waterMarkDiff)
          watermark
        }

        //抽取當前資料的時間作為eventTime
        override def extractTimestamp(element: (String, Long), l: Long): Long = {
          val eventTime = element._2
          currentMaxTimestamp = Math.max(eventTime, currentMaxTimestamp)
          val id = Thread.currentThread().getId
          println("currentThreadId:" + id + ",key:" + element._1 + ",eventtime:[" + element._2 + "|" + sdf.format(element._2) + "],currentMaxTimestamp:[" + currentMaxTimestamp + "|" + sdf.format(currentMaxTimestamp) + "],watermark:[" + this.checkAndGetNextWatermark(element, l).getTimestamp + "|" + sdf.format(this.checkAndGetNextWatermark(element, l).getTimestamp) + "]")
          eventTime
        }
      })


    val outputTag: OutputTag[(String, Long)] = new OutputTag[(String, Long)]("late_data")
    val outputWindow: DataStream[String] = waterMarkStream
      .keyBy(0)
      .window(TumblingEventTimeWindows.of(Time.seconds(3)))
      // .allowedLateness(Time.seconds(2))//允許資料遲到2S
      .sideOutputLateData(outputTag)
      //function: (K, W, Iterable[T], Collector[R]) => Unit
      .apply(new MyWindowFunction)


    val sideOuptut: DataStream[(String, Long)] = outputWindow.getSideOutput(outputTag)

    sideOuptut.print()
    outputWindow.print()

    //執行程式
    env.execute()

  }
}

class MyWindowFunction extends WindowFunction[(String, Long), String, Tuple, TimeWindow] {
  override def apply(key: Tuple, window: TimeWindow, input: Iterable[(String, Long)], out: Collector[String]): Unit = {
    val keyStr = key.toString
    val arrBuf = ArrayBuffer[Long]()
    val ite = input.iterator
    while (ite.hasNext) {
      val tup2 = ite.next()
      arrBuf.append(tup2._2)
    }
    val arr = arrBuf.toArray
    Sorting.quickSort(arr)
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    val result = keyStr + "," + arr.length + "," + sdf.format(arr.head) + "," + sdf.format(arr.last) + "," + sdf.format(window.getStart) + "," + sdf.format(window.getEnd)
    out.collect(result)
  }
}

我們來輸入一些資料驗證一下輸入

0001 1569895882000
0001 1569895885000
0001 1569895888000
0001 1569895890000
0001 1569895891000
0001 1569895895000
0001 1569895898000
0001 1569895900000
0001 1569895911000
0001 1569895948000
0001 1569895945000
0001 1569895947000
0001 1569895950000
0001 1569895960000
0001 1569895949000
輸入兩條遲到的資料，會被收集起來
0001 1569895948000
0001 1569895945000

此時，針對這幾條遲到的資料，都通過 sideOutputLateData 保存到了 outputTag 中，

7、多并行度的watermark機制

前面代碼中設定了并行度為 1

env.setParallelism(1);

如果這里不設定的話，代碼在運行的時候會默認讀取本機 CPU 數量設定并行度，把代碼的并行度代碼注釋掉

//env.setParallelism(1)

然后在輸出內容前面加上執行緒 id

在這里插入圖片描述
會出現如下資料：輸入如下幾行內容：

輸出：

會發現 window 沒有被觸發，

因為此時，這 7 條資料都是被不同的執行緒處理的，每個執行緒都有一個 watermark，

因為在多并行度的情況下，watermark 對齊會取所有 channel 最小的 watermark 但是我們現在默認有 8 個并行度，這 7 條資料都被不同的執行緒所處理，到現在還沒獲取到最小的 watermark，所以 window 無法被觸發執行，
在這里插入圖片描述
下面我們來驗證一下，把代碼中的并行度調整為 2.

env.setParallelism(2)

輸入如下內容：

0001 1569895890000
0001 1569895903000
0001 1569895908000

輸出：
在這里插入圖片描述
此時會發現，當第三條資料輸入完以后，[10:11:30,10:11:33)這個 window 被觸發了，

前兩條資料輸入之后，獲取到的最小 watermark 是 10:11:20，這個時候對應的 window 中沒有資料，

第三條資料輸入之后，獲取到的最小 watermark 是 10:11:33，這個時候對應的視窗就是 [10:11:30,10:11:33)，所以就觸發了，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/294163.html

標籤：其他

上一篇：企業運維實戰--ELK日志分析平臺之elasticsearch實戰

下一篇：Flink 的 Time 三兄弟

一篇文章搞懂 Flink 的 watermark 機制

前言

正文