当水印小于窗口结束时，触发基于事件时间的Flink窗口操作。

Question

我正在测试事件-时间和水印在Flink.下面是我的密码。

object WatermarkTest {

     def main(args: Array[String]): Unit = {
         val env = StreamExecutionEnvironment.getExecutionEnvironment
         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

         val properties = new Properties()
         properties.setProperty("bootstrap.servers", "127.0.0.1:9092")
         properties.setProperty("group.id", "enven-test")

         env.getConfig.setAutoWatermarkInterval(1L)
         val input = env.addSource(new FlinkKafkaConsumer011[String]("event-time-topic", new SimpleStringSchema(), properties))

         val inputMap = input.map(f=> {
           val arr = f.split(",")
           val code = arr(0)
           val time = arr(1).toLong
           MyEvent(code, time)
        })

        val watermark = inputMap.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator())
        val window = watermark
           .keyBy(_.code)
           .window(TumblingEventTimeWindows.of(Time.seconds(5)))
           .apply(new WindowFunctionTest)

        window.print()

       env.execute()
    }

   class WindowFunctionTest extends WindowFunction[MyEvent,(String, Int,String,String,String,String),String,TimeWindow]{
       override def apply(key: String, window: TimeWindow, input: Iterable[MyEvent], out: Collector[(String, Int,String,String,String,String)]): Unit = {
       val list = input.toList.sortBy(_.time)
       val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
 out.collect(key,input.size,format.format(list.head.time),format.format(list.last.time),format.format(window.getStart),format.format(window.getEnd))
    }
  }
}

下面是事件时间和水印生成器：

class BoundedOutOfOrdernessGenerator extends 
      AssignerWithPeriodicWatermarks[MyEvent] {
      val maxOutOfOrderness = 10000L

      var currentMaxTimestamp: Long = _

      val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")

      var watermark: Watermark = null
      var timestamp: Long = _

      override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = {
           timestamp = element.time
           currentMaxTimestamp = if (timestamp > currentMaxTimestamp) timestamp else currentMaxTimestamp
           println("timestamp:" + element.code +","+ element.time + "|" +format.format(element.time) +", currentMaxTimestamp: "+  currentMaxTimestamp + "|"+ format.format(currentMaxTimestamp) + ", watermark: "+ format.format(watermark.getTimestamp))
           timestamp;
      }

      override def getCurrentWatermark(): Watermark = {
           watermark = new Watermark((currentMaxTimestamp - maxOutOfOrderness)/1000*1000);
          watermark
      }
 }

这是一些测试数据。在我看来，第一次计算后应该水印: 2016-04-27 19:34:25.000.测试结果表明水印后触发的计算值为: 2016-04-27 19:34:24.000。有人能解释一下吗？

​

​

Answer 1

我建议您在getCurrentWatermark和extractTimestamp中打印水印。这应该能说明到底发生了什么。

问题是，extractTimestamp被调用从事件中提取时间戳，时间戳为19:34:35 --该事件将导致当前水印提前到19:34:25，从而触发窗口--此时您正在打印当前水印。在执行extractTimestamp中的println时，水印尚未提前以反映这一新事件。但是在extractTimestamp返回后不久，getCurrentWatermark将被调用，这将将当前水印提升到19:34:25，这将触发窗口。