flink水印生成

72qzrwbm  于 2021-06-24  发布在  Flink
关注(0)|答案(1)|浏览(498)

这是一个来自官方网站2.event time和listen端口的word count示例的修改版本

  1. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  2. //listening to the port
  3. val text = env.socketTextStream("localhost", 9999)
  4. .assignAscendingTimestamps(item => {
  5. val line = item.split(" ")
  6. //simply print timestamp
  7. println(line.apply(1))
  8. line.apply(1).toLong*1000 - 3000
  9. })

执行下面的转换

  1. // the process here
  2. text.map { each_input =>
  3. {
  4. val line = each_input.split(" ")
  5. (line.apply(0),1,line.apply(1))
  6. }}
  7. .process(new SimpleProcessFunc)
  8. .print()

其实逻辑从过程功能上变化不大

  1. val mark = context.timerService().currentWatermark()
  2. val timestamp = context.timestamp()
  3. //print some infomation
  4. println(sdf.format(mark) + "===> watermark ===>" + mark)
  5. println(sdf.format(timestamp) + "===> timestamp in context ===> " + timestamp)
  6. collector.collect(i)

我使用cmd通过socket发送数据,但是从ide控制台看,水印是如何生成的似乎很奇怪,后面似乎没有逻辑

  1. 1585977022
  2. 03/12/292269055 00:47:04===> watermark ===>-9223372036854775808
  3. 04/04/2020 13:10:19===> timestamp in context ===> 1585977019000
  4. 2> (epoch,1,1585977022)
  5. 1585977034
  6. 04/04/2020 13:10:18===> watermark ===>1585977018999
  7. 04/04/2020 13:10:31===> timestamp in context ===> 1585977031000
  8. 3> (montanin,1,1585977034)
  9. 1585977053
  10. 04/04/2020 13:10:30===> watermark ===>1585977030999
  11. 04/04/2020 13:10:50===> timestamp in context ===> 1585977050000
  12. 4> (song,1,1585977053)
rslzwgfq

rslzwgfq1#

以下是水印值背后的逻辑:
初始水印的值为long.min_,即-9223372036854775808。
碰巧的是,水印跟踪在流元素后面,流元素的时间戳被用作创建水印的基础。而水印则是关于某个时间点上的流的完整性的陈述。因此,时间1585977019000的流元素在时间1585977018999的水印之前(因为在水印之后可能还有时间1585977019000的另一个流元素,所以该水印的时间戳为1585977019000是错误的)。
升序时间戳水印生成器是一种周期性水印生成器,默认情况下,它每200毫秒生成一个新的水印,但前提是水印已提前。
在单个输入中访问当前水印时 ProcessFunction ,您将获取该示例接收到的最新水印。在 processElement() 方法,则该水印将不会反映水印生成器在处理当前传递给的事件时所学习的内容 processElement() --水印更新将在200毫秒计时器关闭后进行。
有关水印的更多信息,请参阅flink training中有关水印的页面。

相关问题