Flink，水印的几种简单写法

小鱼儿 2024-04-17 05:53 63阅读 0赞

>     package com.coder.flink.restart
>     
>     
>     import org.apache.flink.api.scala._
>     import org.apache.flink.streaming.api.functions.source.SourceFunction
>     import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
>     import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
>     import org.apache.flink.streaming.api.watermark.Watermark
>     import org.apache.flink.streaming.api.windowing.time.Time
>     
>     object WaterMarkDemo {
>       def main(args: Array[String]): Unit = {
>         val env = StreamExecutionEnvironment.getExecutionEnvironment
>         val input = List(("a", 2L, 1), ("b", 2L, 1), ("c", 3L, 1))
>         env.addSource(new SourceFunction[(String, Long, Int)] {
>           override def run(ctx: SourceFunction.SourceContext[(String, Long, Int)]): Unit = {
>             input.foreach(value => {
>               //固定套路 调用 collectWithTimestamp 指定哪个字段是水印字段
>               ctx.collectWithTimestamp(value, value._2)
>               ctx.emitWatermark(new Watermark(value._2 - 1)) //设置最大延迟为1
>             })
>             //设置默认水印
>             ctx.emitWatermark(new Watermark(Long.MaxValue))
>           }
>     
>           override def cancel(): Unit = {}
>         })
>     
>         val data: DataStream[(String, Long, Int)] = env.fromCollection(List(("a", 2L, 1), ("b", 2L, 1), ("c", 3L, 1)))
>         data.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.seconds(10)) {
>     
>           //定义抽取EventTIme 时间戳逻辑
>           override def extractTimestamp(element: (String, Long, Int)): Long = {
>             element._2
>           }
>         })
>     
>         //使用系统默认的 升序 分配时间信息和水印，我们只需指定一个字段就可以了
>         val watermark: DataStream[(String, Long, Int)] = data.assignAscendingTimestamps(t => t._2)
>        val  rs =  watermark.keyBy(1).timeWindow(Time.seconds(1L)).sum(2)
>         rs.print()
>     //    data.print()
>         env.execute()
>     
>       }
>     }