1. Flink 時間語義

Flink定義了三類時間

• 處理時間（Process Time）數(shù)據(jù)進入Flink被處理的系統(tǒng)時間（Operator處理數(shù)據(jù)的系統(tǒng)時間）
• 事件時間（Event Time）數(shù)據(jù)在數(shù)據(jù)源產生的時間洪己，一般由事件中的時間戳描述，比如用戶日志中的TimeStamp

• 攝取時間（Ingestion Time）數(shù)據(jù)進入Flink的時間呈础，記錄被Source節(jié)點觀察到的系統(tǒng)時間

  
  
  Flink時間概念

在Flink中默認使用的是Process Time技俐，絕大部分的業(yè)務都會使用eventTime堕花，一般只在eventTime無法使用時哮内，才會被迫使用ProcessingTime或者IngestionTime。
如果要使用EventTime泽铛，那么需要引入EventTime的時間屬性尚辑，引入方式如下所

//設置時間語義為Ingestion Time 
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
//設置時間語義為Event Time 我們還需要指定一下數(shù)據(jù)中哪個字段是事件時間（下文會講） 
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

2. WaterMark

2.1 Why WaterMark

我們知道，流處理從事件產生厚宰，到流經source腌巾，再到operator，中間是有一個過程和時間的铲觉，雖然大部分情況下澈蝙，流到operator的數(shù)據(jù)都是按照事件產生的時間順序來的，但是在遇到特殊情況下撵幽，比如遇到網(wǎng)絡延遲或者使用Kafka（多分區(qū)） 很難保證數(shù)據(jù)都是按照事件時間的順序進入Flink灯荧，很有可能是亂序進入。

亂序數(shù)據(jù)

那么此時出現(xiàn)一個問題盐杂，一旦出現(xiàn)亂序逗载，如果只根據(jù)eventTime決定window的運行，我們不能明確數(shù)據(jù)是否全部到位链烈，但又不能無限期的等下去厉斟，此時必須要有個機制來保證一個特定的時間后，必須觸發(fā)window去進行計算了强衡，這個特別的機制擦秽，就是Watermark。

2.2 WaterMark 概念 (what)

• Watermark是一種衡量Event Time進展的機制漩勤。
• Watermark是用于處理亂序事件的感挥，而正確的處理亂序事件，通常用Watermark機制結合window來實現(xiàn)越败。
• 數(shù)據(jù)流中的Watermark用于表示timestamp小于Watermark的數(shù)據(jù)触幼，都已經到達了，因此究飞，window的執(zhí)行也是由Watermark觸發(fā)的置谦。
• Watermark可以理解成一個延遲觸發(fā)機制，我們可以設置Watermark的延時時長t亿傅，每次系統(tǒng)會校驗已經到達的數(shù)據(jù)中最大的maxEventTime霉祸，然后認定eventTime小于maxEventTime - t的所有數(shù)據(jù)都已經到達，如果有窗口的停止時間等于maxEventTime – t袱蜡，那么這個窗口被觸發(fā)執(zhí)行丝蹭。

有序流的Watermarker如下圖所示：（Watermark設置為0）

in order Watermark

Out of order waterMark

Watermark 就是觸發(fā)前一窗口的“關窗時間”，一旦觸發(fā)關門那么以當前時刻為準在窗口范圍內的所有所有數(shù)據(jù)都會收入窗中食拜。
只要沒有達到水位那么不管現(xiàn)實中的時間推進了多久都不會觸發(fā)關窗鸵熟。

注意：如果數(shù)據(jù)不會亂序進入Flink，沒必要使用Watermark ProcessTime 是沒有亂序的

2.3 WaterMark 引入 （how） 怎樣使用WaterMark

DataStream關于WaterMark的方法

根據(jù) DataStream類负甸，WaterMark相關的兩個方法：

2.3.1 assignAscendingTimestamps

一種簡單的特殊情況是旅赢，如果我們事先得知數(shù)據(jù)流的時間戳是單調遞增的，也就是說沒有亂序惑惶，那我們可以使用assignAscendingTimestamps煮盼，這個方法會直接使用數(shù)據(jù)的時間戳生成watermark。

val stream: DataStream[SensorReading] = ...
val withTimestampsAndWatermarks = stream
.assignAscendingTimestamps(e => e.timestamp)

>> result:  E(1), W(1), E(2), W(2), ...

2.3.2 assignTimestampsAndWatermarks

• assignTimestampsAndWatermarks(watermarkStrategy: WatermarkStrategy[T])
• assignTimestampsAndWatermarks(assigner: AssignerWithPeriodicWatermarks[T])
• assignTimestampsAndWatermarks(assigner: AssignerWithPunctuatedWatermarks[T])

