当制作人（kafka）不产生新行时，如何让Flink刷新最后一行

Question

当我的Flink程序处于事件时间模式时，接收器将不会获得最后一行（比如A行）。如果我将新线（线B）馈送到Flink，我将获得A线，但我仍然无法得到线b。

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

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

    val consumer = new FlinkKafkaConsumer[String]("topic", new SimpleStringSchema(), properties)

    val stream: DataStream[String] = env.addSource(consumer).setParallelism(1)

    stream.map { m =>
      val result = JSON.parseFull(m).asInstanceOf[Some[Map[String, Any]]].get
      val msg = result("message").asInstanceOf[String]
      val num = parseMessage(msg)
      val key = s"${num.zoneId} ${num.subZoneId}"
      (key, num, num.onlineNum)
    }.filter { data =>
      data._2.subZoneId == 301 && data._2.zoneId == 5002
    }.assignTimestampsAndWatermarks(new MyTimestampExtractor()).keyBy(0)
      .window(TumblingEventTimeWindows.of(Time.seconds(1)))
        .allowedLateness(Time.minutes(1))
      .maxBy(2).addSink { v =>
      System.out.println(s"${v._2.time} ${v._1}: ${v._2.onlineNum} ")
    }

class MyTimestampExtractor() extends AscendingTimestampExtractor[(String, OnlineNum, Int)](){
  val byMinute = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:SS")
  override def extractAscendingTimestamp(element: (String, OnlineNum, Int)): Long = {
    val dateTimeString =  element._2.date + " " + element._2.time
    val c1 = byMinute.parse(dateTimeString).getTime
    if ( element._2.time.contains("22:59") && element._2.subZoneId == 301){
      //System.out.println(s"${element._2.time} ${element._1}: ${element._2.onlineNum} ")
      // System.out.println(s"${element._2.time} ${c1 - getCurrentWatermark.getTimestamp}")
    }

    // System.out.println(s"${element._2.time} ${c1} ${c1 - getCurrentWatermark.getTimestamp}")
    return c1
  }
}

数据样本：

01:01:14 5002 301: 29 
01:01:36 5002 301: 27 
01:02:05 5002 301: 27 
01:02:31 5002 301: 29 
01:03:02 5002 301: 29 
01:03:50 5002 301: 29 
01:04:52 5002 301: 29 
01:07:24 5002 301: 26 
01:09:28 5002 301: 21 
01:11:04 5002 301: 22 
01:12:11 5002 301: 24 
01:13:54 5002 301: 23 
01:15:13 5002 301: 22 
01:16:04 5002 301: 19 (I can not get this line )

然后我推新线到Flink（通过kafka）

01:17:28 5002 301: 15 

我会得到01:16:04 5002 301: 19，但01:17:28 5002 301: 15可能会在Flink举行。

Answer 1

发生这种情况是因为它是事件时间，事件的时间戳用于测量窗口的时间流。

在这种情况下，当窗口中只有一个事件时，Flink不知道应该省略窗口。因此，当您添加下一个事件时，将关闭上一个窗口并发出元素（在您的情况下为19），然后再创建下一个窗口（在您的情况下为15）。

在这种情况下，最好的想法可能是添加自定义ProcessingTimeTrigger，它基本上允许你在飞行一段时间之后发射窗口，无论事件是否流动。你可以在documentation找到有关Trigger的信息。

Answer 2

请问最终解决方案是什么？我也遇到了类似的情况，可以通过使用新的Watermark（System.CurrtTimeMillis（））来解决，但它似乎不符合Watermark的目的。这不是一个常见问题，还是应用程序开发人员故意忽略它而社区忽视它？

Why not on-time when I consumed kafka message using flink streaming sql group by TUMBLE(rowtime)?