Flink assignAscendingTimestamps 生成水印的三個重載方法

先簡單介紹一下Timestamp 和Watermark 的概念: 

1. Timestamp和Watermark都是基於事件的時間字段生成的
2. Timestamp和Watermark是兩個不一樣的東西，而且一旦生成都跟事件數據沒有關係了（全部即便事件中再也不包含生成Timestamp和Watermark的字段也不要緊）
3. 事件數據和 Timestamp 一一對應（事件在流中傳遞以StreamRecord對象表示，value 和 timestamp 是它的兩個成員變量)
4. Watermark 在生成以後與事件數據沒有直接關係，Watermark 做爲一個消息，和事件數據同樣在流中傳遞（Watermark 和StreamRecord 具備相同的父類：StreamElement）
5. Timestamp 與 Watermark 在生成以後，會在下游window算子中作比較，判斷事件數據是不是過時數據
6. 只有window算子纔會用Watermark判斷事件數據是否過時

Flink 在流上手動生成水印有三個重載的方法（忽略過時的一個）

 

1. assignTimestamps(extractor: TimestampExtractor[T]): DataStream[T]

此方法是數據流的快捷方式，其中已知元素時間戳在每一個並行流中單調遞增。在這種狀況下，系統能夠經過跟蹤上升時間戳自動且完美地生成水印。

val input = env.addSource(source)
.map(json => {
        val id = json.get("id").asText()
        val createTime = json.get("createTime").asText()
        val amt = json.get("amt").asText()
        LateDataEvent("key", id, createTime, amt)
      })
      // flink auto create timestamp & watermark
      .assignAscendingTimestamps(element => sdf.parse(element.createTime).getTime)

注：這種方法建立時間戳與水印最簡單，返回一個long類型的數字就能夠了

 

2.assignTimestampsAndWatermarks(assigner: AssignerWithPeriodicWatermarks[T]): DataStream[T] 

基於給定的水印生成器生成水印，即便沒有新元素到達也會按期檢查給定水印生成器的新水印，以指定容許延遲時間

val input = env.addSource(source)
      .map(json => {
        val id = json.get("id").asText()
        val createTime = json.get("createTime").asText()
        val amt = json.get("amt").asText()
        LateDataEvent("key", id, createTime, amt)
      })
      // assign timestamp & watermarks periodically(按期生成水印)
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LateDataEvent](Time.milliseconds(50)) {
       override def extractTimestamp(element: LateDataEvent): Long = {
         println("want watermark : " + sdf.parse(element.createTime).getTime)
         sdf.parse(element.createTime).getTime
       }
     })
      

3.assignTimestampsAndWatermarks(assigner: AssignerWithPeriodicWatermarks[T]): DataStream[T]

此方法僅基於流元素建立水印，對於經過[[AssignerWithPunctuatedWatermarks＃extractTimestamp（Object，long）]]處理的每一個元素，
調用[[AssignerWithPunctuatedWatermarks＃checkAndGetNextWatermark（）]]方法，若是返回的水印值大於之前的水印，會發出新的水印，
此方法能夠徹底控制水印的生成，可是要注意，每秒生成數百個水印會影響性能

val input = env.addSource(source)
      .map(json => {
        val id = json.get("id").asText()
        val createTime = json.get("createTime").asText()
        val amt = json.get("amt").asText()
        LateDataEvent("key", id, createTime, amt)
      })
      // assign timestamp & watermarks every event
      .assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks[LateDataEvent]() {
      // check extractTimestamp emitted watermark is non-null and large than previously
      override def checkAndGetNextWatermark(lastElement: LateDataEvent, extractedTimestamp: Long): Watermark = {
        new Watermark(extractedTimestamp)
      }
      // generate next watermark
      override def extractTimestamp(element: LateDataEvent, previousElementTimestamp: Long): Long = {
        val eventTime = sdf.parse(element.createTime).getTime
        eventTime
      }
    })

注：本文基於所有事件時間

 歡迎關注Flink菜鳥公衆號，會不按期更新Flink（開發技術）相關的推文