AssignerWithPeriodicWatermarks
AssignerWithPunctuatedWatermarks
以上兩個接口都繼承自TimestampAssigner带污。 在1.11已經不建議使用僵控。

2.3.2.1. AssignerWithPeriodicWatermarks

周期性的生成watermark：系統(tǒng)會周期性的將watermark插入到流中(水位線也是一種特殊的事件!)。默認周期是200毫秒鱼冀”ㄆ疲可以使用ExecutionConfig.setAutoWatermarkInterval()方法進行設置。

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

// 每隔5秒產生一個watermark
env.getConfig.setAutoWatermarkInterval(5000)

產生watermark的邏輯：每隔5秒鐘千绪，F(xiàn)link會調用AssignerWithPeriodicWatermarks的getCurrentWatermark()方法充易。如果方法返回一個時間戳大于之前水位的時間戳，新的watermark會被插入到流中荸型。這個檢查保證了水位線是單調遞增的盹靴。如果方法返回的時間戳小于等于之前水位的時間戳，則不會產生新的watermark瑞妇。
例子稿静，自定義一個周期性的時間戳抽取：

class PeriodicAssigner extends AssignerWithPeriodicWatermarks[SensorReading] {
val bound: Long = 60 * 1000 // 延時為1分鐘
var maxTs: Long = Long.MinValue // 觀察到的最大時間戳

override def getCurrentWatermark: Watermark = {
new Watermark(maxTs - bound)
}

override def extractTimestamp(r: SensorReading, previousTS: Long) = {
maxTs = maxTs.max(r.timestamp)
r.timestamp
}
}

而對于亂序數(shù)據(jù)流辕狰，如果我們能大致估算出數(shù)據(jù)流中的事件的最大延遲時間改备，就可以使用如下代碼：

val stream: DataStream[SensorReading] = ...
val withTimestampsAndWatermarks = stream.assignTimestampsAndWatermarks(
new SensorTimeAssigner
)

class SensorTimeAssigner extends BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(5)) {
// 抽取時間戳
override def extractTimestamp(r: SensorReading): Long = r.timestamp
}

>> relust:  E(10), W(0), E(8), E(7), E(11), W(1), ...

2.3.2.2. AssignerWithPunctuatedWatermarks

間斷式地生成watermark。和周期性生成的方式不同蔓倍，這種方式不是固定時間的悬钳，而是可以根據(jù)需要對每條數(shù)據(jù)進行篩選和處理盐捷。直接上代碼來舉個例子，我們只給sensor_1的傳感器的數(shù)據(jù)流插入watermark：

class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[SensorReading] {
  val bound: Long = 60 * 1000

  override def checkAndGetNextWatermark(r: SensorReading, extractedTS: Long): Watermark = {
    if (r.id == "sensor_1") {
    new Watermark(extractedTS - bound)
    } else {
    null
    }
}
  override def extractTimestamp(r: SensorReading, previousTS: Long): Long = {
    r.timestamp
  }
}

2.3.2.3. WatermarkStrategy

新的 WatermarkAssigner 接口將之前的 AssignerWithPunctuatedWatermarks 和 AssignerWithPeriodicWatermarks 的兩類 Watermark 的接口進行了整合默勾，從而簡化了后續(xù)開發(fā)支持插入 Watermark 的 Source 實現(xiàn)復雜度碉渡。

WatermarkStrategy 實現(xiàn)

• 1. new WatermarkStrategy 直接實現(xiàn)WatermarkStrategy 接口

public interface WatermarkStrategy<T> extends TimestampAssignerSupplier<T>, WatermarkGeneratorSupplier<T>{

    /**
     * Instantiates a {@link TimestampAssigner} for assigning timestamps according to this
     * strategy.
     */
    @Override
    TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context);

    /**
     * Instantiates a WatermarkGenerator that generates watermarks according to this strategy.
     */
    @Override
    WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}

需要實現(xiàn)createWatermarkGenerator 方法創(chuàng)建watermark
以及實現(xiàn)createTimestampAssigner方法將數(shù)據(jù)指定時間戳

通常不會自己實現(xiàn)此接口，而是調用WatermarkStrategy中的靜態(tài)方法

• 2. WatermarkStrategy 靜態(tài)方法

WatermarkStrategy
  .forBoundedOutOfOrderness[(Long, String)](Duration.ofSeconds(20))
  .withTimestampAssigner(new SerializableTimestampAssigner[(Long, String)] {
    override def extractTimestamp(element: (Long, String), recordTimestamp: Long): Long = element._1
  })

TimestampAssigner 和WatermarkGenerator 作用和介紹

TimestampAssigner 負責從事件中提取時間戳字段
WatermarkGenerator 負責生成watermark

