Flink assignAscendingTimestamps 生成水印的三个重载方法

2019-08-24 10:02:55 阅读：681 来源： 互联网

标签：val Watermark Flink 水印 assignAscendingTimestamps json asText createTime

先简单介绍一下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 在流上手动生成水印有三个重载的方法（忽略过期的一个）

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（开发技术）相关的推文

标签：val,Watermark,Flink,水印,assignAscendingTimestamps,json,asText,createTime
来源： https://www.cnblogs.com/Springmoon-venn/p/11403665.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

Flink assignAscendingTimestamps 生成水印的三个重载方法