/**
 * The {@code WatermarkGenerator} generates watermarks either based on events or
 * periodically (in a fixed interval).
 *
 * <p><b>Note:</b> This WatermarkGenerator subsumes the previous distinction between the
 * {@code AssignerWithPunctuatedWatermarks} and the {@code AssignerWithPeriodicWatermarks}.
 */
@Public
public interface WatermarkGenerator<T> {

    /**
     * Called for every event, allows the watermark generator to examine and remember the
     * event timestamps, or to emit a watermark based on the event itself.
     */
    void onEvent(T event, long eventTimestamp, WatermarkOutput output);

    /**
     * Called periodically, and might emit a new watermark, or not.
     *
     * <p>The interval in which this method is called and Watermarks are generated
     * depends on {@link ExecutionConfig#getAutoWatermarkInterval()}.
     */
    void onPeriodicEmit(WatermarkOutput output);
}

onEvent(): 每來一條數(shù)據(jù)灾测，將這條數(shù)據(jù)與maxTimesStamp比較，看是否需要更新watermark
onPeriodicEmit：周期性更新watermark 間隔時間看setAutoWatermarkInterval的參數(shù)

WatermarkStrategy 通過這兩個方法來整合了periodic and punctuated兩種watermark的生成垦巴。

周期性水酉碧隆（Periodic Watermark）根據(jù)事件或者處理時間周期性的觸發(fā)水印生成器(Assigner)，骤宣。 通過onEvent()來觀察傳入事件秦爆，然后通過調用onPeriodicEmit周期性的生成watermark
間歇性水印（Punctuated Watermark）在觀察到事件后憔披，會依據(jù)用戶指定的條件來決定是否發(fā)射
水印等限。通過onEvent()查看事件并等待特殊的標記事件或根據(jù)條件標點，這些事件或標點在流中攜帶水印信息芬膝。會立即emit水印望门。通常，間歇性水印生成器不會調用onPeriodicEmit()發(fā)出水印锰霜。

WatermarkGenerator 周期性生成水印

• 周期性的生成器監(jiān)測流事件并周期性生成水映镂蟆（可能周期性事件可能根據(jù)流數(shù)據(jù)的元素，或基于處理時間）癣缅。

• 通過ExecutionConfig.setAutoWatermarkInterval(...)定義生成水印的間隔時間（每n毫秒）厨剪。每到這個間隔時間，生成器的onPeriodicEmit()方法每次都會被調用友存。如果返回的watermark非空且大于前一個watermark祷膳，則將發(fā)出新的watermark。

下面使用兩個周期性水印生成的WatermarkGenerator 的簡單示例屡立。請注意直晨，F(xiàn)link附帶了 BoundedOutOfOrdernessWatermarks，它的WatermarkGenerator工作原理與以下BoundedOutOfOrdernessGenerator所示類似膨俐。

/**
 * This generator generates watermarks assuming that elements arrive out of order,
 * but only to a certain degree. The latest elements for a certain timestamp t will arrive
 * at most n milliseconds after the earliest elements for timestamp t.
 */
class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {

    val maxOutOfOrderness = 3500L // 3.5 seconds

    var currentMaxTimestamp: Long = _

    override def onEvent(element: MyEvent, eventTimestamp: Long): Unit = {
        currentMaxTimestamp = max(eventTimestamp, currentMaxTimestamp)
    }

    override def onPeriodicEmit(): Unit = {
        // emit the watermark as current highest timestamp minus the out-of-orderness bound
        output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness - 1));
    }
}

/**
 * This generator generates watermarks that are lagging behind processing time by a fixed amount.
 * It assumes that elements arrive in Flink after a bounded delay.
 */
class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {

    val maxTimeLag = 5000L // 5 seconds

    override def onEvent(element: MyEvent, eventTimestamp: Long): Unit = {
        // don't need to do anything because we work on processing time
    }

    override def onPeriodicEmit(): Unit = {
        output.emitWatermark(new Watermark(System.currentTimeMillis() - maxTimeLag));
    }
}

WatermarkGenerator 間歇性生成水印

WatermarkGenerator 將監(jiān)測事件流抡秆，看到符合條件的事件元素就會emit watermark
這樣，可以實現(xiàn)一個間歇性的WatermarkGenerator 吟策，該Generator 在事件表明它帶有特定標記時會發(fā)出水尤迨俊：

class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[MyEvent] {

    override def onEvent(element: MyEvent, eventTimestamp: Long): Unit = {
        if (event.hasWatermarkMarker()) {
            output.emitWatermark(new Watermark(event.getWatermarkTimestamp()))
        }
    }

    override def onPeriodicEmit(): Unit = {
        // don't need to do anything because we emit in reaction to events above
    }
}

注意：可以在每個事件上生成水印。但是檩坚，由于每個水印都會在下游引起一些計算着撩，因此過多的水印會降低性能诅福。

參考 Flink官網(wǎng) Timely Stream Processing